""" Chat agent backend — calls Claude API with tools for agent orchestration. Loaded by serve.py as a blueprint. """ import json import subprocess import time from pathlib import Path import anthropic AGENTS_DIR = Path("/data/cameron/para/.agents") PROJECT_ROOT = Path("/data/cameron/para") # Load context files for system prompt def _load_system_prompt(): context_files = { "CLAUDE.md (project overview)": PROJECT_ROOT / "CLAUDE.md", "scientist.MD (verification guidelines)": PROJECT_ROOT / "scientist.MD", "GUIDELINES.md (communication protocol)": AGENTS_DIR / "shared" / "GUIDELINES.md", "REPORT_FORMAT.md (report structure)": AGENTS_DIR / "shared" / "REPORT_FORMAT.md", } context = "" for label, path in context_files.items(): if path.exists(): context += f"\n\n--- {label} ---\n{path.read_text()}" # Load agent roles and statuses agent_status = "" for agent_dir in sorted(AGENTS_DIR.iterdir()): role_file = agent_dir / "ROLE.md" status_file = agent_dir / "status.md" if role_file.exists(): status = status_file.read_text().strip() if status_file.exists() else "unknown" agent_status += f"\n- **{agent_dir.name}**: status={status}" return f"""You are the PARA project manager chat assistant. You help Cameron coordinate experiments, analyze results, and manage the agent team. You have access to tools to interact with experiment agents running in tmux tabs, read files, and run commands. ACTIVE AGENTS:{agent_status} PROJECT CONTEXT:{context} Be concise and direct. When asked about experiment status, use your tools to check the latest state rather than relying on stale information.""" # Tool definitions TOOLS = [ { "name": "send_task", "description": "Send a task/message to an agent via tmux. The agent will see it as a user message in their Claude Code session.", "input_schema": { "type": "object", "properties": { "agent": {"type": "string", "description": "Agent name: backbones, vid_model, or droid"}, "message": {"type": "string", "description": "The task or message to send"}, }, "required": ["agent", "message"], }, }, { "name": "read_outbox", "description": "Read an agent's outbox for their latest results/response.", "input_schema": { "type": "object", "properties": { "agent": {"type": "string", "description": "Agent name"}, }, "required": ["agent"], }, }, { "name": "get_agent_status", "description": "Get an agent's current status (idle, working, done, blocked).", "input_schema": { "type": "object", "properties": { "agent": {"type": "string", "description": "Agent name"}, }, "required": ["agent"], }, }, { "name": "capture_pane", "description": "Capture recent terminal output from an agent's tmux pane to see what they're doing.", "input_schema": { "type": "object", "properties": { "agent": {"type": "string", "description": "Agent name"}, "lines": {"type": "integer", "description": "Number of lines to capture (default 30)", "default": 30}, }, "required": ["agent"], }, }, { "name": "read_file", "description": "Read a file from the filesystem.", "input_schema": { "type": "object", "properties": { "path": {"type": "string", "description": "Absolute file path"}, "max_lines": {"type": "integer", "description": "Max lines to read (default 200)", "default": 200}, }, "required": ["path"], }, }, { "name": "list_reports", "description": "List all reports on the experiment dashboard.", "input_schema": { "type": "object", "properties": {}, }, }, { "name": "run_command", "description": "Run a shell command. Use for checking GPU status, disk usage, process lists, etc. Do NOT use for destructive operations.", "input_schema": { "type": "object", "properties": { "command": {"type": "string", "description": "Shell command to run"}, "timeout": {"type": "integer", "description": "Timeout in seconds (default 30)", "default": 30}, }, "required": ["command"], }, }, { "name": "list_directory", "description": "List contents of a directory.", "input_schema": { "type": "object", "properties": { "path": {"type": "string", "description": "Directory path"}, }, "required": ["path"], }, }, ] def _find_window(agent_name): """Find tmux window by name.""" result = subprocess.run( ["tmux", "list-windows", "-a", "-F", "#{session_name}:#{window_index} #{window_name}"], capture_output=True, text=True, ) for line in result.stdout.strip().split("\n"): parts = line.split(" ", 1) if len(parts) == 2 and parts[1] == agent_name: return parts[0] return None def execute_tool(name, input_data): """Execute a tool and return the result string.""" try: if name == "send_task": agent = input_data["agent"] message = input_data["message"] # Write to inbox inbox = AGENTS_DIR / agent / "inbox.md" from datetime import datetime timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with open(inbox, "a") as f: f.write(f"\n---\n## Task — {timestamp}\n\n{message}\n") # Send via tmux target = _find_window(agent) if target: subprocess.run(["tmux", "send-keys", "-t", target, message, "Enter"]) time.sleep(1) subprocess.run(["tmux", "send-keys", "-t", target, "Enter"]) return f"Task sent to {agent} via tmux and written to inbox." return f"Task written to {agent}/inbox.md but tmux window not found." elif name == "read_outbox": agent = input_data["agent"] outbox = AGENTS_DIR / agent / "outbox.md" if outbox.exists(): content = outbox.read_text().strip() return content if content else "(outbox is empty)" return "(no outbox file)" elif name == "get_agent_status": agent = input_data["agent"] status_file = AGENTS_DIR / agent / "status.md" if status_file.exists(): return status_file.read_text().strip() return "unknown" elif name == "capture_pane": agent = input_data["agent"] lines = input_data.get("lines", 30) target = _find_window(agent) if not target: return f"tmux window '{agent}' not found" result = subprocess.run( ["tmux", "capture-pane", "-t", target, "-p", "-S", f"-{lines}"], capture_output=True, text=True, ) return result.stdout or "(empty pane)" elif name == "read_file": path = Path(input_data["path"]) max_lines = input_data.get("max_lines", 200) if not path.exists(): return f"File not found: {path}" lines = path.read_text().split("\n")[:max_lines] return "\n".join(lines) elif name == "list_reports": reports_dir = AGENTS_DIR / "reports" reports = [] for agent_dir in sorted(reports_dir.iterdir()): if agent_dir.is_dir() and agent_dir.name not in ("media",): for f in sorted(agent_dir.glob("*.html"), reverse=True): from datetime import datetime date = datetime.fromtimestamp(f.stat().st_mtime).strftime("%Y-%m-%d %H:%M") reports.append(f"{agent_dir.name}/{f.name} ({date})") return "\n".join(reports) if reports else "No reports yet." elif name == "run_command": cmd = input_data["command"] timeout = input_data.get("timeout", 30) # Safety: block destructive commands dangerous = ["rm -rf", "mkfs", "dd if=", "> /dev/", "kill -9", "shutdown", "reboot"] if any(d in cmd for d in dangerous): return "Blocked: potentially destructive command." result = subprocess.run( cmd, shell=True, capture_output=True, text=True, timeout=timeout, cwd=str(PROJECT_ROOT), ) output = result.stdout + result.stderr return output[:5000] if output else "(no output)" elif name == "list_directory": path = Path(input_data["path"]) if not path.is_dir(): return f"Not a directory: {path}" entries = [] for item in sorted(path.iterdir()): prefix = "📁 " if item.is_dir() else "📄 " entries.append(f"{prefix}{item.name}") return "\n".join(entries[:100]) return f"Unknown tool: {name}" except Exception as e: return f"Error: {str(e)}" # Conversation state (in-memory, per-session) conversations = {} def chat(session_id, user_message, model="claude-sonnet-4-20250514"): """ Handle a chat message. Returns response text. Handles tool use loops internally. """ client = anthropic.Anthropic() if session_id not in conversations: conversations[session_id] = [] messages = conversations[session_id] messages.append({"role": "user", "content": user_message}) system_prompt = _load_system_prompt() while True: response = client.messages.create( model=model, max_tokens=4096, system=system_prompt, tools=TOOLS, messages=messages, ) # Collect text and tool use blocks assistant_content = response.content messages.append({"role": "assistant", "content": assistant_content}) # Check if there are tool calls tool_uses = [b for b in assistant_content if b.type == "tool_use"] if not tool_uses: # No tool calls — extract text and return text_parts = [b.text for b in assistant_content if b.type == "text"] return "\n".join(text_parts) # Execute tools and continue the loop tool_results = [] for tool_use in tool_uses: result = execute_tool(tool_use.name, tool_use.input) tool_results.append({ "type": "tool_result", "tool_use_id": tool_use.id, "content": result, }) messages.append({"role": "user", "content": tool_results}) # Loop continues — model will process tool results def clear_conversation(session_id): """Clear conversation history for a session.""" if session_id in conversations: del conversations[session_id]