Claude Code: refactor app.py with AgentRouter communication manager

- Add AgentRouter class for centralized message routing between agents
- Implement asyncio.Queue for each agent to prevent state corruption
- Add AgentState model with last_heartbeat timestamp tracking
- Add /health/agents endpoint returning status of all agents
- Add /agents/send, /agents/{agent_id}/receive, /agents/{agent_id}/state endpoints for inter-agent communication
- Integrate agent router with FastAPI lifespan management

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

app.py

@@ -9,6 +9,8 @@ from fastapi.staticfiles import StaticFiles
 from fastapi.responses import FileResponse
 from fastapi import WebSocket
 from pydantic import BaseModel
+from typing import Optional, Dict, List
+from enum import Enum
 import os
 import sys
 from datetime import datetime
@@ -30,19 +32,132 @@ from error_handlers import handle_status_file_read, handle_brain_response, handl
 START_TIME = time.time()
 
 
+# ============================================================================
+# Agent Communication Manager
+# ============================================================================
+
+class AgentRole(str, Enum):
+    """Agent roles in the HuggingClaw World family."""
+    ADAM = "adam"
+    EVE = "eve"
+    CAIN = "cain"
+
+
+class AgentMessage(BaseModel):
+    """Message structure for inter-agent communication."""
+    sender: AgentRole
+    recipient: AgentRole
+    content: str
+    timestamp: str
+    message_id: Optional[str] = None
+
+
+class AgentState(BaseModel):
+    """State tracking for each agent including heartbeat."""
+    agent_id: AgentRole
+    current_state: str = "idle"
+    last_heartbeat: float = 0.0
+    message_queue_size: int = 0
+    is_active: bool = False
+
+
+class AgentRouter:
+    """
+    Centralized message router for inter-agent communication.
+    Uses asyncio.Queue to prevent state corruption.
+    """
+
+    def __init__(self):
+        self._queues: Dict[AgentRole, asyncio.Queue] = {
+            AgentRole.ADAM: asyncio.Queue(),
+            AgentRole.EVE: asyncio.Queue(),
+            AgentRole.CAIN: asyncio.Queue(),
+        }
+        self._states: Dict[AgentRole, AgentState] = {
+            role: AgentState(agent_id=role, last_heartbeat=time.time())
+            for role in AgentRole
+        }
+        self._heartbeat_task: Optional[asyncio.Task] = None
+
+    async def start(self):
+        """Start the agent router background tasks."""
+        self._heartbeat_task = asyncio.create_task(self._heartbeat_loop())
+
+    async def stop(self):
+        """Stop the agent router background tasks."""
+        if self._heartbeat_task:
+            self._heartbeat_task.cancel()
+            try:
+                await self._heartbeat_task
+            except asyncio.CancelledError:
+                pass
+
+    async def _heartbeat_loop(self):
+        """Update heartbeat timestamps every 10 seconds."""
+        while True:
+            await asyncio.sleep(10)
+            now = time.time()
+            for state in self._states.values():
+                state.last_heartbeat = now
+                state.message_queue_size = self._queues[state.agent_id].qsize()
+                # Mark inactive if no heartbeat for 30 seconds
+                state.is_active = (now - state.last_heartbeat) < 30
+
+    async def send_message(self, message: AgentMessage) -> bool:
+        """Send a message to a specific agent's queue."""
+        recipient = message.recipient
+        if recipient not in self._queues:
+            return False
+        await self._queues[recipient].put(message)
+        return True
+
+    async def receive_message(self, agent: AgentRole, timeout: float = 1.0) -> Optional[AgentMessage]:
+        """Receive a message from an agent's queue."""
+        if agent not in self._queues:
+            return None
+        try:
+            return await asyncio.wait_for(self._queues[agent].get(), timeout=timeout)
+        except asyncio.TimeoutError:
+            return None
+
+    def get_all_states(self) -> List[AgentState]:
+        """Get current state of all agents."""
+        return list(self._states.values())
+
+    def get_state(self, agent: AgentRole) -> Optional[AgentState]:
+        """Get state of a specific agent."""
+        return self._states.get(agent)
+
+    def update_state(self, agent: AgentRole, state: str) -> bool:
+        """Update the state of a specific agent."""
+        if agent not in self._states:
+            return False
+        self._states[agent].current_state = state
+        self._states[agent].last_heartbeat = time.time()
+        return True
+
+
+# Global agent router instance
+agent_router = AgentRouter()
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     """Lifespan context manager for startup/shutdown events."""
     # Startup: log system startup
     log_startup("Cain", "1.0.0")
 
+    # Start agent router
+    await agent_router.start()
+
     # Start heartbeat task
     heartbeat_task = asyncio.create_task(heartbeat_loop())
 
     yield
 
-    # Shutdown: cancel heartbeat
+    # Shutdown: cancel heartbeat and stop router
     heartbeat_task.cancel()
+    await agent_router.stop()
 
 
 # Background heartbeat loop
@@ -149,6 +264,55 @@ async def chat(msg: ChatMessage):
     }
 
 
+@app.post("/agents/send")
+async def agent_send(msg: AgentMessage):
+    """Send a message to another agent through the router."""
+    success = await agent_router.send_message(msg)
+    return {
+        "success": success,
+        "message": "Message queued" if success else "Failed to queue message",
+        "timestamp": datetime.utcnow().isoformat() + "+00:00"
+    }
+
+
+@app.get("/agents/{agent_id}/receive")
+async def agent_receive(agent_id: str, timeout: float = 1.0):
+    """Receive a message from the agent's queue."""
+    try:
+        role = AgentRole(agent_id)
+    except ValueError:
+        return {"error": "Invalid agent ID", "valid_agents": [r.value for r in AgentRole]}
+
+    message = await agent_router.receive_message(role, timeout=timeout)
+    if message:
+        return {
+            "message": message.dict(),
+            "timestamp": datetime.utcnow().isoformat() + "+00:00"
+        }
+    return {
+        "message": None,
+        "queue_size": agent_router.get_state(role).message_queue_size,
+        "timestamp": datetime.utcnow().isoformat() + "+00:00"
+    }
+
+
+@app.post("/agents/{agent_id}/state")
+async def agent_update_state(agent_id: str, state: str):
+    """Update the state of an agent."""
+    try:
+        role = AgentRole(agent_id)
+    except ValueError:
+        return {"error": "Invalid agent ID", "valid_agents": [r.value for r in AgentRole]}
+
+    success = agent_router.update_state(role, state)
+    return {
+        "success": success,
+        "agent_id": agent_id,
+        "new_state": state if success else None,
+        "timestamp": datetime.utcnow().isoformat() + "+00:00"
+    }
+
+
 @app.websocket("/ws")
 async def websocket_endpoint(websocket: WebSocket):
     """WebSocket endpoint for real-time dashboard updates."""
@@ -219,6 +383,43 @@ async def debug_health():
     }
 
 
+@app.get("/health/agents")
+async def health_agents():
+    """Health check endpoint returning status of all agents."""
+    agents_status = []
+    now = time.time()
+
+    for state in agent_router.get_all_states():
+        # Calculate time since last heartbeat
+        heartbeat_age = now - state.last_heartbeat
+
+        # Determine health status
+        if heartbeat_age < 15:
+            health = "healthy"
+        elif heartbeat_age < 30:
+            health = "degraded"
+        else:
+            health = "unhealthy"
+
+        agents_status.append({
+            "agent_id": state.agent_id,
+            "current_state": state.current_state,
+            "is_active": state.is_active,
+            "health": health,
+            "last_heartbeat": state.last_heartbeat,
+            "heartbeat_age_seconds": round(heartbeat_age, 2),
+            "message_queue_size": state.message_queue_size,
+            "last_heartbeat_iso": datetime.fromtimestamp(state.last_heartbeat).isoformat() + "+00:00"
+        })
+
+    return {
+        "agents": agents_status,
+        "total_agents": len(agents_status),
+        "active_agents": sum(1 for a in agents_status if a["is_active"]),
+        "timestamp": datetime.utcnow().isoformat() + "+00:00"
+    }
+
+
 if __name__ == "__main__":
     import uvicorn
     uvicorn.run(app, host="0.0.0.0", port=7860)