External/roboco
1
0
Fork 0
mirror of https://github.com/rennf93/roboco.git synced 2026-08-03 07:23:24 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
afde0d54419ed6adc3fdb69951634aadc8049bdc
roboco/roboco/runtime/orchestrator.py
T

2097 lines
71 KiB
Python
Raw Blame History

"""
Agent Orchestrator
Manages Claude Code containers for all RoboCo agents.
Handles spawning, monitoring, health checks, and graceful shutdown.
The orchestrator is the BRAIN of the system:
- Checks for work BEFORE spawning agents (no wasteful spawns)
- Claims tasks on behalf of agents before spawning
- Agents receive their assignment at spawn time
- Agents scan for more work after completing a task
- Agents only call roboco_agent_idle() when truly no work remains
"""
import asyncio
import contextlib
import json
import os
import tempfile
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, ClassVar
import httpx
import structlog
from fastapi import status as http_status
from roboco.agents_config import get_agent_role
from roboco.config import settings
from roboco.models.runtime import (
MODEL_MAP,
ROLE_MODEL_MAP,
AgentInstance,
OrchestratorAgentConfig,
OrchestratorAgentState,
WaitingRecord,
)
from roboco.seeds.initial_data import AGENT_UUIDS
logger = structlog.get_logger()
# Reverse mapping: UUID -> slug
UUID_TO_SLUG = {uuid: slug for slug, uuid in AGENT_UUIDS.items()}
# Re-export for backwards compatibility
AgentState = OrchestratorAgentState
AgentConfig = OrchestratorAgentConfig
# Docker configuration
AGENT_NETWORK = "roboco_default"
AGENT_BASE_IMAGE = "roboco-agent-base"
# Role -> Image mapping
# Specialized images extend the base with role-specific tools
AGENT_IMAGES: dict[str, str] = {
# Backend
"be-dev-1": "roboco-agent-dev-be",
"be-dev-2": "roboco-agent-dev-be",
"be-qa": "roboco-agent-qa-be",
"be-pm": "roboco-agent-pm",
"be-doc": "roboco-agent-doc",
# Frontend
"fe-dev-1": "roboco-agent-dev-fe",
"fe-dev-2": "roboco-agent-dev-fe",
"fe-qa": "roboco-agent-qa-fe",
"fe-pm": "roboco-agent-pm",
"fe-doc": "roboco-agent-doc",
# UX/UI
"ux-dev": "roboco-agent-ux",
"ux-qa": "roboco-agent-ux", # Uses same as dev for now
"ux-pm": "roboco-agent-pm",
"ux-doc": "roboco-agent-doc",
# Board
"main-pm": "roboco-agent-pm",
"product-owner": "roboco-agent-pm",
"head-marketing": "roboco-agent-pm",
"auditor": "roboco-agent-pm",
}
def get_agent_image(agent_id: str) -> str:
"""Get the Docker image for an agent."""
return AGENT_IMAGES.get(agent_id, AGENT_BASE_IMAGE)
# When running in a container, we need host paths for volume mounts.
# These can be overridden via environment variables.
CLAUDE_AUTH_HOST_PATH = os.environ.get(
"ROBOCO_HOST_CLAUDE_DIR",
str(Path.home() / ".claude"),
)
PROJECT_HOST_PATH = os.environ.get("ROBOCO_HOST_PROJECT_DIR", "")
DATA_HOST_PATH = os.environ.get("ROBOCO_HOST_DATA_DIR", "")
# =============================================================================
# ORCHESTRATOR
# =============================================================================
class AgentOrchestrator:
"""
Manages Claude Code containers for all agents.
Responsibilities:
- Spawn agents as Docker containers
- Monitor health via docker inspect
- Handle waiting states and respawning
- Provide status API
- Cost-efficient on-demand spawning
"""
def __init__(
self,
blueprints_dir: Path | None = None,
mcp_config_dir: Path | None = None,
project_root: Path | None = None,
dispatcher_interval: int = 30,
):
self.blueprints_dir = blueprints_dir or Path("agents/blueprints")
self.mcp_config_dir = mcp_config_dir or Path(".mcp")
self.project_root = project_root or Path.cwd()
self.dispatcher_interval = dispatcher_interval
self._instances: dict[str, AgentInstance] = {}
self._waiting_records: dict[str, WaitingRecord] = {}
self._health_task: asyncio.Task | None = None
self._dispatcher_task: asyncio.Task | None = None
self._running = False
self._lock = asyncio.Lock()
# =========================================================================
# LIFECYCLE
# =========================================================================
async def start(self) -> None:
"""Start the orchestrator."""
self._running = True
# Ensure agent image is built
await self._ensure_agent_image()
# Ensure agent Claude settings have MCP tools allowed
self._ensure_agent_claude_settings()
# Start background tasks
self._health_task = asyncio.create_task(self._health_loop())
self._dispatcher_task = asyncio.create_task(self._dispatcher_loop())
logger.info(
"Orchestrator started",
dispatcher_interval=self.dispatcher_interval,
internal_api_url=self._api_url,
)
async def stop(self) -> None:
"""Stop the orchestrator and all agents."""
self._running = False
# Cancel background tasks
if self._health_task:
self._health_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await self._health_task
if self._dispatcher_task:
self._dispatcher_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await self._dispatcher_task
# Stop all agents
for agent_id in list(self._instances.keys()):
await self.stop_agent(agent_id)
logger.info("Orchestrator stopped")
async def _ensure_agent_image(self, agent_id: str | None = None) -> None:
"""Ensure the agent Docker images are built.
Builds base image first, then specialized image if agent_id provided.
"""
# Determine build context
if PROJECT_HOST_PATH:
build_context = PROJECT_HOST_PATH
docker_dir = f"{PROJECT_HOST_PATH}/docker"
else:
build_context = str(self.project_root)
docker_dir = str(self.project_root / "docker")
# Always ensure base image exists
await self._build_image_if_missing(
AGENT_BASE_IMAGE,
f"{docker_dir}/agent-base.Dockerfile",
build_context,
)
# Build specialized image if agent specified
if agent_id:
image = get_agent_image(agent_id)
if image != AGENT_BASE_IMAGE:
# Map image name to dockerfile
dockerfile_map = {
"roboco-agent-pm": "agent-pm.Dockerfile",
"roboco-agent-dev-be": "agent-dev-be.Dockerfile",
"roboco-agent-dev-fe": "agent-dev-fe.Dockerfile",
"roboco-agent-qa-be": "agent-qa-be.Dockerfile",
"roboco-agent-qa-fe": "agent-qa-fe.Dockerfile",
"roboco-agent-doc": "agent-doc.Dockerfile",
"roboco-agent-ux": "agent-ux.Dockerfile",
}
dockerfile = dockerfile_map.get(image)
if dockerfile:
await self._build_image_if_missing(
image,
f"{docker_dir}/{dockerfile}",
build_context,
)
async def _build_image_if_missing(
self, image_name: str, dockerfile_path: str, build_context: str
) -> None:
"""Build a Docker image if it doesn't exist."""
# Check if image exists
proc = await asyncio.create_subprocess_exec(
"docker",
"image",
"inspect",
image_name,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL,
)
await proc.wait()
if proc.returncode != 0:
logger.info("Building Docker image...", image=image_name)
proc = await asyncio.create_subprocess_exec(
"docker",
"build",
"-t",
image_name,
"-f",
dockerfile_path,
build_context,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
_, stderr = await proc.communicate()
if proc.returncode != 0:
raise RuntimeError(
f"Failed to build image {image_name}: {stderr.decode()}"
)
logger.info("Docker image built successfully", image=image_name)
def _ensure_agent_claude_settings(self) -> None:
"""
Ensure agent Claude settings have RoboCo MCP tools pre-allowed.
This prevents agents from needing interactive permission approval
for essential MCP tools like roboco_agent_idle.
"""
# RoboCo MCP tools that should always be allowed for agents
roboco_allowed_tools = [
# Task management - always needed
"mcp__roboco-task__*",
# Messaging - always needed for communication
"mcp__roboco-message__*",
# Notifications - always needed
"mcp__roboco-notify__*",
# Journal - always needed for reflection
"mcp__roboco-journal__*",
# File operations for documenters and developers
# Note: // prefix = absolute path (container paths like /app/docs)
"Write(//app/docs/**)",
"Write(//app/CHANGELOG.md)",
"Write(//app/README.md)",
"Edit(//app/docs/**)",
"Edit(//app/CHANGELOG.md)",
"Edit(//app/README.md)",
]
# Path to agent Claude settings (shared across all agents)
# Always use CLAUDE_AUTH_HOST_PATH - agents mount from this location
claude_dir = Path(CLAUDE_AUTH_HOST_PATH)
settings_path = claude_dir / "settings.json"
# Load existing settings or create new
if settings_path.exists():
try:
settings = json.loads(settings_path.read_text())
except json.JSONDecodeError:
settings = {}
else:
settings = {}
# Ensure permissions structure exists
if "permissions" not in settings:
settings["permissions"] = {}
if "allow" not in settings["permissions"]:
settings["permissions"]["allow"] = []
# Add RoboCo tools if not already present
existing_allow = set(settings["permissions"]["allow"])
tools_added = []
for tool in roboco_allowed_tools:
if tool not in existing_allow:
settings["permissions"]["allow"].append(tool)
tools_added.append(tool)
# Only write if we added tools
if tools_added:
claude_dir.mkdir(parents=True, exist_ok=True)
settings_path.write_text(json.dumps(settings, indent=2))
logger.info(
"Updated agent Claude settings with allowed MCP tools",
tools_added=tools_added,
)
# =========================================================================
# AGENT SPAWNING
# =========================================================================
async def spawn_agent(
self,
agent_id: str,
initial_prompt: str | None = None,
task_id: str | None = None,
model: str | None = None,
) -> AgentInstance:
"""
Spawn a Claude Code container for an agent.
Args:
agent_id: Agent identifier (e.g., "be-dev-1")
initial_prompt: Optional initial prompt
task_id: Optional task ID being worked on
model: Override model selection
Returns:
AgentInstance handle
"""
async with self._lock:
# Check if already running
if agent_id in self._instances:
existing = self._instances[agent_id]
if existing.state not in (AgentState.OFFLINE, AgentState.WAITING_LONG):
logger.warning(
"Agent already running",
agent_id=agent_id,
state=existing.state,
)
return existing
# Get blueprint path
blueprint_path = self._get_blueprint_path(agent_id)
if not blueprint_path.exists():
raise FileNotFoundError(f"Blueprint not found: {blueprint_path}")
# Ensure agent Claude settings have MCP tools allowed
self._ensure_agent_claude_settings()
# Ensure agent-specific Docker image is built
await self._ensure_agent_image(agent_id)
# Generate MCP config
mcp_config_path = await self._generate_mcp_config(agent_id)
# Determine model using canonical role name from agents_config
if not model:
canonical_role = get_agent_role(agent_id)
model = ROLE_MODEL_MAP.get(canonical_role, "sonnet")
# Create config
config = AgentConfig(
agent_id=agent_id,
blueprint_path=blueprint_path,
model=model,
mcp_config_path=mcp_config_path,
)
# Create instance
instance = AgentInstance(
agent_id=agent_id,
state=AgentState.STARTING,
config=config,
current_task_id=task_id,
)
self._instances[agent_id] = instance
# Spawn the container
try:
container_id = await self._spawn_container(config, initial_prompt)
instance.container_id = container_id
instance.state = AgentState.ACTIVE
instance.started_at = datetime.now(UTC)
instance.last_activity = datetime.now(UTC)
logger.info(
"Agent spawned",
agent_id=agent_id,
container_id=container_id[:12],
model=model,
task_id=task_id,
)
return instance
except Exception as e:
instance.state = AgentState.OFFLINE
instance.error_count += 1
logger.error(
"Failed to spawn agent",
agent_id=agent_id,
error=str(e),
)
raise
async def _spawn_container(
self,
config: AgentConfig,
initial_prompt: str | None = None,
) -> str:
"""Spawn a Docker container for the agent."""
container_name = f"roboco-agent-{config.agent_id}"
# Remove existing container if any
await self._remove_container(container_name)
# Determine host paths for volume mounts
# When running in a container, use PROJECT_HOST_PATH; otherwise use local paths
if not config.mcp_config_path:
raise RuntimeError("MCP config path not set")
if PROJECT_HOST_PATH:
# Running inside orchestrator container - use host paths
blueprints_host = f"{PROJECT_HOST_PATH}/agents/blueprints"
docs_host = f"{PROJECT_HOST_PATH}/docs"
claude_host = CLAUDE_AUTH_HOST_PATH
mcp_config_host = (
f"{DATA_HOST_PATH}/mcp-configs/{config.mcp_config_path.name}"
)
else:
# Running directly on host
blueprints_host = str(self.blueprints_dir.absolute())
docs_host = str(self.blueprints_dir.parent / "docs")
claude_host = CLAUDE_AUTH_HOST_PATH
mcp_config_host = str(config.mcp_config_path)
# Build docker run command
cmd = [
"docker",
"run",
"-d",
"--name",
container_name,
"--network",
AGENT_NETWORK,
# Mount Claude auth (needs write access for debug logs)
"-v",
f"{claude_host}:/home/agent/.claude",
# Mount blueprints
"-v",
f"{blueprints_host}:/app/agents/blueprints:ro",
# Mount docs directory (documenters need write access)
"-v",
f"{docs_host}:/app/docs",
# Mount MCP config
"-v",
f"{mcp_config_host}:/app/mcp-config.json:ro",
# Environment
"-e",
f"ROBOCO_AGENT_ID={config.agent_id}",
"-e",
"ROBOCO_API_URL=http://roboco-orchestrator:8000",
# The image (role-specific)
get_agent_image(config.agent_id),
# Claude Code arguments
"--model",
MODEL_MAP.get(config.model, config.model),
"--system-prompt-file",
f"/app/agents/blueprints/{self._get_blueprint_rel_path(config.agent_id)}",
"--mcp-config",
"/app/mcp-config.json",
"--output-format",
"stream-json",
"--verbose",
# Always provide a prompt (required for non-interactive mode)
# If no task assignment provided, agent should follow standard workflow:
# SCAN for work -> CLAIM if available -> or IDLE if no work
"-p",
initial_prompt
or (
"You may have been spawned without a specific task assignment. "
"Follow your standard workflow:\n\n"
"1. Call `roboco_task_scan()` to find available work for your role\n"
"2. If tasks are found, claim one with `roboco_task_claim(task_id)` "
"and begin the full task lifecycle "
"(UNDERSTAND -> PLAN -> EXECUTE -> VERIFY -> HANDOFF)\n"
"3. If no tasks are available, call `roboco_agent_idle()` "
"to shutdown gracefully\n\n"
"Start now by scanning for work."
),
]
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await proc.communicate()
if proc.returncode != 0:
raise RuntimeError(f"Failed to start container: {stderr.decode()}")
container_id = stdout.decode().strip()
return container_id
async def _remove_container(self, container_name: str) -> None:
"""Remove a container if it exists."""
proc = await asyncio.create_subprocess_exec(
"docker",
"rm",
"-f",
container_name,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL,
)
await proc.wait()
async def _generate_mcp_config(self, agent_id: str) -> Path:
"""Generate role-aware MCP config for an agent.
Different roles get different MCP server access:
- All agents: task, message, journal
- PMs only: notify (for sending notifications)
"""
# MCP servers run inside agent containers, need to connect via Docker network
if PROJECT_HOST_PATH:
api_url = "http://roboco-orchestrator:8000"
else:
api_url = f"http://127.0.0.1:{settings.port}"
mcp_env = {
"ROBOCO_API_URL": api_url,
"ROBOCO_AGENT_ID": agent_id,
}
# Base MCP servers - all agents get these
mcp_servers: dict[str, dict[str, Any]] = {
"roboco-task": {
"command": "uv",
"args": ["run", "python", "-m", "roboco.mcp.task_server", agent_id],
"env": mcp_env,
},
"roboco-message": {
"command": "uv",
"args": [
"run",
"python",
"-m",
"roboco.mcp.message_server",
agent_id,
],
"env": mcp_env,
},
"roboco-journal": {
"command": "uv",
"args": [
"run",
"python",
"-m",
"roboco.mcp.journal_server",
agent_id,
],
"env": mcp_env,
},
}
# Notify server - everyone can READ notifications, only PMs can SEND
# (permission check happens at handler level)
mcp_servers["roboco-notify"] = {
"command": "uv",
"args": [
"run",
"python",
"-m",
"roboco.mcp.notify_server",
agent_id,
],
"env": mcp_env,
}
config: dict[str, Any] = {"mcpServers": mcp_servers}
# Write to shared config directory (mounted in both orchestrator and agents)
# When running in container: /app/mcp-configs -> host's ./data/mcp-configs
# When running on host: use temp directory
if DATA_HOST_PATH:
# Running in container - use shared mounted directory
config_dir = Path("/app/mcp-configs")
config_dir.mkdir(parents=True, exist_ok=True)
else:
# Running on host - use temp directory
config_dir = Path(tempfile.gettempdir())
config_path = config_dir / f"roboco-mcp-{agent_id}.json"
config_path.write_text(json.dumps(config, indent=2))
return config_path
def _get_blueprint_path(self, agent_id: str) -> Path:
"""Get blueprint path for an agent."""
role = self._get_blueprint_role(agent_id)
team = self._get_agent_team(agent_id)
if team == "backend":
cell_dir = "backend"
elif team == "frontend":
cell_dir = "frontend"
elif team == "ux_ui":
cell_dir = "ux_ui"
else:
cell_dir = "board"
blueprint_file = f"{role.replace('_', '-')}.md"
return self.blueprints_dir / cell_dir / blueprint_file
def _get_blueprint_rel_path(self, agent_id: str) -> str:
"""Get relative blueprint path for container mount."""
role = self._get_blueprint_role(agent_id)
team = self._get_agent_team(agent_id)
if team == "backend":
cell_dir = "backend"
elif team == "frontend":
cell_dir = "frontend"
elif team == "ux_ui":
cell_dir = "ux_ui"
else:
cell_dir = "board"
blueprint_file = f"{role.replace('_', '-')}.md"
return f"{cell_dir}/{blueprint_file}"
def _get_blueprint_role(self, agent_id: str) -> str:
"""Get blueprint-specific role name from agent_id (used for file paths)."""
role_map = {
"be-dev-1": "be-dev",
"be-dev-2": "be-dev",
"fe-dev-1": "fe-dev",
"fe-dev-2": "fe-dev",
"ux-dev": "ux-dev",
"be-qa": "be-qa",
"fe-qa": "fe-qa",
"ux-qa": "ux-qa",
"be-pm": "be-pm",
"fe-pm": "fe-pm",
"ux-pm": "ux-pm",
"be-doc": "be-documenter",
"fe-doc": "fe-documenter",
"ux-doc": "ux-documenter",
"main-pm": "main-pm",
"product-owner": "product-owner",
"head-marketing": "head-marketing",
"auditor": "auditor",
}
return role_map.get(agent_id, agent_id)
# Static team mappings for management agents
_AGENT_TEAM_MAP: ClassVar[dict[str, str]] = {
"main-pm": "main_pm",
"product-owner": "board",
"auditor": "board",
"head-marketing": "marketing",
}
def _get_agent_team(self, agent_id: str) -> str | None:
"""Get team from agent_id."""
# Check static mappings first
if agent_id in self._AGENT_TEAM_MAP:
return self._AGENT_TEAM_MAP[agent_id]
# Check cell prefixes
prefix_map = {"be-": "backend", "fe-": "frontend", "ux-": "ux_ui"}
for prefix, team in prefix_map.items():
if agent_id.startswith(prefix):
return team
return None
def _resolve_agent_slug(self, agent_id_or_uuid: str) -> str:
"""Resolve agent UUID to slug. Returns input if already a slug."""
# Check if it's a known UUID and convert to slug
if agent_id_or_uuid in UUID_TO_SLUG:
return UUID_TO_SLUG[agent_id_or_uuid]
# Already a slug or unknown UUID
return agent_id_or_uuid
# =========================================================================
# AGENT STOPPING
# =========================================================================
async def stop_agent(self, agent_id: str, graceful: bool = True) -> None:
"""Stop an agent container."""
async with self._lock:
if agent_id not in self._instances:
return
instance = self._instances[agent_id]
if instance.container_id:
instance.state = AgentState.STOPPING
container_name = f"roboco-agent-{agent_id}"
if graceful:
# Graceful stop with timeout
proc = await asyncio.create_subprocess_exec(
"docker",
"stop",
"-t",
"10",
container_name,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL,
)
await proc.wait()
else:
# Force kill
proc = await asyncio.create_subprocess_exec(
"docker",
"kill",
container_name,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL,
)
await proc.wait()
# Remove container
await self._remove_container(container_name)
instance.state = AgentState.OFFLINE
instance.container_id = None
logger.info("Agent stopped", agent_id=agent_id)
# =========================================================================
# WAITING STATE MANAGEMENT
# =========================================================================
async def mark_waiting_long(
self,
agent_id: str,
waiting_for: str,
task_id: str | None = None,
context: dict[str, Any] | None = None,
) -> None:
"""
Mark an agent as WAITING_LONG and terminate.
The agent will be respawned when the wait condition is resolved.
"""
record = WaitingRecord(
agent_id=agent_id,
task_id=task_id,
waiting_for=waiting_for,
waiting_since=datetime.now(UTC),
context=context or {},
)
self._waiting_records[agent_id] = record
# Stop the agent
await self.stop_agent(agent_id)
# Update state
if agent_id in self._instances:
self._instances[agent_id].state = AgentState.WAITING_LONG
self._instances[agent_id].waiting_for = waiting_for
self._instances[agent_id].waiting_context = context or {}
logger.info(
"Agent marked as waiting_long",
agent_id=agent_id,
waiting_for=waiting_for,
task_id=task_id,
)
async def resolve_wait(
self,
agent_id: str,
resolution: dict[str, Any],
) -> AgentInstance | None:
"""
Resolve a wait condition and respawn the agent.
Args:
agent_id: The waiting agent
resolution: Details about the resolution
Returns:
Respawned AgentInstance or None
"""
if agent_id not in self._waiting_records:
return None
record = self._waiting_records[agent_id]
del self._waiting_records[agent_id]
# Generate resume prompt
resume_prompt = self._generate_resume_prompt(record, resolution)
# Respawn
return await self.spawn_agent(
agent_id=agent_id,
initial_prompt=resume_prompt,
task_id=record.task_id,
)
def _generate_resume_prompt(
self,
record: WaitingRecord,
resolution: dict[str, Any],
) -> str:
"""Generate a resume prompt for a respawning agent."""
if record.waiting_for == "blocker_resolution":
return f"""
You were working on TASK-{record.task_id} and got blocked.
The blocker has been resolved: {resolution.get("details", "Resolved")}
Resume by:
1. Reading your checkpoint from .tasks/active/TASK-{record.task_id}/
2. Call roboco_task_unblock("{record.task_id}")
3. Continue from where you left off
"""
elif record.waiting_for == "qa_result":
if resolution.get("passed"):
return f"""
TASK-{record.task_id} has passed QA review.
The task is now awaiting documentation.
You may return to scanning for new work with roboco_task_scan().
"""
else:
return f"""
TASK-{record.task_id} needs revision based on QA feedback.
QA notes: {resolution.get("notes", "See task for details")}
Resume by:
1. Reading the QA feedback
2. Updating your TODOs to address each issue
3. Making the fixes
4. Re-submitting for QA
"""
elif record.waiting_for == "answer":
return f"""
You asked a question about TASK-{record.task_id}:
Your question: {record.context.get("question", "Unknown")}
Answer received: {resolution.get("answer", "Unknown")}
Resume by incorporating this information and continuing from where you stopped.
"""
elif record.waiting_for == "assignment":
return f"""
You have been assigned a new task: TASK-{resolution.get("task_id")}
Start by:
1. Call roboco_task_get("{resolution.get("task_id")}") to get details
2. Follow the standard workflow: UNDERSTAND → PLAN → EXECUTE → VERIFY → NOTES
"""
else:
return f"Resuming. Wait condition '{record.waiting_for}' resolved."
# =========================================================================
# HEALTH MONITORING
# =========================================================================
async def _health_loop(self) -> None:
"""Background health check loop."""
while self._running:
try:
await asyncio.sleep(30) # Check every 30 seconds
await self._check_health()
except asyncio.CancelledError:
break
except Exception as e:
logger.error("Health check error", error=str(e))
async def _check_health(self) -> None:
"""Check health of all running agents."""
for agent_id, instance in list(self._instances.items()):
if instance.state not in (AgentState.ACTIVE, AgentState.WAITING_SHORT):
continue
if instance.container_id is None:
continue
# Check if container is still running
container_name = f"roboco-agent-{agent_id}"
proc = await asyncio.create_subprocess_exec(
"docker",
"inspect",
"-f",
"{{.State.Running}}",
container_name,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.DEVNULL,
)
stdout, _ = await proc.communicate()
is_running = stdout.decode().strip() == "true"
if not is_running:
cid = instance.container_id[:12] if instance.container_id else None
logger.warning(
"Agent container stopped",
agent_id=agent_id,
container_id=cid,
)
instance.state = AgentState.OFFLINE
instance.error_count += 1
instance.container_id = None
# Auto-restart if not too many errors
max_retries = 3
if instance.error_count < max_retries:
logger.info("Auto-restarting agent", agent_id=agent_id)
await self.spawn_agent(
agent_id=agent_id,
task_id=instance.current_task_id,
)
# =========================================================================
# STATUS API
# =========================================================================
def get_state(self, agent_id: str) -> AgentState:
"""Get current state of an agent."""
if agent_id not in self._instances:
return AgentState.OFFLINE
return self._instances[agent_id].state
def get_instance(self, agent_id: str) -> AgentInstance | None:
"""Get instance for an agent."""
return self._instances.get(agent_id)
def get_all_instances(self) -> dict[str, AgentInstance]:
"""Get all agent instances."""
return dict(self._instances)
def get_waiting_agents(self) -> dict[str, WaitingRecord]:
"""Get all waiting agents."""
return dict(self._waiting_records)
def get_status_summary(self) -> dict[str, Any]:
"""Get summary of all agent states."""
by_state: dict[str, int] = {}
agents: list[dict[str, Any]] = []
for state in AgentState:
count = sum(1 for i in self._instances.values() if i.state == state)
if count > 0:
by_state[state.value] = count
for agent_id, instance in self._instances.items():
cid = instance.container_id[:12] if instance.container_id else None
agents.append(
{
"agent_id": agent_id,
"state": instance.state.value,
"container_id": cid,
"task_id": instance.current_task_id,
"error_count": instance.error_count,
"started_at": instance.started_at.isoformat()
if instance.started_at
else None,
}
)
return {
"total": len(self._instances),
"by_state": by_state,
"waiting_count": len(self._waiting_records),
"agents": agents,
}
# =========================================================================
# SMART DISPATCHER - API HELPERS
# =========================================================================
@property
def _api_url(self) -> str:
"""Get the internal API URL for task/notification queries."""
return settings.internal_api_url
def _is_agent_active(self, agent_id: str) -> bool:
"""Check if an agent is currently running."""
if agent_id not in self._instances:
return False
return self._instances[agent_id].state == AgentState.ACTIVE
def _select_agent_for_cell(self, cell: str, role: str) -> str | None:
"""
Select the best available agent for a cell and role.
Prefers agents that are not currently active.
For developers, uses round-robin among candidates.
"""
prefix_map = {"backend": "be", "frontend": "fe", "ux_ui": "ux"}
prefix = prefix_map.get(cell)
if not prefix:
return None
# Build candidate list based on role
if role == "dev":
if prefix == "ux":
candidates = ["ux-dev"]
else:
candidates = [f"{prefix}-dev-1", f"{prefix}-dev-2"]
elif role == "qa":
candidates = [f"{prefix}-qa"]
elif role == "doc":
candidates = [f"{prefix}-doc"]
elif role == "pm":
candidates = [f"{prefix}-pm"]
else:
return None
# Prefer non-active agents
for agent_id in candidates:
if not self._is_agent_active(agent_id):
return agent_id
# All active - return first (task will queue for them via scan)
return candidates[0]
async def _claim_task_for_agent(
self,
client: httpx.AsyncClient,
task_id: str,
agent_id: str,
) -> bool:
"""Claim a task on behalf of an agent before spawning."""
try:
resp = await client.post(
f"{self._api_url}/tasks/{task_id}/claim",
json={"agent_id": agent_id},
)
if resp.status_code == http_status.HTTP_200_OK:
logger.info(
"Task claimed for agent",
task_id=task_id,
agent_id=agent_id,
)
return True
logger.warning(
"Failed to claim task",
task_id=task_id,
agent_id=agent_id,
status=resp.status_code,
)
except Exception as e:
logger.error("Claim task error", task_id=task_id, error=str(e))
return False
async def _fetch_tasks(
self,
client: httpx.AsyncClient,
status: str | list[str],
team: str | None = None,
) -> list[dict[str, Any]]:
"""Fetch tasks by status and optional team filter."""
# If multiple statuses, make separate requests and combine results
statuses = status if isinstance(status, list) else [status]
all_tasks: list[dict[str, Any]] = []
for single_status in statuses:
params: dict[str, Any] = {"status": single_status}
if team:
params["team"] = team
try:
resp = await client.get(f"{self._api_url}/tasks", params=params)
if resp.status_code == http_status.HTTP_200_OK:
tasks: list[dict[str, Any]] = resp.json()
all_tasks.extend(tasks)
except Exception as e:
logger.error(
"Fetch tasks error", status=single_status, team=team, error=str(e)
)
return all_tasks
async def _fetch_notifications(
self,
client: httpx.AsyncClient,
notification_type: str,
unacknowledged: bool = True,
) -> list[dict[str, Any]]:
"""Fetch notifications by type."""
params: dict[str, Any] = {
"type": notification_type,
"pending_ack_only": str(unacknowledged).lower(),
}
try:
resp = await client.get(
f"{self._api_url}/notifications",
params=params,
)
if resp.status_code == http_status.HTTP_200_OK:
data = resp.json()
items: list[dict[str, Any]] = data.get("items", [])
return items
except Exception as e:
logger.error(
"Fetch notifications error",
notification_type=notification_type,
error=str(e),
)
return []
# =========================================================================
# SMART ROUTING - TASK CLASSIFICATION
# =========================================================================
# Keywords that indicate strategic/board-level tasks
_BOARD_KEYWORDS = frozenset(
{
"roadmap",
"architecture",
"security",
"budget",
"hiring",
"strategy",
"vision",
"milestone",
"release",
"launch",
}
)
# Keywords that indicate PM coordination is needed
_PM_KEYWORDS = frozenset(
{
"coordinate",
"integration",
"cross-team",
"sync",
"planning",
"milestone",
"dependencies",
"review",
}
)
def _has_board_keywords(self, text: str) -> bool:
"""Check if text contains board-level keywords."""
return any(kw in text for kw in self._BOARD_KEYWORDS)
def _has_pm_keywords(self, text: str) -> bool:
"""Check if text contains PM coordination keywords."""
return any(kw in text for kw in self._PM_KEYWORDS)
# Direct team-to-routing mappings (explicit assignments bypass keyword analysis)
_TEAM_ROUTING_MAP: ClassVar[dict[str, str]] = {
"main_pm": "main_pm",
"board": "board",
"marketing": "marketing",
}
def _classify_task_routing(self, task: dict[str, Any]) -> str:
"""
Classify a task for routing based on team, complexity, and keywords.
Returns one of: "board", "main_pm", "cell_pm", "dev", "marketing"
"""
team = task.get("team")
# Explicit team assignment takes precedence
if team in self._TEAM_ROUTING_MAP:
return self._TEAM_ROUTING_MAP[team]
# For cell teams, use keyword/complexity analysis
title = (task.get("title") or "").lower()
description = (task.get("description") or "").lower()
text = f"{title} {description}"
complexity = task.get("estimated_complexity", "medium").lower()
# Board-level keywords → Board
if self._has_board_keywords(text):
return "board"
# High complexity or cross-team → Main PM
if complexity in ("high", "critical") or not team or team == "all":
return "main_pm"
# PM keywords or medium complexity → Cell PM
if self._has_pm_keywords(text) or complexity == "medium":
return "cell_pm"
# Low complexity, single team → Direct to dev
return "dev"
# Team to PM mapping for routing
_TEAM_PM_MAP: ClassVar[dict[str, str]] = {
"backend": "be-pm",
"frontend": "fe-pm",
"ux_ui": "ux-pm",
}
def _get_routing_target(self, routing: str, task: dict[str, Any]) -> str | None:
"""
Resolve a routing decision to a specific agent slug.
Args:
routing: One of "board", "main_pm", "cell_pm", "dev", "marketing"
task: The task being routed
Returns:
Agent slug (e.g., "main-pm", "be-pm", "be-dev-1") or None
"""
team = task.get("team")
# Static routing targets
static_targets = {
"board": "product-owner",
"main_pm": "main-pm",
"marketing": "head-marketing",
}
if routing in static_targets:
return static_targets[routing]
# Cell PM routing - requires team lookup
if routing == "cell_pm":
return self._TEAM_PM_MAP.get(team, "main-pm") if team else "main-pm"
# Dev routing - requires agent selection
if routing == "dev" and team:
return self._select_agent_for_cell(team, "dev")
return None
def _build_pm_triage_prompt(self, task: dict[str, Any]) -> str:
"""Build prompt for PM to triage and delegate a task."""
task_id = task.get("id", "unknown")
title = task.get("title", "Untitled")
complexity = task.get("complexity", "medium")
team = task.get("team", "unknown")
# Build team-specific info
channel = f"{team}-cell" if team != "ux_ui" else "uxui-cell"
dev_map = {
"backend": ("be-dev-1", "be-dev-2"),
"frontend": ("fe-dev-1", "fe-dev-2"),
"ux_ui": ("ux-dev",),
}
devs = dev_map.get(team, ("be-dev-1",))
primary_dev = devs[0]
dev_options = " or ".join(devs)
return f"""You are the PM for {team} team. This task is assigned to YOU.
TASK: {task_id}
TITLE: {title}
COMPLEXITY: {complexity}
TEAM: {team}
YOUR JOB: Break down this task, create subtasks, and delegate to developers.
You do NOT code. You coordinate and assign. Developers do the actual work.
== IMPORTANT: PLAN vs SUBTASKS ==
These are TWO DIFFERENT THINGS:
1. PLAN = Your PM approach (HOW to do the task)
- Created with roboco_task_plan()
- Just a checklist/strategy attached to the task
- NOT work items
2. SUBTASKS = Real child tasks (WHAT to do)
- Created with roboco_task_create(parent_task_id=...)
- Actual tasks in the database that devs claim and work on
- Parent task DEPENDS on these completing
For any non-trivial task, you MUST create BOTH:
- A plan (your approach)
- Subtasks (the actual work items for devs)
== TASK LIFECYCLE ==
1. You create subtasks with parent_task_id
2. Devs work on subtasks, complete them
3. When ALL subtasks are done → You get respawned
4. You close the parent task
== PM WORKFLOW ==
1. GET TASK DETAILS
roboco_task_get("{task_id}")
Read: description, acceptance criteria, blockers.
2. CREATE YOUR PLAN
roboco_task_plan("{task_id}", ...) with:
- approach: Your PM strategy for this task
- steps: High-level phases (NOT the subtasks)
- risks: Concerns or blockers
3. LOG YOUR DECISION
roboco_journal_decision(data) with:
- title: "PM triage: {{short title}}"
- context, options, chosen, rationale, task_id
4. CREATE SUBTASKS (for medium/complex tasks)
For each piece of work, create a REAL subtask:
roboco_task_create(
title="Specific subtask title",
description="What the dev needs to do",
team="{team}",
acceptance_criteria=["criterion 1", "criterion 2"],
parent_task_id="{task_id}", # REQUIRED - links to parent
assigned_to="{primary_dev}" # Assign to a dev
)
Create 2-5 subtasks that cover all the work.
Available developers: {dev_options}
5. START PARENT TASK
roboco_task_start("{task_id}")
This puts the parent in "in_progress" while subtasks are worked on.
6. NOTIFY TEAM
roboco_message_send(data) to "{channel}":
- content: Task overview, subtasks created, who's assigned
- message_type: "action"
7. FINISH
roboco_agent_idle()
== FOR TRIVIAL TASKS ONLY ==
If task is truly trivial (single file change, obvious fix):
- Skip subtasks, just assign directly:
roboco_task_assign("{task_id}", "{primary_dev}")
- Do NOT call roboco_task_start (dev will do it)
== CRITICAL RULES ==
- NEVER keep tasks for yourself - you delegate, devs execute
- Subtasks MUST have parent_task_id="{task_id}"
- Subtasks MUST have assigned_to (a dev slug like "{primary_dev}")
- When in doubt, create subtasks - it's better to over-structure
Start now: roboco_task_get("{task_id}")
"""
# =========================================================================
# SMART DISPATCHER - MAIN LOOP
# =========================================================================
async def _dispatcher_loop(self) -> None:
"""
Main dispatcher loop - periodically checks for work and spawns agents.
This is the BRAIN of the orchestrator. It:
1. Queries for tasks needing work (pending, awaiting_qa, etc.)
2. Queries for events needing attention (blockers, escalations)
3. Spawns appropriate agents with task assignments
"""
while self._running:
try:
await asyncio.sleep(self.dispatcher_interval)
await self._dispatch_all_work()
except asyncio.CancelledError:
break
except Exception as e:
logger.error("Dispatcher loop error", error=str(e))
async def _dispatch_all_work(self) -> None:
"""Run all dispatchers to check for and assign work."""
# Orchestrator uses SYSTEM role for internal API calls
# Using a well-known UUID for the orchestrator identity
headers = {
"X-Agent-ID": "00000000-0000-0000-0000-000000000000",
"X-Agent-Role": "system",
}
async with httpx.AsyncClient(timeout=30.0, headers=headers) as client:
# PM triage first - routes new tasks to appropriate level
await self._dispatch_pm_work(client)
# PM closure - check parent tasks ready to close
await self._dispatch_pm_closure_work(client)
# Task-based dispatchers (check task statuses)
# Dev work only picks up pre-assigned tasks now
await self._dispatch_dev_work(client)
await self._dispatch_qa_work(client)
await self._dispatch_doc_work(client)
await self._dispatch_pm_review_work(client)
await self._dispatch_marketing_work(client)
# Event-based dispatchers (check blockers, notifications)
await self._dispatch_blocker_work(client)
await self._dispatch_escalation_work(client)
await self._dispatch_approval_work(client)
# Scheduled dispatchers
await self._dispatch_audit_work(client)
# =========================================================================
# SMART DISPATCHER - TASK-BASED DISPATCHERS
# =========================================================================
async def _dispatch_pm_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch PM triage work - routes new tasks to appropriate level.
This is the FIRST dispatcher called - it classifies unassigned tasks
and routes them to Board, Main PM, Cell PM, or directly to devs.
Monitors: pending tasks with no assigned_to
Spawns: product-owner, main-pm, be-pm, fe-pm, ux-pm (or devs for simple)
"""
# Get pending tasks that haven't been assigned yet
tasks = await self._fetch_tasks(client, "pending")
for task in tasks:
# Skip already assigned tasks
if task.get("assigned_to"):
continue
# Classify the task
routing = self._classify_task_routing(task)
agent_id = self._get_routing_target(routing, task)
if not agent_id:
logger.warning(
"No routing target found",
task_id=task.get("id"),
routing=routing,
)
continue
logger.info(
"Routing task",
task_id=task.get("id"),
routing=routing,
agent_id=agent_id,
)
# If target agent is already active, claim for them
if self._is_agent_active(agent_id):
await self._claim_task_for_agent(client, task["id"], agent_id)
continue
# Claim and spawn with appropriate prompt
if await self._claim_task_for_agent(client, task["id"], agent_id):
# Use PM triage prompt for PMs, dev prompt for devs
if routing == "dev":
prompt = self._build_dev_prompt(task)
else:
prompt = self._build_pm_triage_prompt(task)
await self.spawn_agent(
agent_id=agent_id,
task_id=task["id"],
initial_prompt=prompt,
)
async def _dispatch_pm_closure_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch PM closure work - check parent tasks ready to close.
When all subtasks of a parent task are completed, spawn the PM
to review and close the parent task.
Monitors: tasks with completed subtasks but parent still open
Spawns: be-pm, fe-pm, ux-pm (based on parent team)
"""
# Find tasks that have subtasks (check for children)
# We look for claimed/in_progress tasks that might have children
parent_statuses = ["claimed", "in_progress"]
for status in parent_statuses:
tasks = await self._fetch_tasks(client, status)
for task in tasks:
task_id = task.get("id")
if not task_id:
continue
# Check if this task has subtasks
subtasks = await self._fetch_subtasks(client, task_id)
if not subtasks:
continue # Not a parent task
# Check if all subtasks are completed
all_completed = all(st.get("status") == "completed" for st in subtasks)
if not all_completed:
continue # Not ready for closure
# Parent has all subtasks completed - spawn PM to close
team = task.get("team", "backend")
pm_id = self._TEAM_PM_MAP.get(team, "be-pm")
if self._is_agent_active(pm_id):
continue # PM already working
logger.info(
"Parent task ready for closure",
task_id=task_id,
subtasks_count=len(subtasks),
pm_id=pm_id,
)
prompt = self._build_pm_closure_prompt(task, subtasks)
await self.spawn_agent(
agent_id=pm_id,
task_id=task_id,
initial_prompt=prompt,
)
async def _fetch_subtasks(
self, client: httpx.AsyncClient, parent_id: str
) -> list[dict[str, Any]]:
"""Fetch subtasks for a parent task."""
try:
resp = await client.get(
f"{self._api_url}/tasks",
params={"parent_task_id": parent_id},
)
if resp.status_code == http_status.HTTP_200_OK:
data = resp.json()
tasks = data.get("tasks", data) if isinstance(data, dict) else data
return list(tasks) if tasks else []
except Exception as e:
logger.warning(
"Failed to fetch subtasks", parent_id=parent_id, error=str(e)
)
return []
def _build_pm_closure_prompt(
self, task: dict[str, Any], subtasks: list[dict[str, Any]]
) -> str:
"""Build prompt for PM to review and close a parent task."""
task_id = task.get("id", "unknown")
title = task.get("title", "Untitled")
team = task.get("team", "unknown")
subtask_summary = "\n".join(
f" - {st.get('title', 'Untitled')} ({st.get('status', 'unknown')})"
for st in subtasks
)
channel = f"{team}-cell" if team != "ux_ui" else "uxui-cell"
return f"""You are reviewing a parent task for closure.
TASK: {task_id}
TITLE: {title}
TEAM: {team}
ALL SUBTASKS COMPLETED:
{subtask_summary}
== YOUR PM CLOSURE WORKFLOW ==
1. REVIEW
Call roboco_task_get("{task_id}") to review the parent task.
Check: Were all acceptance criteria met by the subtasks?
2. ASSESS SUBTASKS
Review each subtask's completion notes and outcomes.
Were there any issues, learnings, or concerns?
3. JOURNAL (log closure decision)
Call roboco_journal_decision(data) with:
- title: "Task closure: {title}"
- context: Summary of what was accomplished
- options: Close vs Needs refinement
- chosen: Your decision
- rationale: Why
- task_id: "{task_id}"
4. COMMUNICATE
Call roboco_message_send(data) to #{channel}:
- Announce task completion or any follow-up needed
5. CLOSE OR REFINE
- If all criteria met: Call roboco_task_complete("{task_id}")
- If needs more work: Create new subtasks with parent_task_id
6. FINISH
Call roboco_agent_idle()
Begin with step 1: roboco_task_get("{task_id}")
"""
async def _dispatch_dev_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch development work to developers.
NOTE: This now only handles PRE-ASSIGNED tasks (assigned by PM) and
needs_revision tasks. New unassigned pending tasks are handled by
_dispatch_pm_work() which routes them through the PM hierarchy.
Monitors: assigned pending tasks, needs_revision tasks
Spawns: be-dev-1, be-dev-2, fe-dev-1, fe-dev-2, ux-dev
"""
# Get tasks needing dev attention
tasks = await self._fetch_tasks(client, ["pending", "needs_revision"])
for task in tasks:
team = task.get("team")
if team not in ["backend", "frontend", "ux_ui"]:
continue
assigned_to = task.get("assigned_to")
# Resolve UUID to slug for agent operations
agent_slug = self._resolve_agent_slug(assigned_to) if assigned_to else None
# For needs_revision, spawn the assigned dev to fix
if task.get("status") == "needs_revision" and agent_slug:
if not self._is_agent_active(agent_slug):
await self.spawn_agent(
agent_id=agent_slug,
task_id=task["id"],
initial_prompt=self._build_dev_prompt(task),
)
continue
# For pending tasks that ARE already assigned (by PM),
# spawn the assigned dev if not active
if agent_slug and not self._is_agent_active(agent_slug):
await self.spawn_agent(
agent_id=agent_slug,
task_id=task["id"],
initial_prompt=self._build_dev_prompt(task),
)
async def _dispatch_qa_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch QA work to QA agents.
Monitors: awaiting_qa tasks
Spawns: be-qa, fe-qa, ux-qa
"""
tasks = await self._fetch_tasks(client, "awaiting_qa")
for task in tasks:
team = task.get("team")
if team not in ["backend", "frontend", "ux_ui"]:
continue
agent_id = self._select_agent_for_cell(team, "qa")
if not agent_id:
continue
if self._is_agent_active(agent_id):
# QA already running, they'll pick up on scan
continue
# Spawn QA agent with task assignment
await self.spawn_agent(
agent_id=agent_id,
task_id=task["id"],
initial_prompt=self._build_qa_prompt(task),
)
# Only spawn one QA at a time per cell
break
async def _dispatch_doc_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch documentation work to documenters.
Monitors: awaiting_documentation tasks
Spawns: be-doc, fe-doc, ux-doc
"""
tasks = await self._fetch_tasks(client, "awaiting_documentation")
for task in tasks:
team = task.get("team")
if team not in ["backend", "frontend", "ux_ui"]:
continue
agent_id = self._select_agent_for_cell(team, "doc")
if not agent_id:
continue
if self._is_agent_active(agent_id):
continue
await self.spawn_agent(
agent_id=agent_id,
task_id=task["id"],
initial_prompt=self._build_doc_prompt(task),
)
break
async def _dispatch_pm_review_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch PM review work to cell PMs.
Monitors: awaiting_pm_review tasks
Spawns: be-pm, fe-pm, ux-pm
"""
tasks = await self._fetch_tasks(client, "awaiting_pm_review")
for task in tasks:
team = task.get("team")
if team not in ["backend", "frontend", "ux_ui"]:
continue
pm_id = self._TEAM_PM_MAP.get(team, "be-pm")
if self._is_agent_active(pm_id):
continue
await self.spawn_agent(
agent_id=pm_id,
task_id=task["id"],
initial_prompt=self._build_pm_review_prompt(task),
)
break
async def _dispatch_marketing_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch marketing work to head-marketing.
Monitors: pending tasks with team=marketing
Spawns: head-marketing
"""
tasks = await self._fetch_tasks(client, "pending", team="marketing")
for task in tasks:
# Skip already claimed/assigned tasks
if task.get("assigned_to"):
continue
if self._is_agent_active("head-marketing"):
# Already running, they'll pick up on scan
continue
await self.spawn_agent(
agent_id="head-marketing",
task_id=task["id"],
initial_prompt=self._build_marketing_prompt(task),
)
break
# =========================================================================
# SMART DISPATCHER - EVENT-BASED DISPATCHERS
# =========================================================================
async def _dispatch_blocker_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch blocker resolution to Cell PMs.
Monitors: blocked tasks
Spawns: be-pm, fe-pm, ux-pm
"""
tasks = await self._fetch_tasks(client, "blocked")
for task in tasks:
team = task.get("team")
if team not in ["backend", "frontend", "ux_ui"]:
continue
agent_id = self._select_agent_for_cell(team, "pm")
if not agent_id:
continue
if self._is_agent_active(agent_id):
continue
await self.spawn_agent(
agent_id=agent_id,
task_id=task["id"],
initial_prompt=self._build_pm_blocker_prompt(task),
)
break
async def _dispatch_escalation_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch escalations to appropriate managers.
Monitors: escalation notifications (unacknowledged)
Spawns: be-pm, fe-pm, ux-pm, main-pm, product-owner, head-marketing
"""
notifications = await self._fetch_notifications(client, "escalation")
for notif in notifications:
targets = notif.get("to_agents", [])
for agent_id in targets:
valid_targets = [
"be-pm",
"fe-pm",
"ux-pm",
"main-pm",
"product-owner",
"head-marketing",
]
if agent_id not in valid_targets:
continue
if self._is_agent_active(agent_id):
continue
await self.spawn_agent(
agent_id=agent_id,
initial_prompt=self._build_escalation_prompt(notif),
)
break
async def _dispatch_approval_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch approval requests to approvers.
Monitors: approval notifications (unacknowledged)
Spawns: product-owner, head-marketing, main-pm
"""
notifications = await self._fetch_notifications(client, "approval")
for notif in notifications:
targets = notif.get("to_agents", [])
for agent_id in targets:
if agent_id not in ["product-owner", "head-marketing", "main-pm"]:
continue
if self._is_agent_active(agent_id):
continue
await self.spawn_agent(
agent_id=agent_id,
initial_prompt=self._build_approval_prompt(notif),
)
break
async def _dispatch_audit_work(self, client: httpx.AsyncClient) -> None:
"""
Dispatch audit work to the auditor.
Monitors: quality alert notifications
Spawns: auditor
Note: Periodic scheduled audits can be added here in the future.
"""
alerts = await self._fetch_notifications(client, "alert")
for alert in alerts:
targets = alert.get("to_agents", [])
if "auditor" in targets and not self._is_agent_active("auditor"):
await self.spawn_agent(
agent_id="auditor",
initial_prompt=self._build_audit_prompt(alert),
)
return
# TODO: Add scheduled periodic audits
# Check last audit time, spawn if overdue
# =========================================================================
# SMART DISPATCHER - PROMPT BUILDERS
# =========================================================================
def _build_dev_prompt(self, task: dict[str, Any]) -> str:
"""Build initial prompt for a developer with an assigned task."""
task_id = task.get("id", "unknown")
title = task.get("title", "Untitled")
status = task.get("status", "unknown")
team = task.get("team", "unknown")
return f"""You have been assigned a development task.
TASK ID: {task_id}
TITLE: {title}
STATUS: {status}
TEAM: {team}
This task is already CLAIMED for you. Begin work immediately:
1. Call roboco_task_get("{task_id}") for full details and acceptance criteria
2. Follow the workflow: UNDERSTAND → PLAN → EXECUTE → VERIFY → SUBMIT QA
3. When task is submitted for QA, call roboco_task_scan() to check for more work
4. If more work is assigned to you, continue working
5. If no more work, call roboco_agent_idle() to shutdown gracefully
Do NOT scan for work first - your task is already assigned. Begin now.
"""
def _build_qa_prompt(self, task: dict[str, Any]) -> str:
"""Build initial prompt for a QA agent."""
task_id = task.get("id", "unknown")
title = task.get("title", "Untitled")
assigned_to = task.get("assigned_to", "unknown")
team = task.get("team", "unknown")
return f"""A task is ready for QA review.
TASK ID: {task_id}
TITLE: {title}
DEVELOPER: {assigned_to}
TEAM: {team}
Begin QA review:
1. Call roboco_task_get("{task_id}") for full details and acceptance criteria
2. Review the implementation against ALL acceptance criteria
3. Test the changes thoroughly
4. Call roboco_task_qa_pass() with notes if approved
OR roboco_task_qa_fail() with specific issues if rejected
5. Call roboco_task_scan() to check for more QA work
6. If no more work, call roboco_agent_idle() to shutdown gracefully
"""
def _build_doc_prompt(self, task: dict[str, Any]) -> str:
"""Build initial prompt for a documenter."""
task_id = task.get("id", "unknown")
title = task.get("title", "Untitled")
team = task.get("team", "unknown")
return f"""A task is ready for documentation.
TASK ID: {task_id}
TITLE: {title}
TEAM: {team}
Begin documentation:
1. Call roboco_task_get("{task_id}") for full details and dev handoff notes
2. Create or update documentation based on what was implemented
3. Ensure code comments, README updates, API docs as needed
4. Call roboco_task_docs_complete("{task_id}") when documentation is done
5. Call roboco_task_scan() to check for more documentation work
6. If no more work, call roboco_agent_idle() to shutdown gracefully
"""
def _build_pm_review_prompt(self, task: dict[str, Any]) -> str:
"""Build initial prompt for PM to review and complete a task."""
task_id = task.get("id", "unknown")
title = task.get("title", "Untitled")
team = task.get("team", "unknown")
return f"""A task is awaiting your PM review for final completion.
TASK ID: {task_id}
TITLE: {title}
TEAM: {team}
This task has passed QA and documentation. Review and complete:
1. Call roboco_task_get("{task_id}") to review the task details
2. Verify dev_notes, QA notes, and documentation are satisfactory
3. If this task has subtasks, verify all subtasks are completed
4. Call roboco_task_complete("{task_id}") to finalize the task
5. Call roboco_task_scan() to check for more tasks needing review
6. If no more work, call roboco_agent_idle() to shutdown gracefully
"""
def _build_marketing_prompt(self, task: dict[str, Any]) -> str:
"""Build initial prompt for head-marketing with a marketing task."""
task_id = task.get("id", "unknown")
title = task.get("title", "Untitled")
description = task.get("description", "No description")
return f"""You have been assigned a marketing task.
TASK ID: {task_id}
TITLE: {title}
DESCRIPTION: {description}
Begin work:
1. Call roboco_task_get("{task_id}") for full details and acceptance criteria
2. Execute the marketing task (content, campaigns, research, etc.)
3. Coordinate with Product Owner or Main PM if needed
4. Call roboco_task_complete("{task_id}") when done
5. Call roboco_task_scan() to check for more marketing work
6. If no more work, call roboco_agent_idle() to shutdown gracefully
"""
def _build_pm_blocker_prompt(self, task: dict[str, Any]) -> str:
"""Build initial prompt for a Cell PM handling a blocker."""
task_id = task.get("id", "unknown")
title = task.get("title", "Untitled")
assigned_to = task.get("assigned_to", "unknown")
blocker = task.get("blocker", {})
reason = blocker.get("reason", "Unknown")
what_needed = blocker.get("what_needed", "Unknown")
return f"""A task in your cell is BLOCKED and needs your attention.
TASK ID: {task_id}
TITLE: {title}
ASSIGNED TO: {assigned_to}
BLOCKER REASON: {reason}
WHAT'S NEEDED: {what_needed}
Your job:
1. Understand the blocker by reviewing task details
2. Communicate with the blocked developer if needed
3. Resolve the blocker (coordinate resources, make decisions, escalate if needed)
4. Once resolved, the developer can call roboco_task_unblock()
5. Call roboco_task_scan() to check for other blocked tasks in your cell
6. If no more blockers, call roboco_agent_idle() to shutdown gracefully
"""
def _build_escalation_prompt(self, notification: dict[str, Any]) -> str:
"""Build initial prompt for handling an escalation."""
notif_id = notification.get("id", "unknown")
from_agent = notification.get("from_agent", "unknown")
subject = notification.get("subject", "No subject")
priority = notification.get("priority", "normal")
body = notification.get("body", "No details provided")
return f"""You have received an ESCALATION that requires your attention.
FROM: {from_agent}
SUBJECT: {subject}
PRIORITY: {priority}
DETAILS:
{body}
Your job:
1. Acknowledge the notification with roboco_notify_ack("{notif_id}")
2. Assess the escalation and determine action needed
3. Communicate decisions via appropriate channels
4. If this requires further escalation, use roboco_escalate()
5. When resolved, call roboco_task_scan() for other work
6. If no more work, call roboco_agent_idle() to shutdown gracefully
"""
def _build_approval_prompt(self, notification: dict[str, Any]) -> str:
"""Build initial prompt for handling an approval request."""
notif_id = notification.get("id", "unknown")
from_agent = notification.get("from_agent", "unknown")
subject = notification.get("subject", "No subject")
related_task_id = notification.get("related_task_id", "None")
body = notification.get("body", "No details provided")
return f"""You have received an APPROVAL REQUEST.
FROM: {from_agent}
SUBJECT: {subject}
RELATED TASK: {related_task_id}
REQUEST:
{body}
Your job:
1. Review the approval request carefully
2. If related to a task, call roboco_task_get() for context
3. Make your decision and communicate it
4. Acknowledge with roboco_notify_ack("{notif_id}")
5. Call roboco_task_scan() for other work
6. If no more work, call roboco_agent_idle() to shutdown gracefully
"""
def _build_audit_prompt(self, alert: dict[str, Any] | None = None) -> str:
"""Build initial prompt for the auditor."""
if alert:
subject = alert.get("subject", "Quality issue detected")
body = alert.get("body", "Review system quality metrics")
return f"""QUALITY ALERT triggered your attention.
ALERT: {subject}
DETAILS: {body}
Your job:
1. Investigate the quality issue
2. Review relevant channels and task history (you have read access to all)
3. Compile your findings
4. Report to CEO via appropriate channel
5. Call roboco_agent_idle() when complete
"""
return """Periodic AUDIT requested.
Your job:
1. Review recent activity across all cells
2. Check quality metrics (QA pass/fail rates, blocker frequency, etc.)
3. Identify any concerns or patterns
4. Compile audit report for CEO
5. Call roboco_agent_idle() when complete
"""
Reference in New Issue View Git Blame Copy Permalink