From c621710ae6df99637c3c6269a056d07ff9b6de0d Mon Sep 17 00:00:00 2001 From: Renn F Date: Tue, 6 Jan 2026 00:59:09 +0100 Subject: [PATCH] A2A wiring up --- CLAUDE.md | 6 +- README.md | 4 +- agents/blueprints/backend/be-dev.md | 14 +- agents/blueprints/backend/be-documenter.md | 5 + agents/blueprints/backend/be-pm.md | 5 + agents/blueprints/backend/be-qa.md | 5 + agents/blueprints/board/auditor.md | 5 + agents/blueprints/board/head-marketing.md | 5 + agents/blueprints/board/main-pm.md | 5 + agents/blueprints/board/product-owner.md | 5 + agents/blueprints/frontend/fe-dev.md | 5 + agents/blueprints/frontend/fe-documenter.md | 5 + agents/blueprints/frontend/fe-pm.md | 5 + agents/blueprints/frontend/fe-qa.md | 5 + agents/blueprints/ux_ui/ux-dev.md | 5 + agents/blueprints/ux_ui/ux-documenter.md | 5 + agents/blueprints/ux_ui/ux-pm.md | 5 + agents/blueprints/ux_ui/ux-qa.md | 5 + agents/prompts/roles/cell_pm.md | 9 +- agents/prompts/roles/developer.md | 9 +- agents/prompts/roles/qa.md | 9 +- docker-compose.yml | 10 +- docker/agent-base.Dockerfile | 14 +- docker/scripts/a2a-check-hook.sh | 29 + docker/scripts/sdk-startup-hook.sh | 32 + docs/rag/architecture/config-reference.md | 2 +- docs/rag/architecture/tool-permissions.md | 65 ++ docs/rag/roles/auditor.md | 19 + docs/rag/roles/cell-pm.md | 19 + docs/rag/roles/ceo.md | 11 + docs/rag/roles/developer.md | 35 +- docs/rag/roles/documenter.md | 19 + docs/rag/roles/head-marketing.md | 13 +- docs/rag/roles/main-pm.md | 20 + docs/rag/roles/product-owner.md | 14 +- docs/rag/roles/qa.md | 21 + docs/rag/tools/a2a-tools.md | 82 +++ docs/rag/troubleshooting/blocked-tools.md | 56 ++ docs/rag/troubleshooting/common-issues.md | 21 + docs/rag/workflows/a2a-collaboration.md | 74 +++ docs/rag/workflows/task-planning.md | 49 ++ roboco/agent_sdk/__init__.py | 10 + roboco/agent_sdk/models.py | 68 +++ roboco/agent_sdk/server.py | 260 ++++++++ roboco/agents_config.py | 122 ++++ roboco/api/routes/a2a.py | 67 ++- roboco/api/routes/notifications.py | 114 ++++ roboco/config.py | 2 +- roboco/events/__init__.py | 2 - roboco/events/handlers.py | 71 +-- roboco/mcp/a2a_server.py | 382 +++++++----- roboco/mcp/tasks/handlers/_helpers.py | 30 +- roboco/mcp/tasks/handlers/blocking.py | 23 +- roboco/mcp/tasks/handlers/lifecycle.py | 22 + roboco/mcp/tasks/handlers/review.py | 36 +- roboco/mcp/tasks/utils.py | 10 + roboco/models/a2a.py | 4 + roboco/models/base.py | 1 + roboco/models/events.py | 10 + roboco/runtime/orchestrator.py | 565 ++++++++++++++---- roboco/services/a2a.py | 197 +++--- roboco/services/notification.py | 49 ++ roboco/services/optimal_brain/indexes/base.py | 2 +- 63 files changed, 2327 insertions(+), 451 deletions(-) create mode 100644 docker/scripts/a2a-check-hook.sh create mode 100644 docker/scripts/sdk-startup-hook.sh create mode 100644 docs/rag/architecture/tool-permissions.md create mode 100644 docs/rag/tools/a2a-tools.md create mode 100644 docs/rag/troubleshooting/blocked-tools.md create mode 100644 docs/rag/workflows/a2a-collaboration.md create mode 100644 docs/rag/workflows/task-planning.md create mode 100644 roboco/agent_sdk/__init__.py create mode 100644 roboco/agent_sdk/models.py create mode 100644 roboco/agent_sdk/server.py diff --git a/CLAUDE.md b/CLAUDE.md index c4b119ab..79213249 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -75,7 +75,7 @@ pnpm test | Cache/Queue | Redis | | Container Runtime | Docker + Docker Compose | | Cloud LLM | Claude API (claude-opus-4-5-20251101) | -| Local LLM | Ollama (glm-4.6:cloud for HyDE/RAG) | +| Local LLM | Ollama (glm-4.7:cloud for HyDE/RAG) | | Embeddings | embeddinggemma:300m (768 dim) | | Frontend | React / Next.js (future) | @@ -332,7 +332,7 @@ ROBOCO_RAG_USE_HYBRID_SEARCH=true # AI/LLM ROBOCO_DEFAULT_EMBEDDING_MODEL=embeddinggemma:300m -ROBOCO_LOCAL_LLM_MODEL=glm-4.6:cloud +ROBOCO_LOCAL_LLM_MODEL=glm-4.7:cloud ROBOCO_LOCAL_LLM_BASE_URL=http://roboco-ollama:11434/v1 ROBOCO_OLLAMA_BASE_URL=http://roboco-ollama:11434 ``` @@ -359,7 +359,7 @@ The startup order is critical due to dependencies: postgres ──┐ redis ─────┼──> ollama ──> ollama-init ──> orchestrator │ │ │ - │ │ └── Pulls embeddinggemma:300m, glm-4.6:cloud + │ │ └── Pulls embeddinggemma:300m, glm-4.7:cloud │ └── Healthcheck: ollama list └── Healthcheck: pg_isready, redis-cli ping ``` diff --git a/README.md b/README.md index a6da37ef..850190d6 100644 --- a/README.md +++ b/README.md @@ -91,7 +91,7 @@ ROBOCO_WORKSPACE_AUTO_CLONE=true # RAG/LLM ROBOCO_LOCAL_LLM_BASE_URL=http://roboco-ollama:11434/v1 -ROBOCO_LOCAL_LLM_MODEL=glm-4.6:cloud +ROBOCO_LOCAL_LLM_MODEL=glm-4.7:cloud ``` ## Multi-Agent Workspace Structure @@ -177,7 +177,7 @@ uv run mypy roboco/ | Cache/Queue | Redis | | RAG Library | piragi | | Embeddings | embeddinggemma:300m (sentence-transformers) | -| Local LLM | Ollama (glm-4.6:cloud) | +| Local LLM | Ollama (glm-4.7:cloud) | | Cloud LLM | Claude API (Anthropic) | | Package Manager | uv | diff --git a/agents/blueprints/backend/be-dev.md b/agents/blueprints/backend/be-dev.md index ddbdf807..75862048 100644 --- a/agents/blueprints/backend/be-dev.md +++ b/agents/blueprints/backend/be-dev.md @@ -73,6 +73,11 @@ You interact with RoboCo systems through MCP tools. These are your primary inter - `roboco_notify_get(notification_id)` - Read a notification - `roboco_notify_ack(notification_id)` - Acknowledge notification +**A2A (Agent-to-Agent):** +- `roboco_agent_discover(role, team, skill)` - Find agents +- `roboco_agent_request(target, skill, message, task_id)` - Send message +- `roboco_a2a_check()` - Check inbox (auto-notified via hook) + **Agent Lifecycle:** - `roboco_agent_idle()` - Signal no work available (terminates gracefully) @@ -455,12 +460,9 @@ tools: - roboco_session_history_for_task # Get discussion history for your task - roboco_report_blocker - # Claude Code Built-in - - bash (for running commands) - - read/write/edit files - - git (commit, branch, push) - - pytest, ruff, mypy - - web fetch (for docs lookup) + # Git Operations (via roboco MCP tools) + - roboco_git_status, roboco_git_log, roboco_git_diff + - roboco_git_commit, roboco_git_push, roboco_git_create_pr ``` ## Permissions diff --git a/agents/blueprints/backend/be-documenter.md b/agents/blueprints/backend/be-documenter.md index f3f32e48..3d4a8fa2 100644 --- a/agents/blueprints/backend/be-documenter.md +++ b/agents/blueprints/backend/be-documenter.md @@ -65,6 +65,11 @@ You are the Backend Documenter at RoboCo, an AI-powered software company. You tr - `roboco_notify_get(notification_id)` - Read a notification - `roboco_notify_ack(notification_id)` - Acknowledge notification +**A2A (Agent-to-Agent):** +- `roboco_agent_discover(role, team, skill)` - Find agents +- `roboco_agent_request(target, skill, message, task_id)` - Send message +- `roboco_a2a_check()` - Check inbox (auto-notified via hook) + **Agent Lifecycle:** - `roboco_agent_idle()` - Signal no work available diff --git a/agents/blueprints/backend/be-pm.md b/agents/blueprints/backend/be-pm.md index dc0ad5c9..e4485679 100644 --- a/agents/blueprints/backend/be-pm.md +++ b/agents/blueprints/backend/be-pm.md @@ -79,6 +79,11 @@ You interact with RoboCo systems through MCP tools: - `roboco_escalate(escalate_to, subject, description)` - Escalate to Main PM (PM only) - `roboco_request_approval(approver, subject, what_needs_approval)` - Request approval (PM only) +**A2A (Agent-to-Agent):** +- `roboco_agent_discover(role, team, skill)` - Find agents +- `roboco_agent_request(target, skill, message, task_id)` - Send message +- `roboco_a2a_check()` - Check inbox (auto-notified via hook) + **Agent Lifecycle:** - `roboco_agent_idle()` - Signal done (terminates gracefully) diff --git a/agents/blueprints/backend/be-qa.md b/agents/blueprints/backend/be-qa.md index 10d6eacd..e143a43a 100644 --- a/agents/blueprints/backend/be-qa.md +++ b/agents/blueprints/backend/be-qa.md @@ -69,6 +69,11 @@ You interact with RoboCo systems through MCP tools: - `roboco_notify_get(notification_id)` - Read a notification - `roboco_notify_ack(notification_id)` - Acknowledge notification +**A2A (Agent-to-Agent):** +- `roboco_agent_discover(role, team, skill)` - Find agents +- `roboco_agent_request(target, skill, message, task_id)` - Send message +- `roboco_a2a_check()` - Check inbox (auto-notified via hook) + **Agent Lifecycle:** - `roboco_agent_idle()` - Signal no work available (terminates gracefully) diff --git a/agents/blueprints/board/auditor.md b/agents/blueprints/board/auditor.md index 6e1fa1f9..ac1d09b7 100644 --- a/agents/blueprints/board/auditor.md +++ b/agents/blueprints/board/auditor.md @@ -67,6 +67,11 @@ You interact with RoboCo systems through MCP tools: - `roboco_notify_ack(notification_id)` - Acknowledge a notification - `roboco_notify_send(data)` - Can notify anyone (emergency use only) +**A2A (Agent-to-Agent):** +- `roboco_agent_discover(role, team, skill)` - Find agents +- `roboco_agent_request(target, skill, message, task_id)` - Send message +- `roboco_a2a_check()` - Check inbox (auto-notified via hook) + **Agent Lifecycle:** - `roboco_agent_idle()` - Signal observation complete (rare - usually always active) diff --git a/agents/blueprints/board/head-marketing.md b/agents/blueprints/board/head-marketing.md index 20a3965d..287546a2 100644 --- a/agents/blueprints/board/head-marketing.md +++ b/agents/blueprints/board/head-marketing.md @@ -62,6 +62,11 @@ You interact with RoboCo systems through MCP tools: - `roboco_message_send(channel, content)` - Post to board channels - `roboco_channel_history(channel_slug, limit?)` - Read channel history +**A2A (Agent-to-Agent):** +- `roboco_agent_discover(role, team, skill)` - Find agents +- `roboco_agent_request(target, skill, message, task_id)` - Send message +- `roboco_a2a_check()` - Check inbox (auto-notified via hook) + **Agent Lifecycle:** - `roboco_agent_idle()` - Signal no work available (terminates gracefully) diff --git a/agents/blueprints/board/main-pm.md b/agents/blueprints/board/main-pm.md index a84147e9..a07069eb 100644 --- a/agents/blueprints/board/main-pm.md +++ b/agents/blueprints/board/main-pm.md @@ -92,6 +92,11 @@ You interact with RoboCo systems through MCP tools: - `roboco_journal_read_team(target_agent, entry_type?, task_id?, limit?)` - Read Cell PM journals - `roboco_journal_scope()` - See which journals you can access +**A2A (Agent-to-Agent):** +- `roboco_agent_discover(role, team, skill)` - Find agents +- `roboco_agent_request(target, skill, message, task_id)` - Send message +- `roboco_a2a_check()` - Check inbox (auto-notified via hook) + **Agent Lifecycle:** - `roboco_agent_idle()` - Signal no work available (terminates gracefully) diff --git a/agents/blueprints/board/product-owner.md b/agents/blueprints/board/product-owner.md index a735a157..be3e147d 100644 --- a/agents/blueprints/board/product-owner.md +++ b/agents/blueprints/board/product-owner.md @@ -64,6 +64,11 @@ You interact with RoboCo systems through MCP tools: - `roboco_message_send(channel, content)` - Post to board channels - `roboco_channel_history(channel_slug, limit?)` - Read channel history +**A2A (Agent-to-Agent):** +- `roboco_agent_discover(role, team, skill)` - Find agents +- `roboco_agent_request(target, skill, message, task_id)` - Send message +- `roboco_a2a_check()` - Check inbox (auto-notified via hook) + **Agent Lifecycle:** - `roboco_agent_idle()` - Signal no work available (terminates gracefully) diff --git a/agents/blueprints/frontend/fe-dev.md b/agents/blueprints/frontend/fe-dev.md index 011e4d40..121ef60c 100644 --- a/agents/blueprints/frontend/fe-dev.md +++ b/agents/blueprints/frontend/fe-dev.md @@ -74,6 +74,11 @@ You interact with RoboCo systems through MCP tools. These are your primary inter - `roboco_notify_get(notification_id)` - Read a notification - `roboco_notify_ack(notification_id)` - Acknowledge notification +**A2A (Agent-to-Agent):** +- `roboco_agent_discover(role, team, skill)` - Find agents +- `roboco_agent_request(target, skill, message, task_id)` - Send message +- `roboco_a2a_check()` - Check inbox (auto-notified via hook) + **Agent Lifecycle:** - `roboco_agent_idle()` - Signal no work available (terminates gracefully) diff --git a/agents/blueprints/frontend/fe-documenter.md b/agents/blueprints/frontend/fe-documenter.md index a5492892..07ba1d96 100644 --- a/agents/blueprints/frontend/fe-documenter.md +++ b/agents/blueprints/frontend/fe-documenter.md @@ -65,6 +65,11 @@ You are the Frontend Documenter at RoboCo, an AI-powered software company. You t - `roboco_notify_get(notification_id)` - Read a notification - `roboco_notify_ack(notification_id)` - Acknowledge notification +**A2A (Agent-to-Agent):** +- `roboco_agent_discover(role, team, skill)` - Find agents +- `roboco_agent_request(target, skill, message, task_id)` - Send message +- `roboco_a2a_check()` - Check inbox (auto-notified via hook) + **Agent Lifecycle:** - `roboco_agent_idle()` - Signal no work available diff --git a/agents/blueprints/frontend/fe-pm.md b/agents/blueprints/frontend/fe-pm.md index 838549e8..6f7795f1 100644 --- a/agents/blueprints/frontend/fe-pm.md +++ b/agents/blueprints/frontend/fe-pm.md @@ -80,6 +80,11 @@ You interact with RoboCo systems through MCP tools: - `roboco_escalate(escalate_to, subject, description)` - Escalate to Main PM (PM only) - `roboco_request_approval(approver, subject, what_needs_approval)` - Request approval (PM only) +**A2A (Agent-to-Agent):** +- `roboco_agent_discover(role, team, skill)` - Find agents +- `roboco_agent_request(target, skill, message, task_id)` - Send message +- `roboco_a2a_check()` - Check inbox (auto-notified via hook) + **Agent Lifecycle:** - `roboco_agent_idle()` - Signal done (terminates gracefully) diff --git a/agents/blueprints/frontend/fe-qa.md b/agents/blueprints/frontend/fe-qa.md index f0208732..c380fc42 100644 --- a/agents/blueprints/frontend/fe-qa.md +++ b/agents/blueprints/frontend/fe-qa.md @@ -64,6 +64,11 @@ You are the Frontend QA Engineer at RoboCo, an AI-powered software company. You - `roboco_notify_get(notification_id)` - Read a notification - `roboco_notify_ack(notification_id)` - Acknowledge notification +**A2A (Agent-to-Agent):** +- `roboco_agent_discover(role, team, skill)` - Find agents +- `roboco_agent_request(target, skill, message, task_id)` - Send message +- `roboco_a2a_check()` - Check inbox (auto-notified via hook) + **Agent Lifecycle:** - `roboco_agent_idle()` - Signal no work available diff --git a/agents/blueprints/ux_ui/ux-dev.md b/agents/blueprints/ux_ui/ux-dev.md index 741d01cc..af82b4e7 100644 --- a/agents/blueprints/ux_ui/ux-dev.md +++ b/agents/blueprints/ux_ui/ux-dev.md @@ -75,6 +75,11 @@ You interact with RoboCo systems through MCP tools. These are your primary inter - `roboco_notify_get(notification_id)` - Read a notification - `roboco_notify_ack(notification_id)` - Acknowledge notification +**A2A (Agent-to-Agent):** +- `roboco_agent_discover(role, team, skill)` - Find agents +- `roboco_agent_request(target, skill, message, task_id)` - Send message +- `roboco_a2a_check()` - Check inbox (auto-notified via hook) + **Agent Lifecycle:** - `roboco_agent_idle()` - Signal no work available (terminates gracefully) diff --git a/agents/blueprints/ux_ui/ux-documenter.md b/agents/blueprints/ux_ui/ux-documenter.md index 443b1b37..fb5326c6 100644 --- a/agents/blueprints/ux_ui/ux-documenter.md +++ b/agents/blueprints/ux_ui/ux-documenter.md @@ -65,6 +65,11 @@ You are the UX/UI Documenter at RoboCo, an AI-powered software company. You main - `roboco_notify_get(notification_id)` - Read a notification - `roboco_notify_ack(notification_id)` - Acknowledge notification +**A2A (Agent-to-Agent):** +- `roboco_agent_discover(role, team, skill)` - Find agents +- `roboco_agent_request(target, skill, message, task_id)` - Send message +- `roboco_a2a_check()` - Check inbox (auto-notified via hook) + **Agent Lifecycle:** - `roboco_agent_idle()` - Signal no work available diff --git a/agents/blueprints/ux_ui/ux-pm.md b/agents/blueprints/ux_ui/ux-pm.md index 8feee291..b85023a6 100644 --- a/agents/blueprints/ux_ui/ux-pm.md +++ b/agents/blueprints/ux_ui/ux-pm.md @@ -80,6 +80,11 @@ You interact with RoboCo systems through MCP tools: - `roboco_escalate(escalate_to, subject, description)` - Escalate to Main PM (PM only) - `roboco_request_approval(approver, subject, what_needs_approval)` - Request approval (PM only) +**A2A (Agent-to-Agent):** +- `roboco_agent_discover(role, team, skill)` - Find agents +- `roboco_agent_request(target, skill, message, task_id)` - Send message +- `roboco_a2a_check()` - Check inbox (auto-notified via hook) + **Agent Lifecycle:** - `roboco_agent_idle()` - Signal done (terminates gracefully) diff --git a/agents/blueprints/ux_ui/ux-qa.md b/agents/blueprints/ux_ui/ux-qa.md index 14bbf8f2..d8f0cf67 100644 --- a/agents/blueprints/ux_ui/ux-qa.md +++ b/agents/blueprints/ux_ui/ux-qa.md @@ -64,6 +64,11 @@ You are the UX/UI QA Engineer at RoboCo, an AI-powered software company. You ens - `roboco_notify_get(notification_id)` - Read a notification - `roboco_notify_ack(notification_id)` - Acknowledge notification +**A2A (Agent-to-Agent):** +- `roboco_agent_discover(role, team, skill)` - Find agents +- `roboco_agent_request(target, skill, message, task_id)` - Send message +- `roboco_a2a_check()` - Check inbox (auto-notified via hook) + **Agent Lifecycle:** - `roboco_agent_idle()` - Signal no work available diff --git a/agents/prompts/roles/cell_pm.md b/agents/prompts/roles/cell_pm.md index 4616fa59..a94ba490 100644 --- a/agents/prompts/roles/cell_pm.md +++ b/agents/prompts/roles/cell_pm.md @@ -143,13 +143,12 @@ When ALL subtasks done: reflect + complete your task. **Agent-to-Agent (A2A) - Cross-Cell Coordination:** - `roboco_agent_discover(role, team, skill)` - Find agents across cells -- `roboco_agent_request(target_agent, skill, message)` - Request cross-cell help -- `roboco_agent_request_status(a2a_task_id)` - Track requests +- `roboco_agent_request(target_agent, skill, message, task_id)` - Send message (task_id required) +- `roboco_a2a_check()` - Check inbox for incoming messages (auto-notified via hook) **A2A for Cell PM:** -- Need frontend input? → `roboco_agent_request("fe-pm", "task_management", "Need to coordinate...")` -- Find cross-cell expertise: `roboco_agent_discover(skill="security_audit")` -- Handle A2A requests from other cells via `roboco_notify_list()` +- Cross-cell coordination: `roboco_agent_request("fe-pm", "coordination", "...", task_id)` +- Find expertise: `roboco_agent_discover(skill="security_audit")` ## NOT Your Tools diff --git a/agents/prompts/roles/developer.md b/agents/prompts/roles/developer.md index 0bae4ed4..3e1caee5 100644 --- a/agents/prompts/roles/developer.md +++ b/agents/prompts/roles/developer.md @@ -139,13 +139,12 @@ If PMs request changes: **Agent-to-Agent (A2A) - Direct Collaboration:** - `roboco_agent_discover(role, team, skill)` - Find agents who can help -- `roboco_agent_request(target_agent, skill, message)` - Request work from another agent -- `roboco_agent_request_status(a2a_task_id)` - Check request progress +- `roboco_agent_request(target_agent, skill, message, task_id)` - Send message (task_id required) +- `roboco_a2a_check()` - Check inbox for incoming messages (auto-notified via hook) **When to use A2A:** -- Need code review? → `roboco_agent_request("be-qa", "code_review", "Please review my PR...")` -- Need docs help? → `roboco_agent_request("be-doc", "documentation", "Need API docs for...")` -- Find who can help? → `roboco_agent_discover(skill="code_review")` +- Code review → `roboco_agent_request("be-qa", "code_review", "Review please", task_id)` +- Docs help → `roboco_agent_request("be-doc", "documentation", "Need API docs", task_id)` ## NOT Your Tools diff --git a/agents/prompts/roles/qa.md b/agents/prompts/roles/qa.md index 604fd22a..e5d07102 100644 --- a/agents/prompts/roles/qa.md +++ b/agents/prompts/roles/qa.md @@ -76,13 +76,12 @@ Use `roboco_journal_reflect()` before decision. REQUIRED. **Agent-to-Agent (A2A) - Direct Collaboration:** - `roboco_agent_discover(role, team, skill)` - Find agents who can help -- `roboco_agent_request(target_agent, skill, message)` - Request work from another agent -- `roboco_agent_request_status(a2a_task_id)` - Check request progress +- `roboco_agent_request(target_agent, skill, message, task_id)` - Send message (task_id required) +- `roboco_a2a_check()` - Check inbox for incoming messages (auto-notified via hook) **A2A for QA:** -- Developers will request `code_review` from you via A2A -- Check for A2A requests with `roboco_notify_list()` (type: MENTION or TASK_ASSIGNMENT) -- You can request dev clarification: `roboco_agent_request("be-dev-1", "technical_research", "Why did you...")` +- Developers send `code_review` requests - check with `roboco_a2a_check()` +- Request clarification: `roboco_agent_request("be-dev-1", "clarification", "Why...", task_id)` ## NOT Your Tools diff --git a/docker-compose.yml b/docker-compose.yml index 614875e0..d96a2121 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -79,14 +79,14 @@ services: status=$$(echo "$$line" | grep -o '"status":"[^"]*"' | cut -d'"' -f4) [ -n "$$status" ] && echo " $$status" done - echo "=== Pulling LLM model (glm-4.6:cloud) ===" - curl -sN http://ollama:11434/api/pull -d '{"name":"glm-4.6:cloud"}' | while read -r line; do + echo "=== Pulling LLM model (glm-4.7:cloud) ===" + curl -sN http://ollama:11434/api/pull -d '{"name":"glm-4.7:cloud"}' | while read -r line; do status=$$(echo "$$line" | grep -o '"status":"[^"]*"' | cut -d'"' -f4) [ -n "$$status" ] && echo " $$status" done echo "=== Verifying models are available ===" curl -sf http://ollama:11434/api/tags | grep -q "embeddinggemma" && echo " embeddinggemma: OK" - curl -sf http://ollama:11434/api/tags | grep -q "glm-4.6" && echo " glm-4.6: OK" + curl -sf http://ollama:11434/api/tags | grep -q "glm-4.7" && echo " glm-4.7: OK" echo "=== All models ready! ===" # ========================================================================== @@ -219,7 +219,7 @@ services: ROBOCO_PORT: 8000 # Ollama (use container name) ROBOCO_LOCAL_LLM_BASE_URL: http://roboco-ollama:11434/v1 - ROBOCO_LOCAL_LLM_MODEL: glm-4.6:cloud + ROBOCO_LOCAL_LLM_MODEL: glm-4.7:cloud ROBOCO_DEFAULT_EMBEDDING_MODEL: embeddinggemma:300m ROBOCO_OLLAMA_BASE_URL: http://roboco-ollama:11434 # Host paths for spawning agent containers (required for Docker-in-Docker) @@ -236,6 +236,8 @@ services: - ${ROBOCO_DATA_DIR:-./data}/mcp-configs:/app/mcp-configs # Generated prompts directory - composed at runtime from layers - ${ROBOCO_DATA_DIR:-./data}/prompts-generated:/app/prompts-generated + # Per-agent Claude settings (generated at spawn time) + - ${ROBOCO_DATA_DIR:-./data}/agent-settings:/app/agent-settings depends_on: postgres: condition: service_healthy diff --git a/docker/agent-base.Dockerfile b/docker/agent-base.Dockerfile index 48efd256..dc56f2bf 100644 --- a/docker/agent-base.Dockerfile +++ b/docker/agent-base.Dockerfile @@ -6,12 +6,13 @@ FROM python:3.13-bookworm -# Install Node.js 22 (required for Claude Code CLI) +# Install Node.js 22 (required for Claude Code CLI) and jq (for hooks) RUN apt-get update && apt-get install -y --no-install-recommends \ curl \ ca-certificates \ git \ gnupg \ + jq \ && curl -fsSL https://deb.nodesource.com/setup_22.x | bash - \ && apt-get install -y nodejs \ && rm -rf /var/lib/apt/lists/* @@ -45,4 +46,15 @@ RUN uv python install 3.13 && uv sync --frozen --python 3.13 # System prompt mounted at /app/system-prompt.md (composed at spawn time from layers) # MCP config generated at runtime +# Copy SDK server scripts and hooks (need to be root for COPY, then fix permissions) +USER root +COPY --chown=agent:agent docker/scripts/sdk-startup-hook.sh /app/scripts/sdk-startup-hook.sh +COPY --chown=agent:agent docker/scripts/a2a-check-hook.sh /app/scripts/a2a-check-hook.sh +RUN chmod +x /app/scripts/*.sh +USER agent + +# Expose SDK server port +EXPOSE 9000 + +# Claude runs directly - SDK server started via SessionStart hook ENTRYPOINT ["claude"] diff --git a/docker/scripts/a2a-check-hook.sh b/docker/scripts/a2a-check-hook.sh new file mode 100644 index 00000000..6396e70b --- /dev/null +++ b/docker/scripts/a2a-check-hook.sh @@ -0,0 +1,29 @@ +#!/bin/bash +# A2A Check Hook +# +# Claude Code hook that runs after each tool call to check for +# incoming A2A messages. Notifies Claude if messages are pending. +# +# This hook is non-blocking and always succeeds to avoid +# interrupting Claude's workflow. + +SDK_URL="${ROBOCO_SDK_URL:-http://localhost:9000}" + +# Check inbox count (non-consuming endpoint) +response=$(curl -sf "$SDK_URL/inbox/count" 2>/dev/null) + +if [ $? -eq 0 ]; then + total=$(echo "$response" | jq -r '.total // 0') + urgent=$(echo "$response" | jq -r '.urgent // 0') + + if [ "$total" -gt 0 ]; then + if [ "$urgent" -gt 0 ]; then + echo "[A2A] URGENT: You have $urgent urgent message(s). Use roboco_a2a_check() to read them." + else + echo "[A2A] You have $total pending message(s). Use roboco_a2a_check() to read them." + fi + fi +fi + +# Always exit 0 - don't block Claude +exit 0 diff --git a/docker/scripts/sdk-startup-hook.sh b/docker/scripts/sdk-startup-hook.sh new file mode 100644 index 00000000..766961d3 --- /dev/null +++ b/docker/scripts/sdk-startup-hook.sh @@ -0,0 +1,32 @@ +#!/bin/bash +# SDK Server Startup Hook +# +# Called by Claude Code on SessionStart to start the SDK server. +# Runs in background, Claude continues immediately. + +SDK_PORT="${ROBOCO_SDK_PORT:-9000}" +AGENT_ID="${ROBOCO_AGENT_ID:-unknown}" +LOG_FILE="/tmp/sdk-server.log" + +# Check if SDK is already running +if curl -sf "http://localhost:${SDK_PORT}/health" >/dev/null 2>&1; then + echo "[SDK] Already running on port ${SDK_PORT}" + exit 0 +fi + +# Start SDK server in background (nohup to survive hook completion) +echo "[SDK] Starting for agent ${AGENT_ID} on port ${SDK_PORT}..." +nohup uv run python -m roboco.agent_sdk.server > "$LOG_FILE" 2>&1 & +SDK_PID=$! + +# Brief wait for startup (non-blocking - don't hold up Claude) +sleep 2 + +# Check if it started +if curl -sf "http://localhost:${SDK_PORT}/health" >/dev/null 2>&1; then + echo "[SDK] Ready (PID: ${SDK_PID})" +else + echo "[SDK] Starting in background (PID: ${SDK_PID}, check ${LOG_FILE} for status)" +fi + +exit 0 diff --git a/docs/rag/architecture/config-reference.md b/docs/rag/architecture/config-reference.md index 822a6f86..b88ac5da 100644 --- a/docs/rag/architecture/config-reference.md +++ b/docs/rag/architecture/config-reference.md @@ -58,7 +58,7 @@ Environment variables for RoboCo (prefix: `ROBOCO_`). | Variable | Default | Description | |----------|---------|-------------| -| `ROBOCO_LOCAL_LLM_MODEL` | `glm-4.6:cloud` | Local LLM for RAG | +| `ROBOCO_LOCAL_LLM_MODEL` | `glm-4.7:cloud` | Local LLM for RAG | | `ROBOCO_LOCAL_LLM_BASE_URL` | `http://roboco-ollama:11434/v1` | OpenAI-compat API | | `ROBOCO_OLLAMA_BASE_URL` | `http://roboco-ollama:11434` | Native Ollama API | diff --git a/docs/rag/architecture/tool-permissions.md b/docs/rag/architecture/tool-permissions.md new file mode 100644 index 00000000..88e5d989 --- /dev/null +++ b/docs/rag/architecture/tool-permissions.md @@ -0,0 +1,65 @@ +# Tool Permissions by Role + +## Overview + +Agents have role-specific tool permissions enforced via Claude Code settings. +Native tools are blocked; use `roboco_*` MCP tools instead. + +## Developer + +**Allowed:** +- `roboco_task_*` - task lifecycle +- `roboco_git_*` - all git operations +- `roboco_test_*` - run tests, lint, format +- `roboco_journal_*` - journaling +- `roboco_kb_*`, `roboco_rag_*` - knowledge base +- `Read(*)` - read any file +- `Write/Edit` - workspace only + +**Blocked:** +- `Bash(git:*)` - use roboco_git_* instead +- `Write/Edit` outside workspace + +**Workspace:** `/data/workspaces/{project}/{team}/{agent-id}/` + +## QA + +**Allowed:** +- `roboco_git_status`, `roboco_git_log`, `roboco_git_diff` - read-only +- `roboco_test_*` - run tests +- `roboco_task_qa_pass`, `roboco_task_qa_fail` +- `Read(*)` - read any file + +**Blocked:** +- `roboco_git_commit`, `roboco_git_push` - QA doesn't write code +- All `Write/Edit` - review only + +## Documenter + +**Allowed:** +- `roboco_docs_*` - documentation tools +- `roboco_git_*` - all git operations +- `Write/Edit` in `/app/docs/**` only + +**Blocked:** +- `Write/Edit` outside docs directory + +## PM (Cell PM, Main PM) + +**Allowed:** +- `roboco_git_*` - all git operations +- `roboco_docs_*` - documentation +- `roboco_task_*` - full task management +- `roboco_notify_send` - send notifications + +**Blocked:** +- `Bash(git:*)` - use roboco_git_* + +## Auditor + +**Allowed:** +- `roboco_git_status`, `roboco_git_log`, `roboco_git_diff` - read-only +- `Read(*)` - read any file + +**Blocked:** +- All write operations - observer role diff --git a/docs/rag/roles/auditor.md b/docs/rag/roles/auditor.md index 0eb1f928..2798fa6e 100644 --- a/docs/rag/roles/auditor.md +++ b/docs/rag/roles/auditor.md @@ -61,6 +61,18 @@ roboco_task_create({ }) ``` +## Tool Restrictions + +**Read-only observer.** Cannot modify anything. + +| Allowed | Blocked | +|---------|---------| +| `roboco_git_status/log/diff` | All `Write/Edit` | +| `Read(*)` | All git write operations | +| `roboco_kb_search` | Native git commands | + +See: `roboco_kb_search("tool permissions")` + ## Key Tools | Tool | Purpose | @@ -74,6 +86,13 @@ roboco_task_create({ The Auditor primarily observes and reports. Direct intervention is NOT the Auditor's role - issues are escalated to CEO for action. +## A2A + +```python +roboco_agent_request("ceo", "escalation", "Found issue...", task_id) +roboco_a2a_check() # Check inbox +``` + ## Escalation Report directly to CEO when: diff --git a/docs/rag/roles/cell-pm.md b/docs/rag/roles/cell-pm.md index 7512cfe7..4ad0e643 100644 --- a/docs/rag/roles/cell-pm.md +++ b/docs/rag/roles/cell-pm.md @@ -99,6 +99,18 @@ roboco_notify_list() roboco_journal_read_team("be-dev-1", task_id=task_id) ``` +## Tool Restrictions + +**Full MCP access, but use `roboco_git_*` not native git.** + +| Allowed | Blocked | +|---------|---------| +| `roboco_git_*` | Native `Bash(git:*)` | +| `roboco_docs_*` | - | +| `roboco_notify_send` | - | + +See: `roboco_kb_search("tool permissions")` + ## Key Tools | Tool | Purpose | @@ -120,6 +132,13 @@ When receiving escalation: 4. Communicate decision 5. Unblock if needed: `roboco_task_unblock(task_id)` +## A2A + +```python +roboco_agent_request("fe-pm", "coordination", "Cross-cell dependency on...", task_id) +roboco_a2a_check() # Check inbox +``` + ## Escalation Escalate to Main PM when: diff --git a/docs/rag/roles/ceo.md b/docs/rag/roles/ceo.md index 6a0f510a..c739dd28 100644 --- a/docs/rag/roles/ceo.md +++ b/docs/rag/roles/ceo.md @@ -27,6 +27,10 @@ - Cancel tasks (by design - CEO observes/approves, doesn't manage) - Should not be doing day-to-day task management +## Tool Note + +Prefer `roboco_git_*` MCP tools over native git for audit trail. + ## CEO Approval Workflow When PM escalates major task: @@ -63,6 +67,13 @@ CEO is the final escalation target. Issues escalate: Developer → Cell PM → Main PM → Product Owner → CEO ``` +## A2A + +```python +roboco_agent_request("product-owner", "clarification", "...", task_id) +roboco_a2a_check() # Check inbox +``` + ## Communication CEO has access to all channels including: diff --git a/docs/rag/roles/developer.md b/docs/rag/roles/developer.md index e498a0e1..ce1671a1 100644 --- a/docs/rag/roles/developer.md +++ b/docs/rag/roles/developer.md @@ -36,11 +36,28 @@ ## Task Flow ``` -pending → claim → start → work → submit_verification → submit_qa - ↑ ↓ - └──────── needs_revision ←──── (QA fails) +pending → claim → plan → start → work → submit_verification → submit_qa + ↑ ↓ + └──────────── needs_revision ←──────── (QA fails) ``` +## Workflow States + +| State | Meaning | +|-------|---------| +| `NEEDS_PLAN` | Must call `roboco_task_plan()` first | +| `WAITING_FOR_BRANCH` | PM must create branch (git tasks) | +| `READY_TO_START` | Call `roboco_task_start()` | +| `EXECUTING` | Work in progress | +| `REVISION_REQUIRED` | Fix QA/PM feedback | + +## Tool Restrictions + +Use `roboco_*` MCP tools, not native Claude tools: +- Git: `roboco_git_*` (native git blocked) +- Write/Edit: workspace only +- See: `roboco_kb_search("tool permissions")` + ## Key Tools | Tool | Purpose | @@ -67,6 +84,18 @@ pending → claim → start → work → submit_verification → submit_qa 4. Write journal reflection: `roboco_journal_reflect()` 5. Push branch: `roboco_git_push()` +## A2A Collaboration + +Direct peer-to-peer messaging: + +```python +# Request review (task_id required) +roboco_agent_request("be-qa", "code_review", "Please review", task_id) + +# Check inbox for incoming messages +roboco_a2a_check() +``` + ## Escalation Escalate to Cell PM when: diff --git a/docs/rag/roles/documenter.md b/docs/rag/roles/documenter.md index 22381a8e..7d4d8e66 100644 --- a/docs/rag/roles/documenter.md +++ b/docs/rag/roles/documenter.md @@ -41,6 +41,18 @@ awaiting_documentation → claim → start → write → docs_complete awaiting_pm_review ``` +## Tool Restrictions + +**Write access limited to docs directory only.** + +| Allowed | Blocked | +|---------|---------| +| `roboco_docs_*` | `Write/Edit` outside `/app/docs/` | +| `roboco_git_*` | Native git commands | +| `Write/Edit` in `/app/docs/**` | Source code modification | + +See: `roboco_kb_search("tool permissions")` + ## Key Tools | Tool | Purpose | @@ -121,6 +133,13 @@ If documenter == original_developer, the claim is FORBIDDEN. 2. Journal your work: `roboco_journal_entry({type: "documentation"})` 3. Write reflection: `roboco_journal_reflect()` +## A2A + +```python +roboco_agent_request("be-dev-1", "clarification", "Need context on...", task_id) +roboco_a2a_check() # Check inbox +``` + ## Escalation Escalate to Cell PM when: diff --git a/docs/rag/roles/head-marketing.md b/docs/rag/roles/head-marketing.md index 22ed1cb1..d2d65698 100644 --- a/docs/rag/roles/head-marketing.md +++ b/docs/rag/roles/head-marketing.md @@ -27,6 +27,10 @@ - Claim tasks (board observes/approves) - Clear/refresh KB indexes +## Tool Note + +Use `roboco_git_*` MCP tools, not native git commands. + ## Key Permissions | Permission | Access | @@ -46,9 +50,14 @@ Escalates directly to CEO. Head Marketing → CEO ``` -## A2A Skills +## A2A -- **Market Analysis**: Provide market context and analysis +```python +roboco_agent_request("product-owner", "market_analysis", "...", task_id) +roboco_a2a_check() # Check inbox +``` + +Skills: market_analysis ## Communication diff --git a/docs/rag/roles/main-pm.md b/docs/rag/roles/main-pm.md index 1a03a845..f71bb729 100644 --- a/docs/rag/roles/main-pm.md +++ b/docs/rag/roles/main-pm.md @@ -80,6 +80,19 @@ roboco_journal_read_team("be-pm") roboco_journal_read_team("fe-pm") ``` +## Tool Restrictions + +**Full MCP access, but use `roboco_git_*` not native git.** + +| Allowed | Blocked | +|---------|---------| +| `roboco_git_*` | Native `Bash(git:*)` | +| `roboco_docs_*` | - | +| `roboco_notify_send` | - | +| All task management | - | + +See: `roboco_kb_search("tool permissions")` + ## Key Tools | Tool | Purpose | @@ -97,6 +110,13 @@ When Cell PM escalates: 3. Coordinate with other Cell PMs if needed 4. Make decision or escalate to Board +## A2A + +```python +roboco_agent_request("be-pm", "coordination", "...", task_id) +roboco_a2a_check() # Check inbox +``` + ## Escalation Escalate to Product Owner when: diff --git a/docs/rag/roles/product-owner.md b/docs/rag/roles/product-owner.md index 1ee0df8e..cef6809b 100644 --- a/docs/rag/roles/product-owner.md +++ b/docs/rag/roles/product-owner.md @@ -28,6 +28,10 @@ - Claim tasks (board observes/approves) - Clear/refresh KB indexes +## Tool Note + +Use `roboco_git_*` MCP tools, not native git commands. + ## Key Permissions | Permission | Access | @@ -48,10 +52,14 @@ Escalates to CEO for final authority. Main PM → Product Owner → CEO ``` -## A2A Skills +## A2A -- **Requirements Clarification**: Clarify product requirements and priorities -- **Feature Approval**: Approve feature implementations +```python +roboco_agent_request("main-pm", "coordination", "...", task_id) +roboco_a2a_check() # Check inbox +``` + +Skills: requirements_clarification, feature_approval ## Communication diff --git a/docs/rag/roles/qa.md b/docs/rag/roles/qa.md index 212e45f2..00c09a08 100644 --- a/docs/rag/roles/qa.md +++ b/docs/rag/roles/qa.md @@ -43,6 +43,18 @@ awaiting_qa → claim → start → review → pass/fail fail: needs_revision (back to developer) ``` +## Tool Restrictions + +**You are read-only.** Cannot modify code or commit. + +| Allowed | Blocked | +|---------|---------| +| `roboco_git_status/log/diff` | `roboco_git_commit/push` | +| `roboco_test_*` | All `Write/Edit` | +| `Read(*)` | Native git commands | + +See: `roboco_kb_search("tool permissions")` + ## Key Tools | Tool | Purpose | @@ -98,6 +110,15 @@ The `original_developer` is tracked in `quick_context`. If QA agent == original 2. Write reflection: `roboco_journal_reflect()` 3. Provide clear reasoning in pass/fail notes +## A2A Requests + +Developers send code review requests via A2A: + +```python +# Check inbox (auto-notified via hook) +roboco_a2a_check() +``` + ## Escalation Escalate to Cell PM when: diff --git a/docs/rag/tools/a2a-tools.md b/docs/rag/tools/a2a-tools.md new file mode 100644 index 00000000..53d9b10c --- /dev/null +++ b/docs/rag/tools/a2a-tools.md @@ -0,0 +1,82 @@ +# A2A (Agent-to-Agent) Tools + +## Overview + +A2A enables direct peer-to-peer communication between agents about existing tasks. + +**Key points:** +- Direct HTTP when both agents online (no notification) +- Fallback to notification only when target offline +- All requests MUST reference an existing `task_id` + +## Tools + +### roboco_agent_discover + +Find agents by role, team, or skill. + +```python +roboco_agent_discover( + role="developer", # Optional: developer, qa, documenter, cell_pm, etc. + team="backend", # Optional: backend, frontend, ux_ui + skill="code_review" # Optional: specific capability +) +``` + +### roboco_agent_request + +Send A2A message to another agent. + +```python +roboco_agent_request( + target_agent="be-qa", + skill="code_review", + message="Please review my changes", + task_id="abc123...", # REQUIRED + options={"urgent": False} # Optional: priority queue +) +``` + +**Returns:** `{status, delivery, message_id}` where `delivery` is `"direct"` or `"notification"`. + +### roboco_a2a_check + +Poll your inbox for incoming A2A messages. + +```python +roboco_a2a_check() +``` + +**Returns:** `{messages: [...], count: N}` - messages from other agents. + +**Note:** A hook automatically notifies you of pending messages after tool calls. + +## Common Use Cases + +| Need | Action | +|------|--------| +| Code review | `roboco_agent_request("be-qa", "code_review", "...", task_id)` | +| Clarification | `roboco_agent_request("be-pm", "clarification", "...", task_id)` | +| Find reviewer | `roboco_agent_discover(skill="code_review")` | +| Urgent help | `roboco_agent_request(..., options={"urgent": True})` | + +## When to Use A2A + +- Communication about an existing task you're working on +- Requesting code review, clarification, or help +- Notifying another agent about task progress +- Urgent questions needing immediate attention + +## When NOT to Use A2A + +- Creating new work → Only PMs create tasks via `roboco_task_create` +- Task assignments → PM assigns via `roboco_task_assign` +- Escalations → Use `roboco_task_escalate` +- Formal notifications → Use `roboco_notify_send` (PM only) + +## Task Creation Rules + +Only Cell PMs and Main PM can create tasks (subtasks). +If an agent receives an A2A request that requires new work: +1. Escalate to PM: `roboco_task_escalate(task_id, "Needs subtask for...")` +2. PM decides whether to create a subtask diff --git a/docs/rag/troubleshooting/blocked-tools.md b/docs/rag/troubleshooting/blocked-tools.md new file mode 100644 index 00000000..1ec4f384 --- /dev/null +++ b/docs/rag/troubleshooting/blocked-tools.md @@ -0,0 +1,56 @@ +# Blocked Tools + +## Native Git Commands Blocked + +**Symptom:** `Bash(git commit)` or similar git command denied + +**Cause:** Native git commands are blocked for all agents + +**Solution:** Use MCP tools instead: +| Blocked | Use Instead | +|---------|-------------| +| `git commit` | `roboco_git_commit()` | +| `git push` | `roboco_git_push()` | +| `git status` | `roboco_git_status()` | +| `git diff` | `roboco_git_diff()` | +| `git log` | `roboco_git_log()` | + +## Write/Edit Outside Workspace + +**Symptom:** `Write()` or `Edit()` denied for a file path + +**Cause:** Write operations restricted to your workspace + +**Solution:** +- Developers: Only write in `/data/workspaces/{project}/{team}/{agent-id}/` +- Documenters: Only write in `/app/docs/` +- QA: No write access (review only) + +## QA Cannot Commit + +**Symptom:** `roboco_git_commit()` denied for QA agent + +**Cause:** QA role is read-only, cannot modify code + +**Solution:** QA reviews and provides feedback. Developers make fixes. + +## NO_PLAN Error + +**Symptom:** `roboco_task_start()` returns NO_PLAN error + +**Cause:** Task has no plan submitted + +**Solution:** Call `roboco_task_plan()` before `roboco_task_start()` + +See: `roboco_kb_search("task planning workflow")` + +## WAITING_FOR_BRANCH + +**Symptom:** Can't start git task, state is WAITING_FOR_BRANCH + +**Cause:** PM hasn't created the branch yet + +**Solution:** +1. Message PM to create branch +2. Or escalate: `roboco_task_escalate(task_id, "Need branch")` +3. Wait for branch_name to be set on task diff --git a/docs/rag/troubleshooting/common-issues.md b/docs/rag/troubleshooting/common-issues.md index 93cf4d91..61c7d669 100644 --- a/docs/rag/troubleshooting/common-issues.md +++ b/docs/rag/troubleshooting/common-issues.md @@ -120,3 +120,24 @@ roboco_docs_write({ - Check existing: `roboco_docs_list(task_id)` or `roboco_kb_search("topic")` **Note**: `roboco_docs_write()` auto-deduplicates via RAG by **content similarity**. If content is semantically similar (~75%+), it updates instead of creating new. + +## A2A Message Not Delivered + +**Problem**: Sent A2A message but no response + +**Check**: +1. Did you include `task_id`? (required) +2. Check delivery status in response: `"direct"` or `"notification"` +3. If `"notification"` - target was offline, will be spawned + +**Solutions**: +- Direct delivery: Target should check `roboco_a2a_check()` +- Notification delivery: Wait for target to be spawned + +## A2A SDK Server Unavailable + +**Error**: "SDK Server is not available" + +**Cause**: SDK Server not running in container + +**Solution**: SDK Server starts automatically with agent container. If error persists, container may need restart. diff --git a/docs/rag/workflows/a2a-collaboration.md b/docs/rag/workflows/a2a-collaboration.md new file mode 100644 index 00000000..7fb19026 --- /dev/null +++ b/docs/rag/workflows/a2a-collaboration.md @@ -0,0 +1,74 @@ +# A2A Collaboration Workflow + +## Overview + +Agents communicate directly via SDK Server (port 9000) for true peer-to-peer messaging. + +**Key:** A2A requires `task_id` - it's about existing tasks, NOT task creation. + +## Flow + +``` +1. Discover → roboco_agent_discover(role, team, skill) +2. Request → roboco_agent_request(target, skill, message, task_id) +3. Check → roboco_a2a_check() polls your inbox (auto-notified via hook) +4. Respond → Work on task or reply via roboco_agent_request +``` + +## Delivery + +| Target State | Delivery | Creates Notification? | +|--------------|----------|----------------------| +| Online | Direct HTTP to SDK | NO | +| Offline | Fallback via API | YES (spawns target) | + +## Example + +```python +# Request code review for task ABC123 +result = roboco_agent_request( + target_agent="be-qa", + skill="code_review", + message="Please review my changes", + task_id="ABC123" +) +# result.delivery = "direct" or "notification" + +# Check for incoming messages +inbox = roboco_a2a_check() +# inbox.messages = [{from, task_id, skill, message, priority}, ...] +``` + +## Urgency + +```python +roboco_agent_request(..., options={"urgent": True}) # Priority queue +``` + +## Agent Skills + +| Role | Skills | +|------|--------| +| Developer | `code_review`, `implementation`, `debugging`, `revision` | +| QA | `code_review`, `testing`, `qa_review` | +| Documenter | `documentation`, `api_docs` | +| PM | `task_planning`, `coordination`, `clarification` | + +## Task Creation Rules + +**Only PMs can create tasks.** + +If you receive an A2A request that needs new work: +1. Escalate: `roboco_task_escalate(task_id, "Needs subtask for X")` +2. PM decides whether to create subtask + +## Permissions + +All agents can: +- Discover other agents +- Send A2A requests (must include task_id) +- Check request status + +All agents CANNOT: +- Create tasks via A2A (no automatic task creation) +- Send A2A without a task_id diff --git a/docs/rag/workflows/task-planning.md b/docs/rag/workflows/task-planning.md new file mode 100644 index 00000000..7be63e75 --- /dev/null +++ b/docs/rag/workflows/task-planning.md @@ -0,0 +1,49 @@ +# Task Planning Workflow + +## Overview + +Planning is **required** before starting work. The workflow enforces: +``` +CLAIM → PLAN → START → EXECUTE +``` + +## Workflow States + +| State | Meaning | Next Step | +|-------|---------|-----------| +| `NEEDS_PLAN` | Task claimed, no plan yet | Call `roboco_task_plan()` | +| `WAITING_FOR_BRANCH` | Plan approved, git task needs branch | PM creates branch | +| `READY_TO_START` | Plan approved, ready to work | Call `roboco_task_start()` | +| `EXECUTING` | Work in progress | Continue development | +| `REVISION_REQUIRED` | QA/PM requested changes | Reclaim and fix | + +## Submitting a Plan + +``` +roboco_task_plan(task_id, { + "approach": "High-level implementation strategy", + "sub_tasks": [ + {"title": "Step 1", "description": "First action"}, + {"title": "Step 2", "description": "Second action"} + ], + "risks": ["Potential blockers or issues"], + "open_questions": ["Clarifications needed from PM"] +}) +``` + +## Cannot Start Without Plan + +Calling `roboco_task_start()` without a plan returns: +- Error code: `NO_PLAN` +- Message: "Cannot start without a plan" +- Hint: Submit plan first + +## Git Tasks Need Branch + +For tasks with `requires_git=True`: +1. Submit plan +2. PM creates branch: `roboco_git_create_branch()` +3. Task gets `branch_name` field set +4. Then you can call `roboco_task_start()` + +If no branch: workflow state = `WAITING_FOR_BRANCH` diff --git a/roboco/agent_sdk/__init__.py b/roboco/agent_sdk/__init__.py new file mode 100644 index 00000000..2bcbe119 --- /dev/null +++ b/roboco/agent_sdk/__init__.py @@ -0,0 +1,10 @@ +""" +Agent SDK Server for A2A Communication. + +This module provides a lightweight FastAPI server that runs alongside Claude Code +in each agent container, enabling true peer-to-peer agent communication. + +Components: +- models.py: Pydantic models for A2A messages +- server.py: FastAPI application with endpoints for send/receive/poll +""" diff --git a/roboco/agent_sdk/models.py b/roboco/agent_sdk/models.py new file mode 100644 index 00000000..caf7a3ad --- /dev/null +++ b/roboco/agent_sdk/models.py @@ -0,0 +1,68 @@ +""" +Agent SDK Models. + +Pydantic models for A2A messaging between agents. +""" + +from datetime import UTC, datetime +from enum import Enum +from uuid import UUID, uuid4 + +from pydantic import BaseModel, Field + + +class MessagePriority(str, Enum): + """Priority level for A2A messages.""" + + URGENT = "urgent" + NORMAL = "normal" + + +class A2AMessage(BaseModel): + """ + A2A message received by SDK server. + + Represents a message from one agent to another about a specific task. + """ + + id: UUID = Field(default_factory=uuid4) + from_agent: str = Field(..., description="Sender agent slug") + to_agent: str = Field(..., description="Recipient agent slug") + task_id: str = Field(..., description="Related task ID") + skill: str = Field(..., description="Requested skill (e.g., code_review)") + content: str = Field(..., description="Message content") + priority: MessagePriority = Field(default=MessagePriority.NORMAL) + timestamp: datetime = Field(default_factory=lambda: datetime.now(UTC)) + acked: bool = Field(default=False, description="Whether message was acknowledged") + + +class SendRequest(BaseModel): + """Request to send an A2A message.""" + + target_agent: str = Field(..., description="Target agent slug") + skill: str = Field(..., description="Skill being requested") + message: str = Field(..., description="Message content") + task_id: str = Field(..., description="Related task ID") + urgent: bool = Field(default=False, description="Whether this is urgent") + + +class SendResponse(BaseModel): + """Response from sending an A2A message.""" + + status: str = Field(..., description="Status: sent, queued, failed") + message_id: str = Field(..., description="Message UUID") + delivery: str = Field(..., description="Delivery method: direct, notification") + + +class InboxResponse(BaseModel): + """Response from polling the inbox.""" + + messages: list[A2AMessage] = Field(default_factory=list) + count: int = Field(default=0, description="Number of messages returned") + + +class HealthResponse(BaseModel): + """Health check response.""" + + status: str = Field(default="ok") + agent_id: str = Field(..., description="This agent's ID") diff --git a/roboco/agent_sdk/server.py b/roboco/agent_sdk/server.py new file mode 100644 index 00000000..43887d06 --- /dev/null +++ b/roboco/agent_sdk/server.py @@ -0,0 +1,260 @@ +""" +Agent SDK Server. + +Lightweight FastAPI server running inside each agent container. +Provides bidirectional A2A communication capabilities. + +Features: +- Receive A2A messages from other agents +- Priority queue (urgent messages first) +- Inbox polling for Claude Code +- Fallback to main API when target offline +""" + +import os +from collections import deque + +import httpx +import structlog +import uvicorn +from fastapi import FastAPI + +from roboco.agent_sdk.models import ( + A2AMessage, + HealthResponse, + InboxResponse, + MessagePriority, + SendRequest, + SendResponse, +) + +logger = structlog.get_logger() + +# Environment configuration +AGENT_ID = os.environ.get("ROBOCO_AGENT_ID", "unknown") +MAIN_API_URL = os.environ.get("ROBOCO_API_URL", "http://roboco-orchestrator:8000") +SDK_PORT = int(os.environ.get("ROBOCO_SDK_PORT", "9000")) + +app = FastAPI( + title=f"RoboCo SDK Server ({AGENT_ID})", + description="Agent-to-Agent communication server", + version="1.0.0", +) + +# Priority queues (urgent first) +urgent_inbox: deque[A2AMessage] = deque(maxlen=100) +normal_inbox: deque[A2AMessage] = deque(maxlen=500) + + +# ============================================================================= +# HEALTH +# ============================================================================= + + +@app.get("/health", response_model=HealthResponse) +async def health() -> HealthResponse: + """Health check endpoint.""" + return HealthResponse(status="ok", agent_id=AGENT_ID) + + +# ============================================================================= +# RECEIVE (from other agents) +# ============================================================================= + + +@app.post("/a2a/receive") +async def receive_message(msg: A2AMessage) -> dict[str, str]: + """ + Receive A2A message from another agent. + + Messages are queued by priority for Claude Code to poll. + """ + if msg.priority == MessagePriority.URGENT: + urgent_inbox.append(msg) + logger.info( + "Received urgent A2A message", + from_agent=msg.from_agent, + task_id=msg.task_id, + skill=msg.skill, + ) + else: + normal_inbox.append(msg) + logger.info( + "Received A2A message", + from_agent=msg.from_agent, + task_id=msg.task_id, + skill=msg.skill, + ) + + return {"status": "queued", "message_id": str(msg.id)} + + +# ============================================================================= +# SEND (to other agents) +# ============================================================================= + + +@app.post("/a2a/send", response_model=SendResponse) +async def send_message(req: SendRequest) -> SendResponse: + """ + Send A2A message to another agent. + + Attempts direct delivery via HTTP. Falls back to notification + via main API if target agent is offline. + """ + # Container name = roboco-agent-{slug} + target_url = f"http://roboco-agent-{req.target_agent}:{SDK_PORT}/a2a/receive" + + msg = A2AMessage( + from_agent=AGENT_ID, + to_agent=req.target_agent, + task_id=req.task_id, + skill=req.skill, + content=req.message, + priority=MessagePriority.URGENT if req.urgent else MessagePriority.NORMAL, + ) + + async with httpx.AsyncClient() as client: + try: + resp = await client.post( + target_url, + json=msg.model_dump(mode="json"), + timeout=5.0, + ) + resp.raise_for_status() + + logger.info( + "A2A message sent directly", + to_agent=req.target_agent, + task_id=req.task_id, + skill=req.skill, + ) + + return SendResponse( + status="sent", + message_id=str(msg.id), + delivery="direct", + ) + + except (httpx.ConnectError, httpx.TimeoutException, httpx.HTTPStatusError) as e: + # Agent offline or unreachable - fall back to notification + logger.info( + "Target agent offline, falling back to notification", + to_agent=req.target_agent, + error=str(e), + ) + + await _create_notification_fallback(req) + + return SendResponse( + status="sent", + message_id=str(msg.id), + delivery="notification", + ) + + +async def _create_notification_fallback(req: SendRequest) -> None: + """Create notification via main API when target agent is offline.""" + async with httpx.AsyncClient() as client: + try: + await client.post( + f"{MAIN_API_URL}/api/v1/a2a/message/send", + json={ + "message": { + "role": "user", + "parts": [{"type": "text", "text": req.message}], + "taskId": req.task_id, + }, + "metadata": { + "from_agent": AGENT_ID, + "target_agent": req.target_agent, + "skill": req.skill, + "urgent": req.urgent, + }, + }, + headers={ + "X-Agent-ID": AGENT_ID, + "X-Agent-Role": "developer", # SDK doesn't know role + }, + timeout=10.0, + ) + logger.info( + "Notification fallback created", + to_agent=req.target_agent, + task_id=req.task_id, + ) + except Exception as e: + logger.error( + "Failed to create notification fallback", + to_agent=req.target_agent, + error=str(e), + ) + + +# ============================================================================= +# INBOX (for Claude Code to poll) +# ============================================================================= + + +@app.get("/inbox/poll", response_model=InboxResponse) +async def poll_inbox(limit: int = 10) -> InboxResponse: + """ + Poll inbox for pending A2A messages. + + Returns messages in priority order (urgent first). + Messages are removed from queue once returned. + """ + messages: list[A2AMessage] = [] + + # Urgent first + while urgent_inbox and len(messages) < limit: + messages.append(urgent_inbox.popleft()) + + # Then normal + while normal_inbox and len(messages) < limit: + messages.append(normal_inbox.popleft()) + + if messages: + logger.info( + "Inbox polled", + message_count=len(messages), + urgent_remaining=len(urgent_inbox), + normal_remaining=len(normal_inbox), + ) + + return InboxResponse(messages=messages, count=len(messages)) + + +@app.post("/inbox/ack/{message_id}") +async def ack_message(message_id: str) -> dict[str, str]: + """ + Acknowledge message was processed. + + For now, messages are removed on poll. This endpoint exists + for future Redis persistence where we might need explicit ACK. + """ + logger.info("Message acknowledged", message_id=message_id) + return {"status": "acked", "message_id": message_id} + + +@app.get("/inbox/count") +async def inbox_count() -> dict[str, int]: + """Get count of pending messages without consuming them.""" + return { + "urgent": len(urgent_inbox), + "normal": len(normal_inbox), + "total": len(urgent_inbox) + len(normal_inbox), + } + + +# ============================================================================= +# MAIN +# ============================================================================= + +if __name__ == "__main__": + logger.info( + "Starting SDK Server", + agent_id=AGENT_ID, + port=SDK_PORT, + ) + uvicorn.run(app, host="0.0.0.0", port=SDK_PORT) diff --git a/roboco/agents_config.py b/roboco/agents_config.py index eb5a4b51..80372b38 100644 --- a/roboco/agents_config.py +++ b/roboco/agents_config.py @@ -579,3 +579,125 @@ def get_agent_skills(agent_id: str) -> list[dict]: """Get A2A skills for an agent based on their role.""" role = get_agent_role(agent_id) return list(ROLE_SKILLS.get(role, [])) + + +# ============================================================================= +# A2A PERMISSION ENFORCEMENT +# ============================================================================= +# A2A follows the same hierarchy as escalations and notifications: +# - Within cell: Direct A2A allowed +# - Cross-cell: Must go through Cell PM → Main PM +# - To board: Must go through Main PM +# - To CEO: Must go through board + +# Roles that can reach each other directly (CEO is human - use notifications) +_BOARD_ROLES: Final[frozenset[str]] = frozenset( + {"product_owner", "head_marketing", "auditor", "main_pm"} +) +_MAIN_PM_TARGETS: Final[frozenset[str]] = frozenset( + {"cell_pm", "main_pm", "product_owner", "head_marketing", "auditor"} +) + + +def _check_cell_pm_a2a( + from_team: str | None, to_agent: str, to_role: str, to_team: str | None +) -> tuple[bool, str | None]: + """Check A2A permissions for cell PM.""" + # Own cell, other PMs, or main-pm + if to_team == from_team or to_role in ("cell_pm", "main_pm"): + return True, None + # Board/CEO - escalate + if to_role in _BOARD_ROLES: + return False, f"Cell PMs cannot A2A {to_role}. Escalate through main-pm." + # Other cell members + return False, f"Cannot A2A {to_agent} (different cell). Use main-pm." + + +def _check_cell_member_a2a( + from_agent: str, from_team: str, to_agent: str, to_role: str, to_team: str | None +) -> tuple[bool, str | None]: + """Check A2A permissions for cell members (dev, qa, doc).""" + cell_pm = get_pm_for_team(from_team) + # Same cell - allowed + if to_team == from_team: + return True, None + # Cross-cell + if to_team: + target_pm = get_pm_for_team(to_team) + return ( + False, + f"Cannot A2A {to_agent} (cell: {to_team}). " + f"Ask {cell_pm} to coordinate with {target_pm}.", + ) + # Management - not direct + return False, f"Cannot A2A {to_role}. Route: {from_agent} → {cell_pm} → main-pm." + + +def _check_main_pm_a2a( + to_role: str, to_team: str | None +) -> tuple[bool, str | None]: + """Check A2A permissions for main PM.""" + if to_role in _MAIN_PM_TARGETS: + return True, None + pm = get_pm_for_team(to_team) if to_team else "cell-pm" + return False, f"Main PM cannot A2A {to_role}s. Route through {pm or 'cell-pm'}." + + +def can_a2a_direct(from_agent: str, to_agent: str) -> tuple[bool, str | None]: + """ + Check if from_agent can send A2A directly to to_agent. + + Returns (allowed, error_message). Error explains who to contact instead. + """ + from_role = get_agent_role(from_agent) + to_role = get_agent_role(to_agent) + from_team = get_agent_team(from_agent) + to_team = get_agent_team(to_agent) + + # CEO is human - cannot A2A, use notifications + if to_role == "ceo": + return False, "CEO is human. Use roboco_notify_send() instead of A2A." + + # Board → board/main-pm (not CEO, not cells directly) + if from_role in ("product_owner", "head_marketing", "auditor"): + return (True, None) if to_role in _BOARD_ROLES else ( + False, f"Board cannot A2A {to_role}s. Route through main-pm." + ) + + # Dispatch to role-specific handlers + handlers: dict[str, tuple[bool, str | None]] = { + "main_pm": _check_main_pm_a2a(to_role, to_team), + "cell_pm": _check_cell_pm_a2a(from_team, to_agent, to_role, to_team), + } + if from_role in handlers: + return handlers[from_role] + + # Cell members - use helper + if from_team: + return _check_cell_member_a2a(from_agent, from_team, to_agent, to_role, to_team) + + return False, f"A2A from {from_agent} to {to_agent} not permitted." + + +def get_a2a_route_hint(from_agent: str, to_agent: str) -> str: + """Get a hint for how to properly route an A2A message.""" + to_role = get_agent_role(to_agent) + from_team = get_agent_team(from_agent) + to_team = get_agent_team(to_agent) + + # CEO is human - no A2A route, use notifications + if to_role == "ceo": + return "CEO is human. Use roboco_notify_send() for CEO communication." + + # Cross-cell routing + if from_team and to_team and from_team != to_team: + from_pm = get_pm_for_team(from_team) + to_pm = get_pm_for_team(to_team) + return f"Route: {from_agent}→{from_pm}→main-pm→{to_pm}→{to_agent}" + + # Cell member to management + if from_team: + cell_pm = get_pm_for_team(from_team) + return f"Route: {from_agent}→{cell_pm}→main-pm→board" + + return "Use roboco_task_escalate() for proper escalation." diff --git a/roboco/api/routes/a2a.py b/roboco/api/routes/a2a.py index e43faf29..ff9e28d6 100644 --- a/roboco/api/routes/a2a.py +++ b/roboco/api/routes/a2a.py @@ -30,7 +30,6 @@ from roboco.models.a2a import ( CancelTaskRequest, ListTasksResponse, SendMessageRequest, - SendMessageResponse, ) from roboco.services.a2a import A2AService @@ -94,21 +93,42 @@ async def get_agent_card( async def send_message( request: SendMessageRequest, db: DbSession, -) -> SendMessageResponse: +) -> dict[str, Any]: """ - Send a message to create or update an A2A task. + Send an A2A message (fallback endpoint). - This is the primary A2A interaction endpoint. Messages sent here - create new tasks or continue existing conversations. + This endpoint is used by the SDK Server when the target agent is offline. + It creates a notification that the orchestrator dispatcher will pick up + to spawn the target agent. + + DOES NOT create tasks. Creates notifications only. + task_id is REQUIRED - A2A is about existing tasks. """ service = A2AService(db) message = request.message task_id_str = message.task_id - if task_id_str: - # Update existing task + # task_id is REQUIRED for A2A + if not task_id_str: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail={ + "error": "TASK_ID_REQUIRED", + "message": "A2A requests must include task_id.", + "hint": "A2A is for communication about existing tasks. " + "Use roboco_task_create() first if you need a new task.", + }, + ) + + # Check if this is a response to an existing A2A conversation + metadata = request.metadata or {} + is_response = metadata.get("is_response", False) + + if is_response: + # Update existing task with response message + responder = metadata.get("from_agent") try: - task = await service.update_task_from_message(task_id_str, message) + await service.update_task_from_message(task_id_str, message, responder) except ValueError as e: error_msg = str(e) if "Invalid task ID" in error_msg: @@ -120,19 +140,32 @@ async def send_message( status_code=status.HTTP_404_NOT_FOUND, detail=error_msg, ) from None - else: - # Create new task from message - try: - task = await service.create_task_from_a2a_message(request) - except ValueError as e: + await db.commit() + return {"status": "response_sent", "task_id": task_id_str} + + # Create A2A notification (NOT a task) and route to agent + try: + result = await service.create_a2a_notification(request) + except ValueError as e: + error_str = str(e) + # Check if it's a permission error (includes "Hint:") + if "Hint:" in error_str: + parts = error_str.split(" Hint: ", 1) raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=str(e), + status_code=status.HTTP_403_FORBIDDEN, + detail={ + "error": "A2A_NOT_PERMITTED", + "message": parts[0], + "hint": parts[1] if len(parts) > 1 else "", + }, ) from None + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail={"error": "A2A_ERROR", "message": error_str}, + ) from None await db.commit() - await db.refresh(task) - return SendMessageResponse(task=service.task_to_a2a(task)) + return {"status": "success", "a2a_request": result} @router.post("/message/stream") diff --git a/roboco/api/routes/notifications.py b/roboco/api/routes/notifications.py index 80cff339..3173b6d1 100644 --- a/roboco/api/routes/notifications.py +++ b/roboco/api/routes/notifications.py @@ -226,3 +226,117 @@ async def mark_as_read( if agent_id not in notification.read_by: notification.read_by = [*notification.read_by, agent_id] await db.flush() + + +@router.get( + "/pending-a2a", + summary="Check pending A2A", + description="Check if there's a pending A2A notification to a target about a task.", +) +async def check_pending_a2a( + db: DbSession, + from_agent: str, + to_agent: str, + task_id: str, +) -> dict[str, bool]: + """ + Check if there's already a pending A2A notification. + + Prevents duplicate messages - one message per task until response. + """ + from roboco.models.base import NotificationType + from roboco.seeds.initial_data import AGENT_UUIDS + + from_uuid = AGENT_UUIDS.get(from_agent) + to_uuid = AGENT_UUIDS.get(to_agent) + + if not from_uuid or not to_uuid: + return {"has_pending": False} + + # Validate task_id is a valid UUID + try: + task_uuid = UUID(task_id) + except ValueError: + return {"has_pending": False} + + # Check for unacked A2A_REQUEST from this agent to target about this task + result = await db.execute( + select(NotificationTable).where( + NotificationTable.type == NotificationType.A2A_REQUEST, + NotificationTable.from_agent == UUID(from_uuid), + NotificationTable.related_task_id == task_uuid, + NotificationTable.to_agents.contains([UUID(to_uuid)]), + ) + ) + notifications = result.scalars().all() + + # Check if any are unacked by the target + to_uuid_obj = UUID(to_uuid) + for notif in notifications: + if to_uuid_obj not in notif.acked_by: + return {"has_pending": True} + + return {"has_pending": False} + + +@router.post( + "/ack-a2a", + status_code=status.HTTP_204_NO_CONTENT, + summary="Auto-ack A2A notifications", + description="Acknowledge A2A notifications when responding. Called by SDK.", +) +async def ack_a2a_notifications( + db: DbSession, + data: dict[str, str], +) -> None: + """ + Auto-acknowledge A2A notifications when responding. + + When agent B responds to agent A about a task, this acks any pending + A2A_REQUEST notifications from A to B about that task. + + Body: {from_agent, to_agent, task_id} + """ + from roboco.models.base import NotificationType + from roboco.seeds.initial_data import AGENT_UUIDS + + from_agent_slug = data.get("from_agent", "") + to_agent_slug = data.get("to_agent", "") + task_id_str = data.get("task_id", "") + + # Get UUIDs from slugs + from_agent_uuid = AGENT_UUIDS.get(from_agent_slug) + to_agent_uuid = AGENT_UUIDS.get(to_agent_slug) + + if not from_agent_uuid or not to_agent_uuid or not task_id_str: + return # Silently ignore invalid data + + # Validate task_id is a valid UUID + try: + task_uuid = UUID(task_id_str) + except ValueError: + return # Invalid task ID + + # Find matching A2A_REQUEST notifications + result = await db.execute( + select(NotificationTable).where( + NotificationTable.type == NotificationType.A2A_REQUEST, + NotificationTable.from_agent == UUID(from_agent_uuid), + NotificationTable.related_task_id == task_uuid, + NotificationTable.to_agents.contains([UUID(to_agent_uuid)]), + ) + ) + notifications = result.scalars().all() + + # Acknowledge each matching notification + to_uuid = UUID(to_agent_uuid) + now = datetime.now(UTC).isoformat() + + for notif in notifications: + if to_uuid not in notif.acked_by: + notif.acked_by = [*notif.acked_by, to_uuid] + notif.acked_at = {**notif.acked_at, str(to_uuid): now} + if to_uuid not in notif.read_by: + notif.read_by = [*notif.read_by, to_uuid] + + await db.flush() diff --git a/roboco/config.py b/roboco/config.py index d6849c14..8d7da6c8 100644 --- a/roboco/config.py +++ b/roboco/config.py @@ -178,7 +178,7 @@ class Settings(BaseSettings): # Local LLM for RAG (HyDE, reranking, etc.) local_llm_model: str = Field( - default="glm-4.6:cloud", + default="glm-4.7:cloud", description="Local LLM for HyDE/RAG (non-thinking models are faster)", ) local_llm_base_url: str = Field( diff --git a/roboco/events/__init__.py b/roboco/events/__init__.py index 7f8a8639..3a2c9857 100644 --- a/roboco/events/__init__.py +++ b/roboco/events/__init__.py @@ -15,7 +15,6 @@ from roboco.events.handlers import ( handle_qa_result, handle_question_answered, handle_session_boundary, - handle_task_assigned, handle_task_status_change, register_default_handlers, set_event_context, @@ -35,7 +34,6 @@ __all__ = [ "handle_qa_result", "handle_question_answered", "handle_session_boundary", - "handle_task_assigned", "handle_task_status_change", "init_event_bus", "register_default_handlers", diff --git a/roboco/events/handlers.py b/roboco/events/handlers.py index 9b34e21e..cd9cb2de 100644 --- a/roboco/events/handlers.py +++ b/roboco/events/handlers.py @@ -324,73 +324,6 @@ async def handle_question_answered(event: Event) -> None: ) -# ============================================================================= -# A2A TASK ASSIGNMENT HANDLER -# ============================================================================= - - -async def handle_task_assigned(event: Event) -> None: - """ - Handle TASK_ASSIGNED event from A2A requests. - - When an agent requests work from another agent via A2A: - 1. If target agent is running → send notification - 2. If target agent is idle/stopped → spawn them with task context - """ - data = event.data - agent_slug = data.get("agent_slug") - task_id = data.get("task_id") - skill = data.get("skill", "general") - message = data.get("message", "") - - if not agent_slug or not task_id: - logger.warning("Incomplete TASK_ASSIGNED event", data=data) - return - - logger.info( - "A2A task assignment received", - agent=agent_slug, - task_id=task_id, - skill=skill, - ) - - # Check if orchestrator is available - if not _context.orchestrator: - logger.warning("No orchestrator available to spawn agent") - return - - # Check if agent is already running - running_agents = _context.orchestrator.get_running_agents() - if agent_slug in running_agents: - # Agent is running - they'll pick up the task via notifications - logger.info(f"Agent {agent_slug} already running, will receive notification") - return - - # Agent not running - spawn them with A2A task context - initial_prompt = ( - f"You have received an A2A request.\n\n" - f"**Skill requested:** {skill}\n" - f"**Task ID:** {task_id}\n" - f"**Message:** {message}\n\n" - f"Use roboco_task_get('{task_id}') to see the full task details, " - f"then proceed with the requested work." - ) - - try: - await _context.orchestrator.spawn_agent( - agent_id=agent_slug, - initial_prompt=initial_prompt, - ) - logger.info(f"Spawned agent {agent_slug} for A2A task {task_id}") - except Exception as e: - logger.error( - "Failed to spawn agent for A2A task", - agent=agent_slug, - task_id=task_id, - error=str(e), - ) - - # ============================================================================= # HANDLER REGISTRATION # ============================================================================= @@ -428,7 +361,7 @@ def register_default_handlers(bus: Any = None) -> None: # Question handlers bus.subscribe(EventType.QUESTION_ANSWERED, handle_question_answered) - # A2A task assignment handlers - bus.subscribe(EventType.TASK_ASSIGNED, handle_task_assigned) + # NOTE: A2A routing is now handled by SDK Server directly + # Orchestrator dispatcher handles fallback spawning via notification polling logger.info("Default event handlers registered") diff --git a/roboco/mcp/a2a_server.py b/roboco/mcp/a2a_server.py index dab6b93e..aaece612 100644 --- a/roboco/mcp/a2a_server.py +++ b/roboco/mcp/a2a_server.py @@ -1,28 +1,41 @@ """ A2A MCP Server -Provides tools for agent-to-agent communication using the A2A protocol. -This enables peer-to-peer agent collaboration without going through -the orchestrator for every interaction. +Provides tools for agent-to-agent communication. +Routes messages through the local SDK Server for true peer-to-peer A2A. Tools available to ALL agents: - roboco_agent_discover: Discover other agents by skill/role/team -- roboco_agent_request: Request another agent to perform work -- roboco_agent_request_status: Check status of a pending request +- roboco_agent_request: Send A2A message to another agent (via SDK) +- roboco_a2a_check: Poll inbox for incoming A2A messages """ +import contextlib +import os from typing import Any +import httpx from mcp.server.fastmcp import FastMCP from roboco.agents_config import ( ALL_AGENTS, + can_a2a_direct, + get_a2a_route_hint, get_agent_role, get_agent_skills, get_agent_team, ) -from roboco.mcp.utils import ApiClient, format_error_response -from roboco.seeds.initial_data import AGENT_UUIDS +from roboco.mcp.utils import format_error_response + +# Current agent ID from environment (set by orchestrator) +AGENT_ID = os.environ.get("ROBOCO_AGENT_ID", "unknown") + +# SDK Server configuration +SDK_URL = os.environ.get("ROBOCO_SDK_URL", "http://localhost:9000") + +# Main API URL (for notification auto-ack) +API_URL = os.environ.get("ROBOCO_API_URL", "http://localhost:8000") + # ============================================================================= # TOOL IMPLEMENTATIONS @@ -34,8 +47,7 @@ async def _handle_discover( team: str | None = None, skill: str | None = None, ) -> dict[str, Any]: - """Discover agents by criteria.""" - # Build local discovery (fast path - no API call needed) + """Discover agents by criteria (local lookup, no API call).""" agents = [] for agent_slug in ALL_AGENTS: @@ -72,61 +84,199 @@ async def _handle_discover( "agents": agents, "count": len(agents), "guidance": ( - f"Found {len(agents)} agent(s). Use roboco_agent_request to request " - "work from a specific agent." + f"Found {len(agents)} agent(s). Use roboco_agent_request to send " + "a message to a specific agent." ), } -async def _handle_request_status( - client: ApiClient, - a2a_task_id: str, +def _validate_a2a_target( + from_agent: str, target_agent: str, skill: str +) -> dict[str, Any] | None: + """Validate A2A target and permissions. Returns error dict or None if valid.""" + # Check target exists + if target_agent not in ALL_AGENTS: + return format_error_response( + "AGENT_NOT_FOUND", + f"Agent '{target_agent}' not found. Use roboco_agent_discover.", + ) + + # Check A2A hierarchy permissions + allowed, error_msg = can_a2a_direct(from_agent, target_agent) + if not allowed: + return format_error_response( + "A2A_NOT_PERMITTED", + error_msg or f"Cannot A2A {target_agent} directly.", + hint=get_a2a_route_hint(from_agent, target_agent), + ) + + # Check skill exists + target_skills = get_agent_skills(target_agent) + skill_ids = [s.get("id", "") for s in target_skills] + if skill not in skill_ids: + available = ", ".join(skill_ids) + return format_error_response( + "SKILL_NOT_FOUND", + f"Agent '{target_agent}' lacks skill '{skill}'. Has: {available}", + ) + + return None + + +async def _auto_ack_a2a_notifications( + from_agent: str, target_agent: str, task_id: str +) -> None: + """Auto-acknowledge A2A notifications when responding. + + When agent B responds to agent A about a task, ack any pending + A2A_REQUEST notifications from A about that task. + """ + async with httpx.AsyncClient() as client: + with contextlib.suppress(Exception): + await client.post( + f"{API_URL}/api/v1/notifications/ack-a2a", + json={ + "from_agent": target_agent, # Original sender + "to_agent": from_agent, # Us (the responder) + "task_id": task_id, + }, + timeout=5.0, + ) + + +async def _check_pending_a2a( + from_agent: str, target_agent: str, task_id: str +) -> dict[str, Any] | None: + """Check if there's already a pending A2A to target about this task. + + Returns error dict if pending message exists, None if ok to send. + """ + async with httpx.AsyncClient() as client: + try: + resp = await client.get( + f"{API_URL}/api/v1/notifications/pending-a2a", + params={ + "from_agent": from_agent, + "to_agent": target_agent, + "task_id": task_id, + }, + timeout=5.0, + ) + resp.raise_for_status() + data = resp.json() + if data.get("has_pending"): + return format_error_response( + "A2A_PENDING", + f"Already sent A2A to {target_agent} about this task.", + hint="Wait for their response before sending another message.", + ) + except Exception: + pass # Non-critical check, allow send if check fails + return None + + +async def _send_via_sdk( + target_agent: str, skill: str, message: str, task_id: str, urgent: bool ) -> dict[str, Any]: - """Check status of an A2A request.""" - resp = await client.get(f"/a2a/tasks/{a2a_task_id}") + """Send A2A message via SDK Server.""" + async with httpx.AsyncClient() as client: + try: + resp = await client.post( + f"{SDK_URL}/a2a/send", + json={ + "target_agent": target_agent, + "skill": skill, + "message": message, + "task_id": task_id, + "urgent": urgent, + }, + timeout=10.0, + ) + resp.raise_for_status() + result = resp.json() - if resp.is_status(404): - return format_error_response( - "TASK_NOT_FOUND", - f"A2A task '{a2a_task_id}' not found", - ) + delivery = result.get("delivery", "unknown") + urgency_note = " (URGENT)" if urgent else "" - if not resp.ok: - return format_error_response( - "STATUS_CHECK_FAILED", - f"Failed to check status: {resp.text}", - ) + return { + "status": "success", + "target_agent": target_agent, + "skill": skill, + "task_id": task_id, + "message_id": result.get("message_id"), + "delivery": delivery, + "guidance": ( + f"A2A sent to {target_agent}{urgency_note}. " + f"Delivery: {delivery}." + ), + } - task = resp.json() - status = task.get("status", {}) - state = status.get("state", "unknown") - message = status.get("message", {}) + except httpx.ConnectError: + return format_error_response( + "SDK_UNAVAILABLE", + "SDK Server not available.", + hint="SDK Server should be running alongside Claude Code.", + ) + except httpx.HTTPStatusError as e: + return format_error_response( + "A2A_SEND_FAILED", + f"Failed to send: {e.response.text}", + ) - result_text = None - if message and message.get("parts"): - for part in message["parts"]: - if part.get("type") == "text": - result_text = part.get("text") - break - guidance = "" - if state == "completed": - guidance = "Request completed. Review the result below." - elif state == "working": - guidance = "Agent is still working on this request. Check again later." - elif state == "input_required": - guidance = "Agent needs more information. Review the message and respond." - elif state in ["failed", "cancelled", "rejected"]: - guidance = f"Request ended with state: {state}." +async def _handle_check() -> dict[str, Any]: + """Poll inbox for incoming A2A messages via SDK Server.""" + async with httpx.AsyncClient() as client: + try: + resp = await client.get(f"{SDK_URL}/inbox/poll", timeout=5.0) + resp.raise_for_status() + data = resp.json() - return { - "a2a_task_id": a2a_task_id, - "state": state, - "result": result_text, - "artifacts": task.get("artifacts", []), - "metadata": task.get("metadata", {}), - "guidance": guidance, - } + messages = data.get("messages", []) + count = data.get("count", 0) + + if count == 0: + return { + "messages": [], + "count": 0, + "guidance": "No pending A2A messages.", + } + + # Format messages for display + formatted = [] + for msg in messages: + formatted.append( + { + "id": str(msg.get("id", "")), + "from": msg.get("from_agent", "unknown"), + "task_id": msg.get("task_id", ""), + "skill": msg.get("skill", ""), + "message": msg.get("content", ""), + "priority": msg.get("priority", "normal"), + "timestamp": msg.get("timestamp", ""), + } + ) + + return { + "messages": formatted, + "count": count, + "guidance": ( + f"You have {count} A2A message(s). " + "Review and respond to each as appropriate." + ), + } + + except httpx.ConnectError: + return format_error_response( + "SDK_UNAVAILABLE", + "SDK Server is not available. Cannot check inbox.", + hint="The SDK Server should be running alongside Claude Code.", + ) + except httpx.HTTPStatusError as e: + return format_error_response( + "INBOX_CHECK_FAILED", + f"Failed to check inbox: {e.response.text}", + ) # ============================================================================= @@ -145,7 +295,6 @@ def create_a2a_mcp_server(agent_id: str) -> FastMCP: Configured FastMCP server """ mcp = FastMCP(f"roboco-a2a-{agent_id}", json_response=True) - client = ApiClient(agent_id) @mcp.tool() async def roboco_agent_discover( @@ -173,112 +322,69 @@ def create_a2a_mcp_server(agent_id: str) -> FastMCP: target_agent: str, skill: str, message: str, - task_id: str | None = None, - blocking: bool = False, + task_id: str, + options: dict[str, Any] | None = None, ) -> dict[str, Any]: """ - Request another agent to perform work using A2A protocol. + Send an A2A message to another agent about a specific task. - This enables direct peer-to-peer collaboration between agents. + Messages are delivered directly if the agent is online, + or via notification if they are offline. Args: - target_agent: Agent slug to request (e.g., "be-qa", "fe-dev-1") - skill: Skill to invoke (e.g., "code_review", "code_implementation") - message: Description of what you need - task_id: Related task ID (optional, for context) - blocking: Wait for response (default: false, async) + target_agent: Agent slug to message (e.g., "be-qa", "fe-dev-1") + skill: Skill being requested (e.g., "code_review", "clarification") + message: Your message content + task_id: REQUIRED - The task this message is about + options: Optional dict with 'urgent' (bool) flag Returns: - A2A task ID for tracking the request + Status of the A2A message delivery """ - # Validate target agent exists - if target_agent not in ALL_AGENTS: + if not task_id: return format_error_response( - "AGENT_NOT_FOUND", - f"Agent '{target_agent}' not found. " - "Use roboco_agent_discover to find agents.", + "TASK_ID_REQUIRED", + "A2A messages must reference a task. Provide task_id.", + hint="A2A is for communication about existing tasks.", ) - # Validate skill exists for target - target_skills = get_agent_skills(target_agent) - skill_ids = [s.get("id", "") for s in target_skills] - if skill not in skill_ids: - return format_error_response( - "SKILL_NOT_FOUND", - f"Agent '{target_agent}' does not have skill '{skill}'. " - f"Available skills: {', '.join(skill_ids)}", - ) + # Validate permissions (hierarchy enforcement) + validation_error = _validate_a2a_target(agent_id, target_agent, skill) + if validation_error: + return validation_error - # Resolve target agent UUID - target_uuid = AGENT_UUIDS.get(target_agent) - if not target_uuid: - return format_error_response( - "AGENT_UUID_NOT_FOUND", - f"Could not resolve UUID for agent '{target_agent}'", - ) + # Check if we already have a pending A2A to this agent about this task + pending_error = await _check_pending_a2a(agent_id, target_agent, task_id) + if pending_error: + return pending_error - # Build A2A message payload - context_id = task_id or f"request-{agent_id}-to-{target_agent}" - payload = { - "message": { - "role": "user", - "parts": [{"type": "text", "text": message}], - "contextId": context_id, - }, - "configuration": { - "blocking": blocking, - "acceptedOutputModes": ["text/plain", "application/json"], - }, - "metadata": { - "from_agent": agent_id, - "target_agent": target_agent, - "skill": skill, - "task_id": task_id, - }, - } + opts = options or {} + urgent = opts.get("urgent", False) - # Send A2A request - resp = await client.post("/a2a/message/send", json=payload) + # Auto-ack any pending A2A notifications from target about this task + # (responding = acknowledging the original request) + await _auto_ack_a2a_notifications(agent_id, target_agent, task_id) - if not resp.ok: - return format_error_response( - "A2A_REQUEST_FAILED", - f"Failed to send A2A request: {resp.text}", - ) - - result = resp.json() - a2a_task = result.get("task", {}) - a2a_task_id = a2a_task.get("id", "unknown") - a2a_state = a2a_task.get("status", {}).get("state", "submitted") - - return { - "status": "submitted", - "a2a_task_id": a2a_task_id, - "target_agent": target_agent, - "skill": skill, - "state": a2a_state, - "guidance": ( - f"Request sent to {target_agent}. " - f"Task ID: {a2a_task_id}. " - "Use roboco_agent_request_status to check progress, or wait for " - "a notification when complete." - ), - } + return await _send_via_sdk( + target_agent=target_agent, + skill=skill, + message=message, + task_id=task_id, + urgent=urgent, + ) @mcp.tool() - async def roboco_agent_request_status( - a2a_task_id: str, - ) -> dict[str, Any]: + async def roboco_a2a_check() -> dict[str, Any]: """ - Check the status of an A2A request. + Check for incoming A2A messages. - Args: - a2a_task_id: The A2A task ID returned from roboco_agent_request + Poll your inbox for messages from other agents. + Messages are removed from the queue once retrieved. Returns: - Current status and any results + List of pending A2A messages """ - return await _handle_request_status(client, a2a_task_id) + return await _handle_check() return mcp diff --git a/roboco/mcp/tasks/handlers/_helpers.py b/roboco/mcp/tasks/handlers/_helpers.py index 9241f8ae..8b0d98e5 100644 --- a/roboco/mcp/tasks/handlers/_helpers.py +++ b/roboco/mcp/tasks/handlers/_helpers.py @@ -322,21 +322,33 @@ async def validate_task_start( ) if task_status == "claimed" and not task.get("plan"): + task_id = task.get("id", "unknown") return format_error_response( "NO_PLAN", - "Cannot start without a plan.", + "Cannot start without a plan. Workflow: CLAIM → PLAN → START", { - "required_action": "roboco_task_plan(task_id, approach, steps)", - "workflow": "claim → PLAN → start", + "current_status": "claimed", + "next_action": "roboco_task_plan", + "workflow": ["claimed", "plan submitted", "start", "in_progress"], + "why": ( + "Planning ensures you understand the task, break it into steps, " + "identify risks, and ask questions before writing code." + ), "example": { - "task_id": task.get("id"), - "approach": "Describe your implementation approach", - "steps": [ - {"title": "Step 1", "description": "What to do first"}, - {"title": "Step 2", "description": "What to do next"}, - ], + "tool": "roboco_task_plan", + "args": { + "task_id": task_id, + "approach": "Describe your high-level implementation strategy", + "sub_tasks": [ + {"title": "Step 1", "description": "First action to take"}, + {"title": "Step 2", "description": "Second action to take"}, + ], + "risks": ["Potential issue that could block progress"], + "open_questions": ["Question to clarify with PM if any"], + }, }, }, + hint="roboco_kb_search('task plan workflow')", ) if task_status == "claimed": diff --git a/roboco/mcp/tasks/handlers/blocking.py b/roboco/mcp/tasks/handlers/blocking.py index 41676724..97d0169b 100644 --- a/roboco/mcp/tasks/handlers/blocking.py +++ b/roboco/mcp/tasks/handlers/blocking.py @@ -49,6 +49,10 @@ async def handle_task_block( {"status_code": block_resp.status_code, "detail": block_resp.text}, ) + # Determine PM agent based on agent's team + team_prefix = agent_id[:2] if agent_id else "be" + pm_agent = f"{team_prefix}-pm" + return format_task_response( block_resp.json(), "RESOLVE_BLOCKER", @@ -60,6 +64,11 @@ async def handle_task_block( "2. SWITCH - Call roboco_task_scan for other work\n" "3. ESCALATE - Use roboco_task_escalate() if PM is unresponsive\n\n" "You'll be notified when the task is unblocked.", + a2a_suggestion=( + f"Need immediate PM attention? Use A2A:\n" + f"roboco_agent_request('{pm_agent}', 'coordination', " + f"'Blocked: {data.reason}', options={{'urgent': True}})" + ), ) @@ -106,8 +115,20 @@ async def handle_task_unblock( if not unblock_resp.ok: return format_error_response("UNBLOCK_FAILED", "Failed to unblock task") + # Get the assigned developer to notify + assigned_slug = task.get("assigned_to_slug") + return format_task_response( - unblock_resp.json(), "CONTINUE", "Task unblocked. Resume from last checkpoint." + unblock_resp.json(), + "CONTINUE", + "Task unblocked. Resume from last checkpoint.", + a2a_suggestion=( + f"Notify developer that blocker is resolved:\n" + f"roboco_agent_request(target_agent='{assigned_slug or 'developer'}', " + f"skill='resume', message='Blocker resolved for task {task_id}')" + ) + if assigned_slug + else None, ) diff --git a/roboco/mcp/tasks/handlers/lifecycle.py b/roboco/mcp/tasks/handlers/lifecycle.py index 74b4d752..cfd42806 100644 --- a/roboco/mcp/tasks/handlers/lifecycle.py +++ b/roboco/mcp/tasks/handlers/lifecycle.py @@ -78,6 +78,10 @@ async def handle_docs_complete( {"status_code": docs_resp.status_code, "api_error": docs_resp.text}, ) + # Determine PM agent based on documenter's team + team_prefix = agent_id[:2] if agent_id else "be" + pm_agent = f"{team_prefix}-pm" + return format_task_response( docs_resp.json(), "AWAITING_PM", @@ -87,6 +91,11 @@ async def handle_docs_complete( " roboco_kb_index_docs(['/docs/backend/your-doc.md'])\n" "You can still index after submitting - unindexed docs won't be searchable!\n\n" "Call roboco_task_scan for next documentation task.", + a2a_suggestion=( + f"Notify PM that docs are ready:\n" + f"roboco_agent_request(target_agent='{pm_agent}', " + f"skill='coordination', message='Task {task_id} ready for review')" + ), ) @@ -400,11 +409,24 @@ async def handle_ceo_reject( {"status_code": resp.status_code, "api_error": resp.text}, ) + # Get the original developer from quick_context to suggest A2A notification + quick_context = task.get("quick_context", "") + from roboco.services.task import extract_original_developer + + original_dev = extract_original_developer(quick_context) + return format_task_response( resp.json(), "NEEDS_REVISION", f"Task rejected and returned for revision.\nReason: {reason}\n" "The developer will address feedback and resubmit.", + a2a_suggestion=( + f"Notify developer immediately (urgent):\n" + f"roboco_agent_request('{original_dev}', 'revision', " + f"'CEO rejected: {reason[:50]}...', options={{'urgent': True}})" + ) + if original_dev + else None, ) diff --git a/roboco/mcp/tasks/handlers/review.py b/roboco/mcp/tasks/handlers/review.py index ab112d45..5c6b1c1f 100644 --- a/roboco/mcp/tasks/handlers/review.py +++ b/roboco/mcp/tasks/handlers/review.py @@ -134,7 +134,7 @@ async def _validate_qa_submission( async def _save_notes_and_submit( - client: ApiClient, task_id: str, dev_notes: str, handoff_summary: str + client: ApiClient, task_id: str, dev_notes: str, handoff_summary: str, agent_id: str ) -> dict[str, Any]: """Save notes and submit for QA. Returns response dict.""" combined_notes = f"{dev_notes}\n\n---\nHandoff Summary:\n{handoff_summary}" @@ -151,6 +151,10 @@ async def _save_notes_and_submit( if not qa_resp.ok: return format_error_response("SUBMIT_FAILED", "Failed to submit for QA") + # Determine QA agent based on agent's team + team_prefix = agent_id[:2] if agent_id else "be" + qa_agent = f"{team_prefix}-qa" + return format_task_response( qa_resp.json(), "WAIT_FOR_QA", @@ -160,6 +164,11 @@ async def _save_notes_and_submit( "- If PASS: goes to documentation, then PM review\n" "- If FAIL: returns to you with feedback for revision\n\n" "Call roboco_task_scan for other work while waiting.", + a2a_suggestion=( + f"Notify QA directly for faster review:\n" + f"roboco_agent_request(target_agent='{qa_agent}', " + f"skill='qa_review', message='Task {task_id} ready for QA review')" + ), ) @@ -182,7 +191,9 @@ async def handle_task_submit_qa( if error: return error - return await _save_notes_and_submit(client, task_id, dev_notes, handoff_summary) + return await _save_notes_and_submit( + client, task_id, dev_notes, handoff_summary, agent_id + ) def _validate_qa_role(agent_id: str, action: str) -> dict[str, Any] | None: @@ -240,11 +251,20 @@ async def handle_task_qa_pass( {"status_code": pass_resp.status_code, "api_error": pass_resp.text}, ) + # Determine documenter based on QA's team + team_prefix = agent_id[:2] if agent_id else "be" + doc_agent = f"{team_prefix}-doc" + return format_task_response( pass_resp.json(), "NOTIFY_DEV", "Task passed QA. Documenter will be notified.\n" "Call roboco_task_scan for next QA task.", + a2a_suggestion=( + f"Notify documenter directly:\n" + f"roboco_agent_request(target_agent='{doc_agent}', " + f"skill='documentation', message='Task {task_id} ready for docs')" + ), ) @@ -307,9 +327,21 @@ async def handle_task_qa_fail( {"status_code": fail_resp.status_code, "api_error": fail_resp.text}, ) + # Get the original developer from quick_context + quick_context = task.get("quick_context", "") + original_dev = extract_original_developer(quick_context) + issues_summary = "; ".join(issues[:3]) # First 3 issues for message + return format_task_response( fail_resp.json(), "NOTIFY_DEV", f"Task returned for revision with {len(issues)} issue(s).\n" "Developer will be notified.\nCall roboco_task_scan for next QA task.", + a2a_suggestion=( + f"Notify developer immediately (urgent):\n" + f"roboco_agent_request('{original_dev}', 'revision', " + f"'QA failed: {issues_summary}', options={{'urgent': True}})" + ) + if original_dev + else None, ) diff --git a/roboco/mcp/tasks/utils.py b/roboco/mcp/tasks/utils.py index 5ca39f81..a9916842 100644 --- a/roboco/mcp/tasks/utils.py +++ b/roboco/mcp/tasks/utils.py @@ -22,12 +22,20 @@ def format_task_response( next_step: str, guidance: str, project: dict[str, Any] | None = None, + a2a_suggestion: str | None = None, ) -> dict[str, Any]: """ Format a standardized task response with guidance. Includes both JSON task data and TOON-encoded version for token-efficient LLM consumption. + + Args: + task: Task data dictionary + next_step: Next workflow step identifier + guidance: Human-readable guidance text + project: Optional project data + a2a_suggestion: Optional A2A communication suggestion for handoffs """ # Encode task data as TOON for token efficiency when LLM processes response task_toon = _toon.encode(task) @@ -42,6 +50,8 @@ def format_task_response( if project: response["project"] = project response["project_toon"] = _toon.encode(project) + if a2a_suggestion: + response["a2a_suggestion"] = a2a_suggestion return response diff --git a/roboco/models/a2a.py b/roboco/models/a2a.py index f10a0f58..ffe6e855 100644 --- a/roboco/models/a2a.py +++ b/roboco/models/a2a.py @@ -319,6 +319,10 @@ class SendMessageConfiguration(RobocoBase): blocking: bool = Field( default=False, description="Wait for task completion before responding" ) + urgent: bool = Field( + default=False, + description="Priority request - interrupts busy agents, goes to front of queue", + ) push_notification_config: dict[str, Any] | None = Field( default=None, alias="pushNotificationConfig", diff --git a/roboco/models/base.py b/roboco/models/base.py index fdd5ea3a..28c098c0 100644 --- a/roboco/models/base.py +++ b/roboco/models/base.py @@ -151,6 +151,7 @@ class NotificationType(str, Enum): BROADCAST = "broadcast" KNOWLEDGE_SHARE = "knowledge_share" # Cross-agent learning notification MENTION = "mention" # @mention in chat + A2A_REQUEST = "a2a_request" # Agent-to-agent direct request class NotificationPriority(str, Enum): diff --git a/roboco/models/events.py b/roboco/models/events.py index 08771426..cb5a5525 100644 --- a/roboco/models/events.py +++ b/roboco/models/events.py @@ -153,6 +153,12 @@ class NotificationServiceProtocol(Protocol): to_documenter: str, ) -> None: ... + async def send_a2a_notification( + self, + task_id: str, + a2a_context: dict[str, Any], + ) -> None: ... + class OrchestratorAccessProtocol(Protocol): """Protocol for orchestrator access.""" @@ -161,6 +167,10 @@ class OrchestratorAccessProtocol(Protocol): def get_running_agents(self) -> set[str]: ... + def is_agent_busy(self, agent_id: str) -> bool: ... + + def queue_priority_work(self, agent_id: str, work: dict[str, Any]) -> None: ... + async def resolve_wait(self, agent_id: str, resolution: dict[str, Any]) -> Any: ... async def spawn_agent( diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 0c663cca..83dc2b4e 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -16,6 +16,7 @@ import asyncio import contextlib import json import os +import shutil import tempfile from datetime import UTC, datetime from pathlib import Path @@ -144,8 +145,8 @@ class AgentOrchestrator: # Ensure agent image is built await self._ensure_agent_image() - # Ensure agent Claude settings have MCP tools allowed - self._ensure_agent_claude_settings() + # Note: Per-agent settings are now generated at spawn time + # via _generate_agent_settings() - no shared settings needed # Start background tasks self._health_task = asyncio.create_task(self._health_loop()) @@ -182,6 +183,23 @@ class AgentOrchestrator: """Get set of currently running agent IDs.""" return set(self._instances.keys()) + def is_agent_busy(self, agent_id: str) -> bool: + """ + Check if agent has active work. + + An agent is busy if: + 1. They're in the running instances, AND + 2. They have a task claimed/in_progress/verifying + + Note: This is a lightweight check based on instance status. + For full busy detection, the event handler queries the database. + """ + if agent_id not in self._instances: + return False + instance = self._instances[agent_id] + # If agent is running with a task, they're busy + return instance.current_task_id is not None + async def _ensure_agent_image(self, agent_id: str | None = None) -> None: """Ensure the agent Docker images are built. @@ -259,81 +277,201 @@ class AgentOrchestrator: ) 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. + # ========================================================================= + # PER-AGENT SETTINGS GENERATION + # ========================================================================= - This prevents agents from needing interactive permission approval - for essential MCP tools like roboco_agent_idle. + def _get_role_permissions( + self, role: str, workspace_path: str + ) -> dict[str, list[str]]: + """Get role-specific allow/deny lists for Claude Code tools. + + Args: + role: Agent role (developer, qa, documenter, cell_pm, main_pm, etc.) + workspace_path: Path to agent's workspace directory + + Returns: + Dict with 'allow' and 'deny' lists for Claude Code permissions """ - # RoboCo MCP tools that should always be allowed for agents - roboco_allowed_tools = [ - # Task management - always needed + configs: dict[str, dict[str, list[str]]] = { + "developer": { + "allow": [ + "mcp__roboco-git__*", + "mcp__roboco-test__*", + # ONLY allow Write/Edit in their workspace + f"Write(/{workspace_path}/**)", + f"Edit(/{workspace_path}/**)", + ], + "deny": [], + }, + "qa": { + "allow": [ + # QA gets read-only git access + "mcp__roboco-git__roboco_git_status", + "mcp__roboco-git__roboco_git_log", + "mcp__roboco-git__roboco_git_diff", + "mcp__roboco-test__*", + ], + "deny": [ + # QA cannot write anything - review only + "mcp__roboco-git__roboco_git_commit", + "mcp__roboco-git__roboco_git_push", + "mcp__roboco-git__roboco_git_create_pr", + ], + }, + "documenter": { + "allow": [ + "mcp__roboco-docs__*", + "mcp__roboco-git__*", + # Documenters write to docs/ only + "Write(//app/docs/**)", + "Edit(//app/docs/**)", + "Write(//app/CHANGELOG.md)", + "Edit(//app/CHANGELOG.md)", + "Write(//app/README.md)", + "Edit(//app/README.md)", + ], + "deny": [], + }, + "cell_pm": { + "allow": [ + "mcp__roboco-git__*", + "mcp__roboco-docs__*", + ], + "deny": [], + }, + "main_pm": { + "allow": [ + "mcp__roboco-git__*", + "mcp__roboco-docs__*", + ], + "deny": [], + }, + "product_owner": { + "allow": [ + "mcp__roboco-git__*", + "mcp__roboco-docs__*", + ], + "deny": [], + }, + "auditor": { + "allow": [ + # Auditor is read-only observer + "mcp__roboco-git__roboco_git_status", + "mcp__roboco-git__roboco_git_log", + "mcp__roboco-git__roboco_git_diff", + ], + "deny": [], + }, + } + + return configs.get(role, {"allow": [], "deny": []}) + + def _generate_agent_settings( + self, + agent_id: str, + role: str, + workspace_path: str, + ) -> Path: + """Generate per-agent Claude Code settings file with role-specific permissions. + + This replaces the shared settings approach. Each agent gets their own + settings.json with: + - Base MCP tools allowed for all agents + - Role-specific tool permissions + - Explicit deny list blocking native git/file operations + + Args: + agent_id: Agent identifier (e.g., "be-dev-1") + role: Agent role (e.g., "developer") + workspace_path: Path to agent's workspace directory + + Returns: + Path to the generated settings file + """ + # Base MCP tools for all agents + base_allow = [ "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__*", - # Knowledge base/RAG - needed for research "mcp__roboco-optimal__*", - # Git - branch management, commits, PRs - # Role-based permissions enforced at handler level - "mcp__roboco-git__*", - # Agent-to-Agent protocol - cross-cell coordination "mcp__roboco-a2a__*", - # Test tools - run tests, lint, format - "mcp__roboco-test__*", - # Documentation file management - "mcp__roboco-docs__*", - # 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)", + "Read(*)", # All agents can read any file ] - # 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) + # Base denials for all agents - block native tools + base_deny = [ + # Block ALL native git commands - must use roboco_git_* tools + "Bash(git:*)", + # Block file ops outside workspace (role-specific allows override) + "Write(*)", + "Edit(*)", + ] - settings_path = claude_dir / "settings.json" + # Get role-specific permissions + role_config = self._get_role_permissions(role, workspace_path) - # Load existing settings or create new - if settings_path.exists(): - try: - settings = json.loads(settings_path.read_text()) - except json.JSONDecodeError: - settings = {} + # Combine base + role-specific + settings: dict[str, Any] = { + "permissions": { + "allow": base_allow + role_config["allow"], + "deny": base_deny + role_config["deny"], + }, + "hooks": { + # Start SDK server on session start (for A2A communication) + "SessionStart": [ + { + "hooks": [ + { + "type": "command", + "command": "/app/scripts/sdk-startup-hook.sh", + } + ] + } + ], + # Check for incoming A2A messages after each tool use + "PostToolUse": [ + { + "matcher": "*", + "hooks": [ + { + "type": "command", + "command": "/app/scripts/a2a-check-hook.sh", + } + ], + } + ], + }, + } + + # Write to per-agent settings file + # When running in container: write to /app/agent-settings (mounted to host) + # When running on host: use temp directory + if DATA_HOST_PATH: + settings_dir = Path("/app/agent-settings") else: - settings = {} + settings_dir = Path(tempfile.gettempdir()) / "roboco-agent-settings" - # Ensure permissions structure exists - if "permissions" not in settings: - settings["permissions"] = {} - if "allow" not in settings["permissions"]: - settings["permissions"]["allow"] = [] + settings_dir.mkdir(parents=True, exist_ok=True) + settings_path = settings_dir / f"{agent_id}-settings.json" - # 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) + # Handle case where Docker auto-created a directory instead of a file + if settings_path.is_dir(): + shutil.rmtree(settings_path) - # 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, - ) + settings_path.write_text(json.dumps(settings, indent=2)) + + logger.debug( + "Generated per-agent settings", + agent_id=agent_id, + role=role, + settings_path=str(settings_path), + allow_count=len(settings["permissions"]["allow"]), + deny_count=len(settings["permissions"]["deny"]), + ) + + return settings_path # ========================================================================= # AGENT SPAWNING @@ -375,8 +513,23 @@ class AgentOrchestrator: # Generate composed prompt (replaces static blueprints) blueprint_path = self._generate_composed_prompt(agent_id) - # Ensure agent Claude settings have MCP tools allowed - self._ensure_agent_claude_settings() + # Determine role and team for this agent + canonical_role = get_agent_role(agent_id) + team = get_agent_team(agent_id) + + # Determine model using canonical role name + if not model: + model = ROLE_MODEL_MAP.get(canonical_role, "sonnet") + + # Build workspace path for this agent + # Pattern: {workspaces_root}/{project_slug}/{team}/{agent_slug}/ + project_slug = git_context.project_slug if git_context else "default" + workspace_path = f"/data/workspaces/{project_slug}/{team}/{agent_id}" + + # Generate per-agent Claude settings with role-specific permissions + agent_settings_path = self._generate_agent_settings( + agent_id, canonical_role, workspace_path + ) # Ensure agent-specific Docker image is built await self._ensure_agent_image(agent_id) @@ -384,11 +537,6 @@ class AgentOrchestrator: # Generate MCP config with git context if available mcp_config_path = await self._generate_mcp_config(agent_id, git_context) - # 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, @@ -408,9 +556,11 @@ class AgentOrchestrator: self._instances[agent_id] = instance - # Spawn the container + # Spawn the container with per-agent settings try: - container_id = await self._spawn_container(config, initial_prompt) + container_id = await self._spawn_container( + config, initial_prompt, agent_settings_path + ) instance.container_id = container_id instance.state = AgentState.ACTIVE instance.started_at = datetime.now(UTC) @@ -440,8 +590,15 @@ class AgentOrchestrator: self, config: AgentConfig, initial_prompt: str | None = None, + agent_settings_path: Path | None = None, ) -> str: - """Spawn a Docker container for the agent.""" + """Spawn a Docker container for the agent. + + Args: + config: Agent configuration + initial_prompt: Optional initial prompt for the agent + agent_settings_path: Path to per-agent Claude settings file + """ container_name = f"roboco-agent-{config.agent_id}" # Remove existing container if any @@ -465,6 +622,12 @@ class AgentOrchestrator: prompt_host = ( f"{DATA_HOST_PATH}/prompts-generated/{config.agent_id}-prompt.md" ) + # Per-agent settings host path + settings_host = ( + f"{DATA_HOST_PATH}/agent-settings/{config.agent_id}-settings.json" + if agent_settings_path + else None + ) else: # Running directly on host blueprints_host = str(self.blueprints_dir.absolute()) @@ -477,6 +640,8 @@ class AgentOrchestrator: / "roboco-prompts" / f"{config.agent_id}-prompt.md" ) + # Per-agent settings path + settings_host = str(agent_settings_path) if agent_settings_path else None # Build docker run command cmd = [ @@ -487,30 +652,49 @@ class AgentOrchestrator: container_name, "--network", AGENT_NETWORK, - # Mount Claude auth (needs write access for debug logs) + # Mount Claude auth directory (for API keys, etc.) "-v", f"{claude_host}:/home/agent/.claude", - # Mount generated system prompt (composed from layers at runtime) - "-v", - f"{prompt_host}:/app/system-prompt.md:ro", - # Mount blueprints (legacy, kept for reference) - "-v", - f"{blueprints_host}:/app/agents/blueprints:ro", - # Mount docs directory - # - Documenters get write access to create/update docs - # - All other roles get read-only access - "-v", - f"{docs_host}:/app/docs{'' if config.agent_id in ALL_DOCS else ':ro'}", - # 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", ] + # Mount per-agent settings file (overrides shared settings.json) + if settings_host: + cmd.extend( + [ + "-v", + f"{settings_host}:/home/agent/.claude/settings.json:ro", + ] + ) + + cmd.extend( + [ + # Mount generated system prompt (composed from layers at runtime) + "-v", + f"{prompt_host}:/app/system-prompt.md:ro", + # Mount blueprints (legacy, kept for reference) + "-v", + f"{blueprints_host}:/app/agents/blueprints:ro", + # Mount docs directory + # - Documenters get write access to create/update docs + # - All other roles get read-only access + "-v", + f"{docs_host}:/app/docs{'' if config.agent_id in ALL_DOCS else ':ro'}", + # 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", + # SDK Server environment + "-e", + "ROBOCO_SDK_PORT=9000", + "-e", + "ROBOCO_SDK_URL=http://localhost:9000", + ] + ) + # Add git context environment variables if available if config.git_context: if config.git_context.project_slug: @@ -1707,6 +1891,7 @@ Start now: roboco_task_get("{task_id}") await self._dispatch_blocker_work(client) await self._dispatch_escalation_work(client) await self._dispatch_approval_work(client) + await self._dispatch_a2a_work(client) # Scheduled dispatchers await self._dispatch_audit_work(client) @@ -2314,33 +2499,178 @@ Begin with step 1: roboco_task_get("{task_id}") # TODO: Add scheduled periodic audits # Check last audit time, spawn if overdue + async def _dispatch_a2a_work(self, client: httpx.AsyncClient) -> None: + """ + Dispatch A2A (Agent-to-Agent) requests to target agents. + + Monitors: a2a_request notifications (unacknowledged) + Spawns: Any agent that is the target of an A2A request + + This is a fallback mechanism - primary A2A routing happens via events. + If the event-based spawn fails, these notifications will be picked up here. + """ + notifications = await self._fetch_notifications(client, "a2a_request") + + for notif in notifications: + targets = notif.get("to_agents", []) + + for agent_id in targets: + if self._is_agent_active(agent_id): + # Agent is online - SDK handles A2A delivery directly + # No action needed here, SDK server receives messages + continue + + # Agent is offline - spawn them with A2A context + await self.spawn_agent( + agent_id=agent_id, + initial_prompt=self._build_a2a_prompt(notif), + ) + break + # ========================================================================= # SMART DISPATCHER - PROMPT BUILDERS # ========================================================================= + def _get_workflow_state( + self, + status: str, + has_plan: bool, + requires_git: bool, + branch_name: str | None, + ) -> str: + """Determine developer workflow state from task attributes. + + Args: + status: Task status (claimed, in_progress, needs_revision, etc.) + has_plan: Whether task has a plan submitted + requires_git: Whether task requires git workflow + branch_name: Branch name if git task (PM creates this) + + Returns: + Workflow state string (NEEDS_PLAN, READY_TO_START, EXECUTING, etc.) + """ + # Direct status mappings + status_map = { + "in_progress": "EXECUTING", + "needs_revision": "REVISION_REQUIRED", + "verifying": "VERIFYING", + } + + if status in status_map: + return status_map[status] + + # Handle claimed status with sub-states + if status == "claimed": + if not has_plan: + return "NEEDS_PLAN" + if requires_git and not branch_name: + return "WAITING_FOR_BRANCH" + return "READY_TO_START" + + return status.upper() + + def _get_workflow_instructions(self, state: str, task_id: str) -> str: + """Get workflow instructions for the given state. + + Args: + state: Workflow state (NEEDS_PLAN, READY_TO_START, etc.) + task_id: Task ID for tool call examples + + Returns: + Markdown-formatted instructions for the current state + """ + instructions = { + "NEEDS_PLAN": f"""## NEXT STEP: Submit Plan + +You MUST submit a plan before starting work. + +Call roboco_task_plan("{task_id}", {{ + "approach": "Your implementation strategy", + "sub_tasks": [ + {{"title": "Step 1", "description": "First action"}}, + {{"title": "Step 2", "description": "Next action"}} + ], + "risks": ["Potential issues"], + "open_questions": ["Clarifications needed"] +}}) + +You CANNOT call roboco_task_start() until plan is submitted. +""", + "WAITING_FOR_BRANCH": """## BLOCKED: Waiting for Branch + +Your plan is approved, but this is a git task and no branch has been created yet. + +The PM must create a branch for you using: +`roboco_git_create_branch(project_slug, task_id, branch_type)` + +**What to do:** +1. Send a message to your PM requesting branch creation +2. Or escalate: `roboco_task_escalate(task_id, "Need branch created for git task")` +3. Wait for notification that branch is ready + +You CANNOT call roboco_task_start() until branch_name is set on the task. +""", + "READY_TO_START": f"""## NEXT STEP: Start Work + +Your plan is approved. Call roboco_task_start("{task_id}") to begin. + +Then proceed to execute your sub_tasks using roboco_git_* tools. +""", + "EXECUTING": """## IN PROGRESS + +Continue development: +1. Make changes in your workspace +2. roboco_git_commit() for each logical change +3. roboco_task_progress() to update status (0-100%) +4. roboco_journal_* to log decisions/learnings +5. roboco_task_submit_verification() when complete +""", + "REVISION_REQUIRED": f"""## REVISION REQUESTED + +QA or PM requested changes: +1. Call roboco_task_get("{task_id}") to see feedback +2. Call roboco_task_claim("{task_id}") to reclaim +3. Update plan if needed: roboco_task_plan() +4. Call roboco_task_start("{task_id}") to resume +""", + "VERIFYING": """## SELF-VERIFICATION + +Run quality checks and verify against acceptance criteria: +1. Run tests, lint, type checks +2. Review changes with roboco_git_diff() +3. If all good: roboco_task_submit_qa() +4. If issues found: fix and commit +""", + } + return instructions.get( + state, f'Call roboco_task_get("{task_id}") to check status.' + ) + def _build_dev_prompt(self, task: dict[str, Any]) -> str: - """Build initial prompt for a developer with an assigned task.""" + """Build state-aware initial prompt for a developer.""" task_id = task.get("id", "unknown") title = task.get("title", "Untitled") status = task.get("status", "unknown") - team = task.get("team", "unknown") + + # Determine workflow state based on task attributes + has_plan = bool(task.get("plan")) + requires_git = task.get("requires_git", False) + branch_name = task.get("branch_name") + workflow_state = self._get_workflow_state( + status, has_plan, requires_git, branch_name + ) + instructions = self._get_workflow_instructions(workflow_state, task_id) return f"""You have been assigned a development task. TASK ID: {task_id} TITLE: {title} STATUS: {status} -TEAM: {team} +WORKFLOW STATE: {workflow_state} -This task is already CLAIMED for you. Begin work immediately: +{instructions} -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. +Start by calling roboco_task_get("{task_id}") for full details. """ def _build_qa_prompt(self, task: dict[str, Any]) -> str: @@ -2544,4 +2874,35 @@ Your job: 3. Identify any concerns or patterns 4. Compile audit report for CEO 5. Call roboco_agent_idle() when complete +""" + + def _build_a2a_prompt(self, notification: dict[str, Any]) -> str: + """Build initial prompt for handling an A2A (Agent-to-Agent) request.""" + notif_id = notification.get("id", "unknown") + from_agent = notification.get("from_agent", "unknown") + body = notification.get("body", "No message provided") + related_task_id = notification.get("related_task_id") + metadata = notification.get("metadata", {}) + skill = metadata.get("skill", "general") + urgent = metadata.get("urgent", False) + + urgency_note = "**URGENT** - This request has priority.\n\n" if urgent else "" + task_note = f"RELATED TASK: {related_task_id}\n" if related_task_id else "" + + return f"""You have received an A2A (Agent-to-Agent) REQUEST. + +{urgency_note}FROM: {from_agent} +SKILL: {skill} +{task_note} +REQUEST: +{body} + +Your job: + +1. Acknowledge the notification with roboco_notify_ack("{notif_id}") +2. Process the request using your {skill} capabilities +3. Respond to {from_agent} using roboco_agent_request() +4. If you need task context, call roboco_task_get("{related_task_id or "task_id"}") +5. When done, call roboco_task_scan() for other work +6. If no more work, call roboco_agent_idle() to shutdown gracefully """ diff --git a/roboco/services/a2a.py b/roboco/services/a2a.py index e83dba4c..88ac8e48 100644 --- a/roboco/services/a2a.py +++ b/roboco/services/a2a.py @@ -530,48 +530,6 @@ class A2AService: return None - async def route_to_agent( - self, - target_agent_slug: str, - task: TaskTable, - skill: str | None = None, - message: str | None = None, - ) -> None: - """ - Route an A2A task to a specific agent. - - This publishes an event that: - 1. Notifies the agent if they're online (via WebSocket) - 2. Triggers the orchestrator to spawn them if needed - """ - target_uuid = AGENT_UUIDS.get(target_agent_slug) - if not target_uuid: - return - - # Assign task to target agent - task.assigned_to = cast("Any", UUID(target_uuid)) - await self.session.flush() - - # Publish A2A request event for routing - try: - bus = get_event_bus() - if bus.is_connected(): - await bus.publish( - Event( - type=EventType.TASK_ASSIGNED, - data={ - "task_id": str(task.id), - "assigned_to": target_uuid, - "agent_slug": target_agent_slug, - "skill": skill or "general", - "message": message or "", - "source": "a2a", - }, - ) - ) - except Exception: - pass # Don't fail if event bus unavailable - # ========================================================================= # MESSAGE HANDLING # ========================================================================= @@ -627,54 +585,91 @@ class A2AService: ) return result.scalar_one_or_none() - async def create_task_from_a2a_message( + async def create_a2a_notification( self, request: SendMessageRequest, - ) -> TaskTable: + ) -> dict[str, Any]: """ - Create a new task from an A2A message request. + Create an A2A notification for peer-to-peer communication. - Handles the full flow: extract message, resolve target, create task, route. + Does NOT create tasks - A2A is messaging only. + task_id is REQUIRED - A2A is communication about existing tasks. + + Returns dict with notification_id, status, and target_agent. """ + from roboco.services.notification import NotificationService + message = request.message - title, description, message_text = self.extract_message_text(message) - metadata = request.metadata or {} + config = request.configuration + + # task_id is REQUIRED for A2A + task_id = message.task_id + if not task_id: + raise ValueError("A2A requests must reference a task_id") + + from_agent = metadata.get("from_agent") target_agent = self.resolve_target_agent(metadata) - skill = metadata.get("skill") - team = self.get_team_from_agent(target_agent) if target_agent else Team.BACKEND + skill = metadata.get("skill", "general") - creator_agent = await self.resolve_creator_agent(metadata.get("from_agent")) - if creator_agent is None: - raise ValueError("No agent available to create tasks") + # Enforce A2A hierarchy permissions + if from_agent and target_agent: + from roboco.agents_config import can_a2a_direct, get_a2a_route_hint - task = TaskTable( - title=f"[A2A] {title}" if target_agent else title, - description=description, - acceptance_criteria=["Task completed as specified"], - status=TaskStatus.PENDING, - priority=5, - team=team, - created_by=creator_agent.id, - dev_notes=f"A2A Request | Skill: {skill or 'general'}" if skill else None, + allowed, error_msg = can_a2a_direct(from_agent, target_agent) + if not allowed: + hint = get_a2a_route_hint(from_agent, target_agent) + raise ValueError(f"{error_msg} Hint: {hint}") + urgent_from_config = config.urgent if config else False + urgent = urgent_from_config or metadata.get("urgent", False) + + # Extract message content + _, _, message_text = self.extract_message_text(message) + + logger.info( + "Creating A2A notification (fallback)", + task_id=task_id, + from_agent=from_agent, + target_agent=target_agent, + skill=skill, + urgent=urgent, ) - self.session.add(task) - await self.session.flush() - if target_agent: - await self.route_to_agent(target_agent, task, skill, message_text) + # Create notification - orchestrator dispatcher will handle spawning + notification_service = NotificationService() + await notification_service.send_a2a_notification( + task_id=task_id, + a2a_context={ + "from_agent": from_agent or "unknown", + "to_agent": target_agent or "", + "skill": skill, + "message": message_text, + "urgent": urgent, + }, + ) - return task + return { + "status": "sent", + "target_agent": target_agent, + "task_id": task_id, + } async def update_task_from_message( - self, task_id: str, message: A2AMessage + self, + task_id: str, + message: A2AMessage, + responder_agent: str | None = None, ) -> TaskTable: """ - Update an existing task with a new message. + Update an existing task with a new message (response). + + When a response is received, notifies the original requester + and spawns them if offline (bidirectional A2A). Args: task_id: Task UUID string message: A2A message to append + responder_agent: Agent sending the response (for routing back) Returns: Updated TaskTable @@ -696,4 +691,68 @@ class A2AService: raise ValueError(f"Task not found: {task_id}") self.update_task_with_message(task, message) + + # Notify original requester of the response (bidirectional A2A) + await self._notify_original_requester(task, responder_agent) + return task + + async def _notify_original_requester( + self, + task: TaskTable, + responder_agent: str | None = None, + ) -> None: + """ + Notify the original A2A requester of a response. + + If the requester is offline, triggers spawn via event. + This enables bidirectional A2A where both parties can be + spawned as needed until they're both online. + """ + # Extract original requester from dev_notes or task metadata + dev_notes = task.dev_notes or "" + if "A2A Request" not in dev_notes: + return # Not an A2A task + + # The original requester is whoever created the task + created_by = task.created_by + if not created_by: + return + + # Find the agent slug for the creator + from roboco.seeds.initial_data import AGENT_UUIDS + + requester_slug = None + for slug, uuid_str in AGENT_UUIDS.items(): + if uuid_str == str(created_by): + requester_slug = slug + break + + if not requester_slug: + return + + # Don't notify if responder is the same as requester + if responder_agent and responder_agent == requester_slug: + return + + # Publish event to notify/spawn the original requester + try: + bus = get_event_bus() + if bus.is_connected(): + await bus.publish( + Event( + type=EventType.TASK_ASSIGNED, + data={ + "task_id": str(task.id), + "assigned_to": str(created_by), + "agent_slug": requester_slug, + "skill": "a2a_response", + "message": f"Response received for A2A task {task.id}", + "source": "a2a_response", + "urgent": False, + "from_agent": responder_agent or "agent", + }, + ) + ) + except Exception: + pass # Don't fail if event bus unavailable diff --git a/roboco/services/notification.py b/roboco/services/notification.py index 9d7b0482..1b3090d0 100644 --- a/roboco/services/notification.py +++ b/roboco/services/notification.py @@ -6,6 +6,8 @@ Sends notifications through the API with proper enforcement. from __future__ import annotations +from typing import Any + import structlog from roboco.db.base import get_db_context @@ -171,6 +173,53 @@ class NotificationService: ) ) + async def send_a2a_notification( + self, + task_id: str, + a2a_context: dict[str, Any], + ) -> None: + """Send notification for A2A request (when recipient is busy or offline). + + Args: + task_id: Related task ID + a2a_context: Dict with from_agent, to_agent, skill, message, urgent + """ + from_agent = a2a_context.get("from_agent", "unknown") + to_agent = a2a_context.get("to_agent", "") + skill = a2a_context.get("skill", "general") + message = a2a_context.get("message", "") + urgent = a2a_context.get("urgent", False) + + logger.info( + "Sending A2A notification", + task_id=task_id, + from_agent=from_agent, + to_agent=to_agent, + skill=skill, + urgent=urgent, + ) + + urgency_label = "[URGENT] " if urgent else "" + body = ( + f"{urgency_label}A2A request from {from_agent}.\n\n" + f"Skill: {skill}\n\n" + f"Message: {message}" + ) + priority = ( + NotificationPriority.URGENT if urgent else NotificationPriority.NORMAL + ) + await self._create_notification( + CreateNotificationParams( + notification_type=NotificationType.A2A_REQUEST, + priority=priority, + from_agent=from_agent, + to_agents=[to_agent], + subject=f"{urgency_label}A2A: {skill}", + body=body, + related_task_id=task_id, + ) + ) + async def _create_notification(self, params: CreateNotificationParams) -> None: """Create a notification via the database and deliver it.""" async with get_db_context() as db: diff --git a/roboco/services/optimal_brain/indexes/base.py b/roboco/services/optimal_brain/indexes/base.py index 2e75df32..46d99281 100644 --- a/roboco/services/optimal_brain/indexes/base.py +++ b/roboco/services/optimal_brain/indexes/base.py @@ -34,7 +34,7 @@ class IndexConfig: use_hybrid_search: bool = True use_cross_encoder: bool = False embedding_model: str = "embeddinggemma:300m" - llm_model: str = "glm-4.6:cloud" + llm_model: str = "glm-4.7:cloud" llm_base_url: str = "http://roboco-ollama:11434/v1" @classmethod