utils.py

# File 4: utils.py
with open(f"{base_dir}/utils.py", "w") as f:
    f.write('''"""Shared utilities for Swarm CLI"""
import json
import os
import hashlib
import datetime
from typing import Any, Dict, List, Optional
from rich.console import Console
from rich.panel import Panel
from rich.syntax import Syntax
from rich.table import Table
from rich.tree import Tree

console = Console()

class Logger:
    """Activity logger"""
    
    @staticmethod
    def log(action: str, details: Dict[str, Any], agent: str = "system"):
        """Log an action"""
        entry = {
            "timestamp": datetime.datetime.now().isoformat(),
            "agent": agent,
            "action": action,
            "details": details
        }
        
        # Print to console
        console.print(f"[dim][{entry['timestamp']}] {agent}: {action}[/dim]")
        
        # Append to log file
        try:
            with open("swarm.log", "a", encoding="utf-8") as f:
                f.write(json.dumps(entry) + "\n")
        except:
            pass
        
        return entry

class Memory:
    """Simple JSON-based memory store"""
    
    def __init__(self, filepath: str = "memory.json"):
        self.filepath = filepath
        self.data = self._load()
    
    def _load(self) -> Dict:
        if os.path.exists(self.filepath):
            try:
                with open(self.filepath, "r", encoding="utf-8") as f:
                    return json.load(f)
            except:
                return {"facts": [], "conversations": [], "tools_used": []}
        return {"facts": [], "conversations": [], "tools_used": []}
    
    def save(self):
        with open(self.filepath, "w", encoding="utf-8") as f:
            json.dump(self.data, f, indent=2, ensure_ascii=False)
    
    def add_fact(self, key: str, value: Any):
        self.data["facts"].append({
            "key": key,
            "value": value,
            "timestamp": datetime.datetime.now().isoformat()
        })
        self.save()
    
    def get_facts(self, query: str = "") -> List[Dict]:
        if not query:
            return self.data["facts"][-20:]
        return [f for f in self.data["facts"] if query.lower() in f["key"].lower()]
    
    def add_conversation(self, role: str, content: str):
        self.data["conversations"].append({
            "role": role,
            "content": content,
            "timestamp": datetime.datetime.now().isoformat()
        })
        self.data["conversations"] = self.data["conversations"][-100:]
        self.save()
    
    def get_conversation_history(self, limit: int = 10) -> List[Dict]:
        return self.data["conversations"][-limit:]
    
    def record_tool_use(self, tool_name: str, success: bool, details: str = ""):
        self.data["tools_used"].append({
            "tool": tool_name,
            "success": success,
            "details": details,
            "timestamp": datetime.datetime.now().isoformat()
        })
        self.save()

class UI:
    """Rich UI helpers"""
    
    @staticmethod
    def header(text: str):
        console.print(Panel(text, style="bold cyan", border_style="cyan"))
    
    @staticmethod
    def agent_message(agent: str, message: str):
        colors = {
            "orchestrator": "green",
            "planner": "blue",
            "executor": "yellow",
            "reviewer": "magenta",
            "researcher": "cyan",
            "coder": "bright_green",
            "system": "white"
        }
        color = colors.get(agent.lower(), "white")
        console.print(f"[bold {color}]{agent.upper()}:[/bold {color}] {message}")
    
    @staticmethod
    def code_block(code: str, language: str = "python"):
        syntax = Syntax(code, language, theme="monokai", line_numbers=True)
        console.print(Panel(syntax, border_style="dim"))
    
    @staticmethod
    def table(headers: List[str], rows: List[List[str]], title: str = ""):
        table = Table(title=title, show_header=True, header_style="bold magenta")
        for h in headers:
            table.add_column(h)
        for row in rows:
            table.add_row(*row)
        console.print(table)
    
    @staticmethod
    def tree(label: str, items: List[str]):
        tree = Tree(f"[bold]{label}[/bold]")
        for item in items:
            tree.add(item)
        console.print(tree)
    
    @staticmethod
    def confirm(prompt: str) -> bool:
        """Ask for user confirmation"""
        response = console.input(f"[bold yellow]{prompt} (y/n): [/bold yellow]").lower().strip()
        return response in ("y", "yes", "yeah", "sure")
    
    @staticmethod
    def success(message: str):
        console.print(f"[bold green]✓ {message}[/bold green]")
    
    @staticmethod
    def error(message: str):
        console.print(f"[bold red]✗ {message}[/bold red]")
    
    @staticmethod
    def warning(message: str):
        console.print(f"[bold yellow]⚠ {message}[/bold yellow]")
    
    @staticmethod
    def info(message: str):
        console.print(f"[dim]{message}[/dim]")

def hash_string(text: str) -> str:
    """Generate MD5 hash of string"""
    return hashlib.md5(text.encode()).hexdigest()[:8]

def truncate(text: str, max_length: int = 500) -> str:
    """Truncate text with ellipsis"""
    if len(text) <= max_length:
        return text
    return text[:max_length] + "... [truncated]"

def format_size(size_bytes: int) -> str:
    """Format bytes to human readable"""
    for unit in ["B", "KB", "MB", "GB"]:
        if size_bytes < 1024:
            return f"{size_bytes:.1f} {unit}"
        size_bytes /= 1024
    return f"{size_bytes:.1f} TB"
''')

print("utils.py done")