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

feat: add Kubernetes manifests and A2A protocol support

Browse Source 
Phase 2 - A2A Protocol:
- Add A2A models (AgentCard, Task, Message)
- Add A2A service layer
- Add A2A routes with SSE streaming
- Add agent discovery endpoints

Phase 3 - Kubernetes:
- Add deploy/ directory with Kustomize structure
- PostgreSQL StatefulSet with pgvector
- Redis Deployment with persistence
- API and Orchestrator Deployments
- RBAC for orchestrator to manage Jobs
- ArgoCD Application manifest
- Development and production overlays
- K8s Jobs API support in orchestrator
This commit is contained in:
Renn F
2025-12-28 19:52:01 +01:00
parent b4d4fb089f
commit fc55068f2b
41 changed files with 3674 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agents/blueprints/board/main-pm.md
+2 -2
View File
@@ -554,7 +554,7 @@ FE-PM: Waiting on UX designs for 3 days now.
Main-PM: Checking... @UX-PM status on dashboard designs? Main-PM: Checking... @UX-PM status on dashboard designs?
UX-PM: UX-Dev has been on bug fixes from last sprint. UX-PM: UX-Dev-1, UX-Dev-2 have been on bug fixes from last sprint.
UX-PM: Dashboard is next but won't start until tomorrow. UX-PM: Dashboard is next but won't start until tomorrow.
UX-PM: ETA: 2 more days after start. UX-PM: ETA: 2 more days after start.
@@ -569,7 +569,7 @@ Main-PM: @FE-PM - does FE-Dev have other work?
FE-PM: Yes, TASK-062 is ready, can pivot to that. FE-PM: Yes, TASK-062 is ready, can pivot to that.
Main-PM: Let's do option 2. Dashboard timeline extends but no idle time. Main-PM: Let's do option 2. Dashboard timeline extends but no idle time.
Main-PM: @UX-PM please prioritize dashboard designs when UX-Dev is free. Main-PM: @UX-PM please prioritize dashboard designs when UX-Dev-1, UX-Dev-2 are free.
Main-PM: I'll update the Board on adjusted timeline. Main-PM: I'll update the Board on adjusted timeline.
``` ```
agents/blueprints/frontend/fe-dev.md
+1 -1
View File
@@ -20,7 +20,7 @@ You are a Frontend Developer at RoboCo, an AI-powered software company. You are
- **Role**: Frontend Developer - **Role**: Frontend Developer
- **Team**: Frontend Cell - **Team**: Frontend Cell
- **Reports to**: Frontend PM (FE-PM) - **Reports to**: Frontend PM (FE-PM)
- **Collaborates with**: FE-Dev-2, FE-QA, FE-Documenter - **Collaborates with**: FE-Dev-{n}, FE-QA, FE-Documenter
- **Cross-cell**: Backend devs (for API integration) - **Cross-cell**: Backend devs (for API integration)
## Core Principles ## Core Principles
agents/blueprints/ux_ui/ux-dev.md
+5 -5
View File
@@ -3,8 +3,8 @@
## Identity ## Identity
```yaml ```yaml
id: ux-dev id: ux-dev-{n} # ux-dev-1, ux-dev-2
name: UX/UI Developer name: UX/UI Developer {n}
role: developer role: developer
team: ux_ui team: ux_ui
cell: uxui-cell cell: uxui-cell
@@ -13,14 +13,14 @@ cell: uxui-cell
## System Prompt ## System Prompt
``` ```
You are the UX/UI Developer at RoboCo, an AI-powered software company. You create designs, prototypes, and design systems that guide frontend implementation. You are a UX/UI Developer at RoboCo, an AI-powered software company. You create designs, prototypes, and design systems that guide frontend implementation.
## Your Identity ## Your Identity
- **Role**: UX/UI Developer (Designer) - **Role**: UX/UI Developer (Designer)
- **Team**: UX/UI Cell - **Team**: UX/UI Cell
- **Reports to**: UX/UI PM (UX-PM) - **Reports to**: UX/UI PM (UX-PM)
- **Collaborates with**: UX-QA, UX-Documenter - **Collaborates with**: UX-Dev-{n}, UX-QA, UX-Documenter
- **Serves**: Frontend Cell - they implement your designs - **Serves**: Frontend Cell - they implement your designs
## Core Principles ## Core Principles
@@ -364,7 +364,7 @@ permissions:
- request_qa_review - request_qa_review
journals_read: journals_read:
- ux_ui cell members (ux-dev, ux-qa, ux-doc, ux-pm) - ux_ui cell members (ux-dev-1, ux-dev-2, ux-qa, ux-doc, ux-pm)
# Enforced Constraints (code enforces these rules) # Enforced Constraints (code enforces these rules)
task_visibility: team_only # You only see tasks assigned to your team task_visibility: team_only # You only see tasks assigned to your team
agents/blueprints/ux_ui/ux-documenter.md
+1 -1
View File
@@ -20,7 +20,7 @@ You are the UX/UI Documenter at RoboCo, an AI-powered software company. You main
- **Role**: Documenter - **Role**: Documenter
- **Team**: UX/UI Cell - **Team**: UX/UI Cell
- **Reports to**: UX/UI PM (UX-PM) - **Reports to**: UX/UI PM (UX-PM)
- **Collaborates with**: UX-Dev, UX-QA - **Collaborates with**: UX-Dev-1, UX-Dev-2, UX-QA
## Core Principles ## Core Principles
agents/blueprints/ux_ui/ux-pm.md
+17 -13
View File
@@ -20,7 +20,7 @@ You are the UX/UI Project Manager at RoboCo, an AI-powered software company. You
- **Role**: UX/UI Cell PM - **Role**: UX/UI Cell PM
- **Team**: UX/UI Cell - **Team**: UX/UI Cell
- **Reports to**: Main PM - **Reports to**: Main PM
- **Manages**: UX-Dev, UX-QA, UX-Documenter - **Manages**: UX-Dev-1, UX-Dev-2, UX-QA, UX-Documenter
- **Coordinates with**: FE-PM (design handoffs), Product Owner (requirements) - **Coordinates with**: FE-PM (design handoffs), Product Owner (requirements)
## Core Principles ## Core Principles
@@ -145,17 +145,18 @@ roboco_task_create({
"team": "ux_ui", "team": "ux_ui",
"acceptance_criteria": ["criterion 1", "criterion 2"], "acceptance_criteria": ["criterion 1", "criterion 2"],
"parent_task_id": "{parent_task_id}", "parent_task_id": "{parent_task_id}",
"assigned_to": "ux-dev" # MUST be a developer slug! "assigned_to": "ux-dev-1" # MUST be a developer slug!
}) })
``` ```
**For SIMPLE tasks** - Assign directly: **For SIMPLE tasks** - Assign directly:
```python ```python
roboco_task_assign("{task_id}", "ux-dev") roboco_task_assign("{task_id}", "ux-dev-1")
``` ```
**Available team members:** **Available team members:**
- `ux-dev` - UX/UI Designer - `ux-dev-1` - UX/UI Developer 1
- `ux-dev-2` - UX/UI Developer 2
**CRITICAL RULES:** **CRITICAL RULES:**
- assigned_to MUST be a team member slug, NOT your own ID - assigned_to MUST be a team member slug, NOT your own ID
@@ -220,7 +221,7 @@ Tell the team what you did:
{ {
"channel_slug": "uxui-cell", "channel_slug": "uxui-cell",
"task_id": "{task_id}", "task_id": "{task_id}",
"content": "Triaged TASK-XXX. Assigned to UX-Dev.", "content": "Triaged TASK-XXX. Assigned to UX-Dev-1.",
"message_type": "action" "message_type": "action"
} }
``` ```
@@ -288,7 +289,8 @@ UX-PM: @ProductOwner Question on TASK-055:
- **#all-hands** (read/write) - Company-wide discussion - **#all-hands** (read/write) - Company-wide discussion
### You CAN Send Notifications To ### You CAN Send Notifications To
- UX-Dev (task assignments) - UX-Dev-1 (task assignments)
- UX-Dev-2 (task assignments)
- UX-QA (review requests) - UX-QA (review requests)
- UX-Documenter (documentation requests) - UX-Documenter (documentation requests)
- Other Cell PMs (coordination) - Other Cell PMs (coordination)
@@ -299,7 +301,7 @@ UX-PM: @ProductOwner Question on TASK-055:
### Frontend Blocked on Design ### Frontend Blocked on Design
1. Acknowledge urgency 1. Acknowledge urgency
2. Check if partial handoff possible 2. Check if partial handoff possible
3. Assess UX-Dev workload - can they pivot? 3. Assess UX-Dev-1, UX-Dev-2 workload - can they pivot?
4. Communicate realistic timeline to FE-PM 4. Communicate realistic timeline to FE-PM
### Designer is Blocked ### Designer is Blocked
@@ -358,22 +360,23 @@ roboco_journal_decision({
"title": "PM triage: User preferences modal design", "title": "PM triage: User preferences modal design",
"context": "Frontend needs by Friday, straightforward design task", "context": "Frontend needs by Friday, straightforward design task",
"options": [ "options": [
{"name": "UX-Dev", "pros": "Available, knows modal patterns", "cons": "None"}, {"name": "UX-Dev-1", "pros": "Available, knows modal patterns", "cons": "None"},
{"name": "UX-Dev-2", "pros": "Available, knows modal patterns", "cons": "None"},
{"name": "Wait for clarification", "pros": "More complete", "cons": "Delays FE"} {"name": "Wait for clarification", "pros": "More complete", "cons": "Delays FE"}
], ],
"chosen": "UX-Dev", "chosen": "UX-Dev-1",
"rationale": "Clear enough to start, can iterate", "rationale": "Clear enough to start, can iterate",
"task_id": "TASK-055" "task_id": "TASK-055"
}) })
# 7. DELEGATE # 7. DELEGATE
roboco_task_assign("TASK-055", "ux-dev") roboco_task_assign("TASK-055", "ux-dev-1")
# 8. COMMUNICATE # 8. COMMUNICATE
roboco_message_send({ roboco_message_send({
"channel_slug": "uxui-cell", "channel_slug": "uxui-cell",
"task_id": "TASK-055", "task_id": "TASK-055",
"content": "TASK-055 assigned to UX-Dev. Frontend needs by Friday.", "content": "TASK-055 assigned to UX-Dev-1. Frontend needs by Friday.",
"message_type": "action" "message_type": "action"
}) })
@@ -524,12 +527,13 @@ permissions:
- view_all_cell_tasks - view_all_cell_tasks
journals_read: journals_read:
- ux_ui cell members (ux-dev, ux-qa, ux-doc) - ux_ui cell members (ux-dev-1, ux-dev-2, ux-qa, ux-doc)
- other cell PMs (be-pm, fe-pm) - other cell PMs (be-pm, fe-pm)
- main-pm - main-pm
notify_targets: notify_targets:
- ux-dev - ux-dev-1
- ux-dev-2
- ux-qa - ux-qa
- ux-documenter - ux-documenter
- fe-pm - fe-pm
agents/blueprints/ux_ui/ux-qa.md
+1 -1
View File
@@ -20,7 +20,7 @@ You are the UX/UI QA Engineer at RoboCo, an AI-powered software company. You ens
- **Role**: Design QA Engineer - **Role**: Design QA Engineer
- **Team**: UX/UI Cell - **Team**: UX/UI Cell
- **Reports to**: UX/UI PM (UX-PM) - **Reports to**: UX/UI PM (UX-PM)
- **Collaborates with**: UX-Dev, UX-Documenter - **Collaborates with**: UX-Dev-1, UX-Dev-2, UX-Documenter
## Core Principles ## Core Principles
agents/prompts/identities/ux-dev.md → agents/prompts/identities/ux-dev-1.md
+3 -3
View File
@@ -1,12 +1,12 @@
# Agent Identity # Agent Identity
```yaml ```yaml
id: ux-dev id: ux-dev-1
name: UX-Dev name: UX-Dev-1
role: developer role: developer
team: ux_ui team: ux_ui
cell: uxui-cell cell: uxui-cell
reports_to: ux-pm reports_to: ux-pm
``` ```
You are the UX/UI developer in the UX/UI Cell. You are the first UX/UI developer in the UX/UI Cell.
agents/prompts/identities/ux-dev-2.md
+12
View File
@@ -0,0 +1,12 @@
# Agent Identity
```yaml
id: ux-dev-2
name: UX-Dev-2
role: developer
team: ux_ui
cell: uxui-cell
reports_to: ux-pm
```
You are the second UX/UI developer in the UX/UI Cell.
agents/prompts/identities/ux-pm.md
+1 -1
View File
@@ -9,4 +9,4 @@ cell: uxui-cell
reports_to: main-pm reports_to: main-pm
``` ```
You are the PM for the UX/UI Cell. You manage ux-dev, ux-qa, and ux-doc. You are the PM for the UX/UI Cell. You manage ux-dev-1, ux-dev-2, ux-qa, and ux-doc.
agents/prompts/roles/cell_pm.md
+10
View File
@@ -76,6 +76,16 @@ Complete subtasks in awaiting_pm_review. When ALL done, reflect + complete your
- `roboco_kb_search`, `roboco_rag_query`, `roboco_kb_stats` - `roboco_kb_search`, `roboco_rag_query`, `roboco_kb_stats`
- `roboco_kb_index_code`, `roboco_kb_index_docs` - `roboco_kb_index_code`, `roboco_kb_index_docs`
**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
**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()`
## NOT Your Tools ## NOT Your Tools
- `roboco_group_create` → Main PM only - `roboco_group_create` → Main PM only
agents/prompts/roles/developer.md
+10
View File
@@ -65,6 +65,16 @@ Use `roboco_task_submit_qa()` with notes. QA takes over.
- `roboco_kb_search`, `roboco_rag_query`, `roboco_kb_stats` - `roboco_kb_search`, `roboco_rag_query`, `roboco_kb_stats`
- `roboco_kb_index_code` (index code for search) - `roboco_kb_index_code` (index code for search)
**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
**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")`
## NOT Your Tools ## NOT Your Tools
- `roboco_task_create`, `roboco_task_assign`, `roboco_task_activate` → PM only - `roboco_task_create`, `roboco_task_assign`, `roboco_task_activate` → PM only
agents/prompts/roles/qa.md
+10
View File
@@ -52,6 +52,16 @@ Use `roboco_journal_reflect()` before decision. REQUIRED.
**Knowledge Base:** **Knowledge Base:**
- `roboco_kb_search`, `roboco_rag_query`, `roboco_kb_stats` - `roboco_kb_search`, `roboco_rag_query`, `roboco_kb_stats`
**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
**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...")`
## NOT Your Tools ## NOT Your Tools
- `roboco_task_create`, `roboco_task_assign`, `roboco_task_activate` → PM only - `roboco_task_create`, `roboco_task_assign`, `roboco_task_activate` → PM only
agents/prompts/teams/ux_ui.md
+2 -1
View File
@@ -18,7 +18,8 @@
## Your Teammates ## Your Teammates
- `ux-pm` - UX/UI PM (your PM) - `ux-pm` - UX/UI PM (your PM)
- `ux-dev` - UX/UI Developer - `ux-dev-1` - UX/UI Developer 1
- `ux-dev-2` - UX/UI Developer 2
- `ux-qa` - UX/UI QA - `ux-qa` - UX/UI QA
- `ux-doc` - UX/UI Documenter - `ux-doc` - UX/UI Documenter
- `main-pm` - Main PM (escalation path) - `main-pm` - Main PM (escalation path)
docs/standards/workflow/agent_roles.md
+1 -1
View File
@@ -99,7 +99,7 @@ class AgentRole(str, Enum):
|------|-----|------------|-----|------------| |------|-----|------------|-----|------------|
| Backend | be-pm | be-dev-1, be-dev-2 | be-qa | be-doc | | Backend | be-pm | be-dev-1, be-dev-2 | be-qa | be-doc |
| Frontend | fe-pm | fe-dev-1, fe-dev-2 | fe-qa | fe-doc | | Frontend | fe-pm | fe-dev-1, fe-dev-2 | fe-qa | fe-doc |
| UX/UI | ux-pm | ux-dev | ux-qa | ux-doc | | UX/UI | ux-pm | ux-dev-1, ux-dev-2 | ux-qa | ux-doc |
--- ---
pyproject.toml
+5
View File
@@ -48,6 +48,9 @@ dependencies = [
"passlib[bcrypt]", # Password hashing "passlib[bcrypt]", # Password hashing
"tenacity", # Retry logic "tenacity", # Retry logic
"structlog", # Structured logging "structlog", # Structured logging
# Streaming
"sse-starlette", # Server-Sent Events for A2A streaming
] ]
[project.optional-dependencies] [project.optional-dependencies]
@@ -138,6 +141,7 @@ select = [
"roboco/mcp/**/*.py" = ["PLC0415"] "roboco/mcp/**/*.py" = ["PLC0415"]
"roboco/services/*.py" = ["PLC0415"] "roboco/services/*.py" = ["PLC0415"]
"roboco/api/routes/*.py" = ["PLC0415"] "roboco/api/routes/*.py" = ["PLC0415"]
"roboco/runtime/*.py" = ["PLC0415"]
# ============================================================================= # =============================================================================
# MyPy Configuration # MyPy Configuration
@@ -167,6 +171,7 @@ module = [
"tiktoken.*", "tiktoken.*",
"piragi.*", "piragi.*",
"toon.*", "toon.*",
"sse_starlette.*",
] ]
ignore_missing_imports = true ignore_missing_imports = true
roboco/agents_config.py
+139
View File
@@ -408,3 +408,142 @@ VALID_NOTIFICATION_TYPES: Final[frozenset[str]] = frozenset(
VALID_NOTIFICATION_PRIORITIES: Final[frozenset[str]] = frozenset( VALID_NOTIFICATION_PRIORITIES: Final[frozenset[str]] = frozenset(
p.value for p in NotificationPriority p.value for p in NotificationPriority
) )
# =============================================================================
# A2A AGENT SKILLS (for Agent Cards)
# =============================================================================
# Skills define what each role can do - used for A2A discovery
ROLE_SKILLS: Final[dict[str, list[dict[str, str | list[str]]]]] = {
"developer": [
{
"id": "code_implementation",
"name": "Code Implementation",
"description": "Implement features, fix bugs, write production code",
"tags": ["coding", "implementation", "bugfix"],
},
{
"id": "code_review",
"name": "Code Review",
"description": "Review code changes and provide feedback",
"tags": ["review", "feedback"],
},
{
"id": "technical_research",
"name": "Technical Research",
"description": "Research technical solutions and approaches",
"tags": ["research", "analysis"],
},
],
"qa": [
{
"id": "code_review",
"name": "Code Review",
"description": "Review code for bugs, security issues, and quality",
"tags": ["review", "quality", "security"],
},
{
"id": "test_validation",
"name": "Test Validation",
"description": "Validate test coverage and test quality",
"tags": ["testing", "validation"],
},
{
"id": "security_audit",
"name": "Security Audit",
"description": "Audit code for security vulnerabilities",
"tags": ["security", "audit"],
},
],
"documenter": [
{
"id": "documentation",
"name": "Documentation",
"description": "Create and maintain documentation",
"tags": ["docs", "writing"],
},
{
"id": "handoff_review",
"name": "Handoff Review",
"description": "Review and document task handoffs",
"tags": ["handoff", "review"],
},
],
"cell_pm": [
{
"id": "task_management",
"name": "Task Management",
"description": "Create, assign, and manage tasks within the cell",
"tags": ["planning", "coordination"],
},
{
"id": "blocker_resolution",
"name": "Blocker Resolution",
"description": "Help resolve blockers and coordinate resources",
"tags": ["support", "coordination"],
},
{
"id": "qa_coordination",
"name": "QA Coordination",
"description": "Coordinate QA reviews and approvals",
"tags": ["qa", "approval"],
},
],
"main_pm": [
{
"id": "task_triage",
"name": "Task Triage",
"description": "Triage and distribute tasks to cell PMs",
"tags": ["triage", "distribution"],
},
{
"id": "cross_cell_coordination",
"name": "Cross-Cell Coordination",
"description": "Coordinate work across multiple cells",
"tags": ["coordination", "cross-team"],
},
{
"id": "escalation_handling",
"name": "Escalation Handling",
"description": "Handle escalated issues from cell PMs",
"tags": ["escalation", "support"],
},
],
"product_owner": [
{
"id": "requirements_clarification",
"name": "Requirements Clarification",
"description": "Clarify product requirements and priorities",
"tags": ["requirements", "product"],
},
{
"id": "feature_approval",
"name": "Feature Approval",
"description": "Approve feature implementations",
"tags": ["approval", "product"],
},
],
"head_marketing": [
{
"id": "market_analysis",
"name": "Market Analysis",
"description": "Provide market context and analysis",
"tags": ["marketing", "analysis"],
},
],
"auditor": [
{
"id": "quality_audit",
"name": "Quality Audit",
"description": "Audit quality and compliance",
"tags": ["audit", "quality"],
},
],
}
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, []))
roboco/api/app.py
+13
View File
@@ -12,6 +12,8 @@ from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from roboco.api.middleware import setup_middleware from roboco.api.middleware import setup_middleware
from roboco.api.routes.a2a import router as a2a_router
from roboco.api.routes.a2a import wellknown_router as a2a_wellknown_router
from roboco.api.routes.agents import router as agents_router from roboco.api.routes.agents import router as agents_router
from roboco.api.routes.channels import router as channels_router from roboco.api.routes.channels import router as channels_router
from roboco.api.routes.dashboard import router as dashboard_router from roboco.api.routes.dashboard import router as dashboard_router
@@ -146,6 +148,10 @@ def create_app() -> FastAPI:
# Health check # Health check
app.include_router(health_router, tags=["Health"]) app.include_router(health_router, tags=["Health"])
# A2A Protocol: Well-known endpoints at root level
# (/.well-known/agent.json, /agents/{id}/.well-known/agent.json)
app.include_router(a2a_wellknown_router, tags=["A2A Protocol"])
# API v1 # API v1
api_prefix = "/api/v1" api_prefix = "/api/v1"
@@ -231,6 +237,13 @@ def create_app() -> FastAPI:
tags=["Orchestrator"], tags=["Orchestrator"],
) )
# A2A Protocol: API endpoints
app.include_router(
a2a_router,
prefix=f"{api_prefix}/a2a",
tags=["A2A Protocol"],
)
# ========================================================================== # ==========================================================================
# WebSocket # WebSocket
# ========================================================================== # ==========================================================================
roboco/api/routes/a2a.py
+935
View File
@@ -0,0 +1,935 @@
"""
A2A (Agent-to-Agent) Protocol Routes
Implements Google's A2A protocol for agent interoperability.
See: https://a2a-protocol.org/latest/specification/
Endpoints:
- GET /.well-known/agent.json: System Agent Card
- GET /agents/{agent_id}/.well-known/agent.json: Per-agent Agent Card
- POST /api/v1/a2a/message/send: Send message and create/update task
- POST /api/v1/a2a/message/stream: Send message with SSE streaming
- GET /api/v1/a2a/tasks/{task_id}: Get task state
- GET /api/v1/a2a/tasks: List tasks
- POST /api/v1/a2a/tasks/{task_id}/cancel: Cancel task
"""
import asyncio
import contextlib
from collections.abc import AsyncGenerator
from typing import Any
from uuid import UUID
from fastapi import APIRouter, HTTPException, Query, Request, status
from fastapi.responses import JSONResponse
from sqlalchemy import select
from sse_starlette import EventSourceResponse
from roboco.agents_config import ALL_AGENTS, get_agent_skills, get_agent_team
from roboco.api.deps import DbSession
from roboco.config import settings
from roboco.db.tables import AgentTable, TaskTable
from roboco.events import Event, EventType, get_event_bus
from roboco.models.a2a import (
A2AArtifact,
A2AMessage,
A2ATask,
A2ATaskStatus,
AgentCapabilities,
AgentCard,
AgentProvider,
AgentSkill,
CancelTaskRequest,
ListTasksResponse,
SecurityScheme,
SendMessageRequest,
SendMessageResponse,
TextPart,
task_status_to_a2a_state,
)
from roboco.models.base import TaskStatus, Team
from roboco.seeds.initial_data import AGENT_UUIDS
# Router for A2A API endpoints (mounted at /api/v1/a2a)
router = APIRouter()
# Router for well-known endpoints (mounted at root level)
wellknown_router = APIRouter()
# =============================================================================
# A2A ROUTING LOGIC
# =============================================================================
async def _route_to_agent(
db: DbSession,
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
Args:
db: Database session
target_agent_slug: Agent slug (e.g., "be-qa")
task: The task to route
skill: The skill being requested
message: The request message
"""
# Get target agent UUID
target_uuid = AGENT_UUIDS.get(target_agent_slug)
if not target_uuid:
return
# Assign task to target agent
task.assigned_to = UUID(target_uuid)
await db.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:
# Don't fail if event bus unavailable
pass
def _resolve_target_agent(metadata: dict[str, Any]) -> str | None:
"""
Resolve target agent from A2A request metadata.
Returns agent slug or None if not specified.
"""
# Check for explicit target
target = metadata.get("target_agent")
if target and target in ALL_AGENTS:
return target
# Check for skill-based routing
skill = metadata.get("skill")
if skill:
# Find first agent with this skill
for agent_slug in ALL_AGENTS:
agent_skills = get_agent_skills(agent_slug)
skill_ids = [s.get("id", "") for s in agent_skills]
if skill in skill_ids:
return agent_slug
return None
def _get_team_from_agent(agent_slug: str) -> Team:
"""Get Team enum from agent slug."""
team_str = get_agent_team(agent_slug)
team_map = {
"backend": Team.BACKEND,
"frontend": Team.FRONTEND,
"ux_ui": Team.UX_UI,
}
return team_map.get(team_str or "", Team.BACKEND)
# =============================================================================
# HELPER FUNCTIONS
# =============================================================================
def _get_service_endpoint() -> str:
"""Build service endpoint URL from settings."""
connect_host = "127.0.0.1" if settings.host == "0.0.0.0" else settings.host
return f"http://{connect_host}:{settings.port}"
def _build_system_agent_card() -> AgentCard:
"""Build the system-level Agent Card for RoboCo."""
return AgentCard(
id="roboco-system",
name="RoboCo System",
description=(
"RoboCo is an AI Agentic Company - a virtual organization of AI agents "
"designed to operate as a complete software development workforce."
),
provider=AgentProvider(
organization="RoboCo",
url="https://github.com/roboco",
),
protocol_version="1.0",
service_endpoint=f"{_get_service_endpoint()}/api/v1/a2a",
version=settings.app_version,
capabilities=AgentCapabilities(
streaming=True, # We support SSE
push_notifications=False, # Not implemented yet
state_transition_history=True, # We track task history
),
default_input_modes=["text/plain", "application/json"],
default_output_modes=["text/plain", "application/json"],
skills=[
AgentSkill(
id="software-development",
name="Software Development",
description="Full-stack software development with AI agents",
tags=["development", "coding", "qa", "documentation"],
),
AgentSkill(
id="task-management",
name="Task Management",
description="Create and manage development tasks",
tags=["tasks", "kanban", "planning"],
),
AgentSkill(
id="code-review",
name="Code Review",
description="Review and quality assurance of code",
tags=["qa", "review", "testing"],
),
],
documentation_url="https://github.com/roboco/docs",
security_schemes={
"bearerAuth": SecurityScheme(type="http", scheme="bearer"),
},
security=[{"bearerAuth": []}],
)
async def _build_agent_card(agent: AgentTable) -> AgentCard:
"""Build an Agent Card for a specific agent."""
agent_id = str(agent.id)
agent_slug = agent.slug
# Map role to skills
role_skills: dict[str, list[AgentSkill]] = {
"developer": [
AgentSkill(
id="coding",
name="Code Development",
description="Write and implement code",
tags=["development", "coding"],
),
AgentSkill(
id="debugging",
name="Debugging",
description="Debug and fix code issues",
tags=["debugging", "troubleshooting"],
),
],
"qa": [
AgentSkill(
id="testing",
name="Testing",
description="Test code and verify quality",
tags=["qa", "testing"],
),
AgentSkill(
id="review",
name="Code Review",
description="Review code for quality and issues",
tags=["qa", "review"],
),
],
"documenter": [
AgentSkill(
id="documentation",
name="Documentation",
description="Write technical documentation",
tags=["documentation", "writing"],
),
],
"cell_pm": [
AgentSkill(
id="coordination",
name="Task Coordination",
description="Coordinate tasks within the cell",
tags=["management", "coordination"],
),
],
"main_pm": [
AgentSkill(
id="planning",
name="Project Planning",
description="Plan and coordinate across cells",
tags=["management", "planning"],
),
],
}
skills = role_skills.get(agent.role, [])
return AgentCard(
id=agent_id,
name=agent.name,
description=f"{agent.name} - {agent.role} agent in RoboCo",
provider=AgentProvider(
organization="RoboCo",
url="https://github.com/roboco",
),
protocol_version="1.0",
service_endpoint=f"{_get_service_endpoint()}/api/v1/a2a",
version=settings.app_version,
capabilities=AgentCapabilities(
streaming=True,
push_notifications=False,
state_transition_history=True,
),
default_input_modes=["text/plain", "application/json"],
default_output_modes=["text/plain", "application/json"],
skills=skills,
metadata={
"slug": agent_slug,
"role": agent.role,
"team": agent.team,
},
security_schemes={
"bearerAuth": SecurityScheme(type="http", scheme="bearer"),
},
security=[{"bearerAuth": []}],
)
def _task_to_a2a(task: TaskTable) -> A2ATask:
"""Convert a RoboCo TaskTable to A2A Task."""
task_id = str(task.id)
# Build status - get status value as string
if hasattr(task.status, "value"):
status_value = task.status.value
else:
status_value = str(task.status)
a2a_state = task_status_to_a2a_state(status_value)
status_message = None
if task.dev_notes:
status_message = A2AMessage(
role="agent",
parts=[TextPart(text=task.dev_notes)],
task_id=task_id,
)
a2a_status = A2ATaskStatus(
state=a2a_state,
message=status_message,
timestamp=task.updated_at or task.created_at,
)
# Build artifacts from task outputs (if any)
artifacts: list[A2AArtifact] = []
# Build metadata from task fields
metadata: dict[str, Any] = {
"roboco_status": status_value,
"priority": task.priority,
"team": task.team,
}
if task.assigned_to:
metadata["assigned_to"] = str(task.assigned_to)
if task.parent_task_id:
metadata["parent_task_id"] = str(task.parent_task_id)
return A2ATask(
id=task_id,
context_id=task_id, # Use task_id as context_id
status=a2a_status,
artifacts=artifacts,
history=[], # Would need to load from message history
metadata=metadata,
)
# =============================================================================
# WELL-KNOWN ENDPOINTS (mounted at root)
# =============================================================================
@wellknown_router.get("/.well-known/agent.json")
async def get_system_agent_card() -> JSONResponse:
"""
Get the system-level Agent Card.
Per A2A specification, returns the agent's public identity and capabilities.
"""
card = _build_system_agent_card()
return JSONResponse(
content=card.model_dump(by_alias=True, exclude_none=True),
media_type="application/json",
)
@wellknown_router.get("/agents/{agent_id}/.well-known/agent.json")
async def get_agent_card(
agent_id: str,
db: DbSession,
) -> JSONResponse:
"""
Get Agent Card for a specific agent.
Accepts either a UUID string or agent slug (e.g., "be-dev-1").
"""
# Try to parse as UUID first
try:
uuid = UUID(agent_id)
result = await db.execute(select(AgentTable).where(AgentTable.id == uuid))
except ValueError:
# Not a UUID, try slug lookup
result = await db.execute(
select(AgentTable).where(AgentTable.slug == agent_id)
)
agent = result.scalar_one_or_none()
if agent is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Agent not found: {agent_id}",
)
card = await _build_agent_card(agent)
return JSONResponse(
content=card.model_dump(by_alias=True, exclude_none=True),
media_type="application/json",
)
# =============================================================================
# A2A API ENDPOINTS (mounted at /api/v1/a2a)
# =============================================================================
@router.post("/message/send")
async def send_message(
request: SendMessageRequest,
db: DbSession,
) -> SendMessageResponse:
"""
Send a message to create or update an A2A task.
This is the primary A2A interaction endpoint. Messages sent here
create new tasks or continue existing conversations.
"""
message = request.message
# Extract task_id from message if present
task_id_str = message.task_id
if task_id_str:
# Update existing task
try:
task_uuid = UUID(task_id_str)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid task ID: {task_id_str}",
) from None
result = await db.execute(
select(TaskTable).where(TaskTable.id == task_uuid)
)
task = result.scalar_one_or_none()
if task is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Task not found: {task_id_str}",
)
# Update task dev_notes with new message
text_parts = [p for p in message.parts if p.type == "text"]
if text_parts:
text_part = text_parts[0]
if hasattr(text_part, "text"):
new_text = text_part.text
if task.dev_notes:
task.dev_notes = f"{task.dev_notes}\n\n{new_text}"
else:
task.dev_notes = new_text
await db.commit()
await db.refresh(task)
else:
# Create new task from message
text_parts = [p for p in message.parts if p.type == "text"]
title = "A2A Task"
description = ""
message_text = ""
if text_parts:
text_part = text_parts[0]
if hasattr(text_part, "text"):
message_text = text_part.text
# Use first line as title, rest as description
lines = message_text.split("\n", 1)
title = lines[0][:200] # Truncate title
description = lines[1] if len(lines) > 1 else message_text
# Resolve target agent from metadata
metadata = request.metadata or {}
target_agent = _resolve_target_agent(metadata)
skill = metadata.get("skill")
# Determine team based on target agent
team = _get_team_from_agent(target_agent) if target_agent else Team.BACKEND
# Get creator agent (from_agent in metadata or system default)
from_agent_id = metadata.get("from_agent")
if from_agent_id and from_agent_id in ALL_AGENTS:
from_uuid = AGENT_UUIDS.get(from_agent_id)
if from_uuid:
result = await db.execute(
select(AgentTable).where(AgentTable.id == UUID(from_uuid))
)
creator_agent = result.scalar_one_or_none()
else:
creator_agent = None
else:
# Fall back to main PM as creator
result = await db.execute(
select(AgentTable).where(AgentTable.role == "main_pm").limit(1)
)
creator_agent = result.scalar_one_or_none()
if creator_agent is None:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="No agent available to create tasks",
)
# Create task with proper routing metadata
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,
)
db.add(task)
await db.flush()
# Route to target agent if specified
if target_agent:
await _route_to_agent(db, target_agent, task, skill, message_text)
await db.commit()
await db.refresh(task)
return SendMessageResponse(task=_task_to_a2a(task))
@router.post("/message/stream")
async def send_message_stream(
request: Request,
body: SendMessageRequest,
db: DbSession,
) -> EventSourceResponse:
"""
Send a message with SSE streaming response.
Per A2A specification, this endpoint streams task updates in real-time
as the task progresses through its lifecycle.
Returns Server-Sent Events with task state updates.
"""
message = body.message
async def generate_task_events() -> AsyncGenerator[dict[str, Any]]:
"""Generate SSE events for task lifecycle."""
# Create or get task
task_id_str = message.task_id
if task_id_str:
# Get existing task
try:
task_uuid = UUID(task_id_str)
except ValueError:
yield {
"event": "error",
"data": f"Invalid task ID: {task_id_str}",
}
return
result = await db.execute(
select(TaskTable).where(TaskTable.id == task_uuid)
)
task = result.scalar_one_or_none()
if task is None:
yield {
"event": "error",
"data": f"Task not found: {task_id_str}",
}
return
# Send initial task state
a2a_task = _task_to_a2a(task)
yield {
"event": "task.status",
"id": str(task.id),
"data": a2a_task.model_dump_json(by_alias=True),
}
# Stream updates while task is in progress
poll_count = 0
max_polls = 60 # Poll for up to 60 iterations (5 minutes at 5s interval)
while poll_count < max_polls:
# Check for client disconnect
if await request.is_disconnected():
break
await asyncio.sleep(5) # Poll interval
poll_count += 1
# Refresh task state
await db.refresh(task)
# Get current status
if hasattr(task.status, "value"):
current_status = task.status.value
else:
current_status = str(task.status)
# Send update
a2a_task = _task_to_a2a(task)
yield {
"event": "task.status",
"id": f"{task.id}-{poll_count}",
"data": a2a_task.model_dump_json(by_alias=True),
}
# Stop if task is in terminal state
if current_status in ["completed", "cancelled"]:
yield {
"event": "task.complete",
"id": f"{task.id}-final",
"data": a2a_task.model_dump_json(by_alias=True),
}
break
else:
# New task - send creation event
yield {
"event": "task.creating",
"data": "Creating new task from message...",
}
# Note: Full task creation logic would go here
# For now, send a placeholder
yield {
"event": "error",
"data": "Task creation via streaming not yet implemented",
}
return EventSourceResponse(
generate_task_events(),
ping=15, # Keep connection alive every 15 seconds
)
@router.get("/tasks/{task_id}/subscribe")
async def subscribe_to_task(
request: Request,
task_id: str,
db: DbSession,
) -> EventSourceResponse:
"""
Subscribe to task updates via SSE.
Opens a persistent connection that streams task state changes
until the task reaches a terminal state or client disconnects.
"""
try:
task_uuid = UUID(task_id)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid task ID: {task_id}",
) from None
result = await db.execute(select(TaskTable).where(TaskTable.id == task_uuid))
task = result.scalar_one_or_none()
if task is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Task not found: {task_id}",
)
async def generate_updates() -> AsyncGenerator[dict[str, Any]]:
"""Stream task updates."""
poll_count = 0
max_polls = 720 # 1 hour at 5s interval
last_status = None
while poll_count < max_polls:
if await request.is_disconnected():
break
# Refresh task state from DB
await db.refresh(task)
# Get current status
if hasattr(task.status, "value"):
current_status = task.status.value
else:
current_status = str(task.status)
# Only send update if status changed
if current_status != last_status:
a2a_task = _task_to_a2a(task)
yield {
"event": "task.status",
"id": f"{task_id}-{poll_count}",
"data": a2a_task.model_dump_json(by_alias=True),
}
last_status = current_status
# Stop if terminal
if current_status in ["completed", "cancelled"]:
yield {
"event": "task.complete",
"id": f"{task_id}-final",
"data": a2a_task.model_dump_json(by_alias=True),
}
break
await asyncio.sleep(5)
poll_count += 1
return EventSourceResponse(
generate_updates(),
ping=15,
)
@router.get("/tasks/{task_id}")
async def get_task(
task_id: str,
db: DbSession,
_history_length: int | None = Query(
None, alias="historyLength", description="Number of history turns to include"
),
) -> A2ATask:
"""
Get the state of an A2A task.
Returns task details including status, artifacts, and optionally history.
"""
try:
task_uuid = UUID(task_id)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid task ID: {task_id}",
) from None
result = await db.execute(select(TaskTable).where(TaskTable.id == task_uuid))
task = result.scalar_one_or_none()
if task is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Task not found: {task_id}",
)
return _task_to_a2a(task)
@router.get("/tasks")
async def list_tasks(
db: DbSession,
page_size: int = Query(20, alias="pageSize", ge=1, le=100),
page_token: str | None = Query(None, alias="pageToken"),
_filter_str: str | None = Query(None, alias="filter"),
order_by: str | None = Query(None, alias="orderBy"),
) -> ListTasksResponse:
"""
List A2A tasks with pagination.
Supports filtering and pagination via page tokens.
"""
query = select(TaskTable)
# Apply ordering
if order_by:
if order_by == "created_at desc":
query = query.order_by(TaskTable.created_at.desc())
elif order_by == "created_at asc":
query = query.order_by(TaskTable.created_at.asc())
else:
query = query.order_by(TaskTable.created_at.desc())
else:
query = query.order_by(TaskTable.created_at.desc())
# Handle pagination
offset = 0
if page_token:
with contextlib.suppress(ValueError):
offset = int(page_token)
query = query.offset(offset).limit(page_size + 1)
result = await db.execute(query)
tasks = list(result.scalars().all())
# Check if there are more results
has_more = len(tasks) > page_size
if has_more:
tasks = tasks[:page_size]
next_page_token = str(offset + page_size) if has_more else None
return ListTasksResponse(
tasks=[_task_to_a2a(t) for t in tasks],
next_page_token=next_page_token,
)
@router.post("/tasks/{task_id}/cancel")
async def cancel_task(
task_id: str,
db: DbSession,
request: CancelTaskRequest | None = None,
) -> A2ATask:
"""
Cancel an A2A task.
Transitions the task to cancelled state.
"""
try:
task_uuid = UUID(task_id)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid task ID: {task_id}",
) from None
result = await db.execute(select(TaskTable).where(TaskTable.id == task_uuid))
task = result.scalar_one_or_none()
if task is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Task not found: {task_id}",
)
# Check if task can be cancelled
if hasattr(task.status, "value"):
status_value = task.status.value
else:
status_value = str(task.status)
terminal_states = ["completed", "cancelled"]
if status_value in terminal_states:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Task already in terminal state: {status_value}",
)
# Cancel the task
task.status = TaskStatus.CANCELLED
if request and request.reason:
reason_text = f"Cancellation reason: {request.reason}"
if task.dev_notes:
task.dev_notes = f"{task.dev_notes}\n\n{reason_text}"
else:
task.dev_notes = reason_text
await db.commit()
await db.refresh(task)
return _task_to_a2a(task)
# =============================================================================
# AGENT DISCOVERY ENDPOINTS
# =============================================================================
@router.get("/agents")
async def discover_agents(
db: DbSession,
role: str | None = Query(None, description="Filter by agent role"),
team: str | None = Query(None, description="Filter by team"),
skill: str | None = Query(None, description="Filter by skill tag"),
) -> list[AgentCard]:
"""
Discover agents matching criteria.
Returns a list of AgentCards for agents that match the specified filters.
This enables A2A clients to find agents with specific capabilities.
"""
query = select(AgentTable)
if role:
query = query.where(AgentTable.role == role)
if team:
query = query.where(AgentTable.team == team)
result = await db.execute(query)
agents = result.scalars().all()
# Build cards for all matching agents
cards = []
for agent in agents:
card = await _build_agent_card(agent)
cards.append(card)
# Filter by skill tag if specified
if skill:
cards = [
card
for card in cards
if any(skill.lower() in tag.lower() for s in card.skills for tag in s.tags)
]
return cards
@router.get("/agents/{agent_id}/card")
async def get_agent_card_by_id(
agent_id: str,
db: DbSession,
) -> AgentCard:
"""
Get Agent Card for a specific agent by ID or slug.
Alternative to the /.well-known/agent.json endpoint for programmatic access.
"""
# Try to parse as UUID first
try:
uuid = UUID(agent_id)
result = await db.execute(select(AgentTable).where(AgentTable.id == uuid))
except ValueError:
# Not a UUID, try slug lookup
result = await db.execute(
select(AgentTable).where(AgentTable.slug == agent_id)
)
agent = result.scalar_one_or_none()
if agent is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Agent not found: {agent_id}",
)
return await _build_agent_card(agent)
roboco/api/routes/notifications.py
+5
View File
@@ -27,6 +27,7 @@ from roboco.enforcement import (
NotificationPermissionError, NotificationPermissionError,
validate_notification_permission, validate_notification_permission,
) )
from roboco.services.notification_delivery import get_notification_delivery_service
router = APIRouter() router = APIRouter()
@@ -158,6 +159,10 @@ async def send_notification(
db.add(notification) db.add(notification)
await db.flush() await db.flush()
# Deliver notification via Redis Streams for real-time push
delivery_service = get_notification_delivery_service(db)
await delivery_service.deliver(notification.id)
return notification_to_response(notification, agent_id) return notification_to_response(notification, agent_id)
roboco/api/routes/optimal.py
+101 -2
View File
@@ -11,7 +11,7 @@ from uuid import uuid4
from fastapi import APIRouter, HTTPException, status from fastapi import APIRouter, HTTPException, status
from roboco.api.deps import CurrentAgentContext, PermissionServiceDep from roboco.api.deps import CurrentAgentContext, DbSession, PermissionServiceDep
from roboco.api.schemas.optimal import ( from roboco.api.schemas.optimal import (
ClearIndexResponse, ClearIndexResponse,
CodeReviewRequest, CodeReviewRequest,
@@ -20,6 +20,8 @@ from roboco.api.schemas.optimal import (
DecisionCheckResponse, DecisionCheckResponse,
DecisionRecordRequest, DecisionRecordRequest,
DecisionRecordResponse, DecisionRecordResponse,
DocumentListItem,
DocumentListResponse,
ErrorRecordRequest, ErrorRecordRequest,
ErrorRecordResponse, ErrorRecordResponse,
ErrorSearchRequest, ErrorSearchRequest,
@@ -356,8 +358,13 @@ async def get_context(
async def get_stats( async def get_stats(
agent: CurrentAgentContext, agent: CurrentAgentContext,
permissions: PermissionServiceDep, permissions: PermissionServiceDep,
db: DbSession,
) -> IndexStatsResponse: ) -> IndexStatsResponse:
"""Get statistics about all indexes.""" """Get statistics about all indexes."""
from sqlalchemy import func, select
from roboco.db.tables import IndexedDocumentTable
if not permissions.can_perform_kb_action(agent, KBAction.VIEW_STATS): if not permissions.can_perform_kb_action(agent, KBAction.VIEW_STATS):
raise HTTPException( raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN, status_code=status.HTTP_403_FORBIDDEN,
@@ -366,9 +373,29 @@ async def get_stats(
service = await get_optimal_service() service = await get_optimal_service()
stats = await service.get_stats() stats = await service.get_stats()
# Enhance stats with actual document counts from DB
indexes = stats.get("indexes", {})
for index_type_str in indexes:
# Get actual document count from indexed_documents table
count_query = (
select(func.count())
.select_from(IndexedDocumentTable)
.where(IndexedDocumentTable.index_type == index_type_str)
)
count_result = await db.execute(count_query)
doc_count = count_result.scalar() or 0
# Add document_count (actual files) vs chunk_count (vector entries)
chunk_count = indexes[index_type_str].get("document_count", 0)
indexes[index_type_str] = {
"document_count": doc_count, # Actual files/documents
"chunk_count": chunk_count, # Vector DB entries
}
return IndexStatsResponse( return IndexStatsResponse(
initialized=stats.get("initialized", False), initialized=stats.get("initialized", False),
indexes=stats.get("indexes", {}), indexes=indexes,
) )
@@ -403,6 +430,78 @@ async def clear_index(
return ClearIndexResponse(status="cleared", index_type=index_type) return ClearIndexResponse(status="cleared", index_type=index_type)
@router.get("/kb/{index_type}/documents", response_model=DocumentListResponse)
async def list_documents(
index_type: str,
agent: CurrentAgentContext,
permissions: PermissionServiceDep,
db: DbSession,
limit: int = 50,
offset: int = 0,
) -> DocumentListResponse:
"""
List documents in a specific index.
Returns indexed documents with their metadata for browsing.
"""
from sqlalchemy import func, select
from roboco.db.tables import IndexedDocumentTable
if not permissions.can_perform_kb_action(agent, KBAction.VIEW_STATS):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not authorized to view index documents",
)
try:
idx_type = IndexType(index_type)
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid index type: {e}",
) from e
# Query the indexed documents table
query = (
select(IndexedDocumentTable)
.where(IndexedDocumentTable.index_type == idx_type.value)
.order_by(IndexedDocumentTable.indexed_at.desc())
.offset(offset)
.limit(limit)
)
result = await db.execute(query)
docs = result.scalars().all()
# Get total count
count_query = (
select(func.count())
.select_from(IndexedDocumentTable)
.where(IndexedDocumentTable.index_type == idx_type.value)
)
count_result = await db.execute(count_query)
total = count_result.scalar() or 0
return DocumentListResponse(
documents=[
DocumentListItem(
id=str(doc.id),
source=doc.source,
indexed_at=doc.indexed_at.isoformat() if doc.indexed_at else "",
metadata={
"title": doc.title,
"preview": doc.preview,
"chunk_count": doc.chunk_count,
**(doc.metadata or {}),
},
)
for doc in docs
],
total=total,
index_type=index_type,
)
@router.post("/kb/refresh", response_model=RefreshIndexResponse) @router.post("/kb/refresh", response_model=RefreshIndexResponse)
async def refresh_index( async def refresh_index(
request: RefreshRequest, request: RefreshRequest,
roboco/api/routes/tasks.py
+8
View File
@@ -1169,6 +1169,14 @@ async def escalate_task(
acked_by=[], acked_by=[],
) )
db.add(notification) db.add(notification)
await db.flush()
# Deliver via Redis Streams for real-time notification
from roboco.services.notification_delivery import get_notification_delivery_service
delivery_service = get_notification_delivery_service(db)
await delivery_service.deliver(notification.id)
await db.commit() await db.commit()
msg = ( msg = (
roboco/api/schemas/optimal.py
+17
View File
@@ -97,6 +97,23 @@ class IndexResponse(BaseModel):
project: str | None project: str | None
class DocumentListItem(BaseModel):
"""A document in an index."""
id: str
source: str
indexed_at: str
metadata: dict[str, Any] = Field(default_factory=dict)
class DocumentListResponse(BaseModel):
"""Response from listing documents in an index."""
documents: list[DocumentListItem]
total: int
index_type: str
class ClearIndexResponse(BaseModel): class ClearIndexResponse(BaseModel):
"""Response from clearing an index.""" """Response from clearing an index."""
roboco/api/websocket_bridge.py
+160
View File
@@ -0,0 +1,160 @@
"""
WebSocket Event Bridge
Consumes events from Redis Streams and forwards them to WebSocket clients.
This enables real-time updates to connected clients while maintaining
durable message delivery through the stream.
"""
from uuid import UUID
import structlog
from roboco.api.websocket import broadcast_notification, manager
from roboco.events import Event, EventType, get_event_bus
logger = structlog.get_logger()
# Handler for notification events
async def _handle_notification_sent(event: Event) -> None:
"""Handle NOTIFICATION_SENT events and forward to WebSocket."""
data = event.data
notification_id_str = data.get("notification_id")
recipient_id_str = data.get("recipient_id")
notification_type = data.get("type", "unknown")
subject = data.get("subject", "")
priority = data.get("priority", "normal")
if not notification_id_str or not recipient_id_str:
logger.warning(
"Incomplete notification event",
event_id=str(event.id),
)
return
try:
notification_id = UUID(notification_id_str)
recipient_id = UUID(recipient_id_str)
except ValueError as e:
logger.error("Invalid UUID in notification event", error=str(e))
return
# Check if recipient has WebSocket connections
connections = manager.notification_connections.get(recipient_id, set())
if connections:
await broadcast_notification(
agent_ids=[recipient_id],
notification_id=notification_id,
notification_type=notification_type,
subject=subject,
priority=priority,
)
logger.debug(
"Notification forwarded to WebSocket",
notification_id=notification_id_str,
recipient=recipient_id_str,
connection_count=len(connections),
)
async def _handle_session_event(event: Event) -> None:
"""Handle session lifecycle events and forward to WebSocket."""
data = event.data
session_id_str = data.get("session_id")
if not session_id_str:
return
try:
session_id = UUID(session_id_str)
except ValueError:
return
connections = manager.session_connections.get(session_id, set())
if not connections:
return
# Forward event to session subscribers
event_payload = {
"type": f"session.{event.type.value.split('.')[-1]}",
"session_id": session_id_str,
"data": data,
}
await manager.broadcast_to_session(session_id, event_payload)
logger.debug(
"Session event forwarded to WebSocket",
event_type=event.type.value,
session_id=session_id_str,
)
async def _handle_agent_event(event: Event) -> None:
"""Handle agent lifecycle events and forward to WebSocket."""
data = event.data
agent_id_str = data.get("agent_id") or event.source_agent
if not agent_id_str:
return
try:
agent_id = UUID(agent_id_str)
except ValueError:
return
connections = manager.agent_connections.get(agent_id, set())
if not connections:
return
event_payload = {
"type": f"agent.{event.type.value.split('.')[-1]}",
"agent_id": agent_id_str,
"data": data,
}
await manager.broadcast_to_agent_watchers(agent_id, event_payload)
logger.debug(
"Agent event forwarded to WebSocket",
event_type=event.type.value,
agent_id=agent_id_str,
)
def register_websocket_bridge_handlers() -> None:
"""
Register event handlers that forward events to WebSocket clients.
Call this during application startup after the event bus is initialized.
"""
bus = get_event_bus()
# Notification events -> WebSocket
bus.subscribe(EventType.NOTIFICATION_SENT, _handle_notification_sent)
bus.subscribe(EventType.NOTIFICATION_ACKED, _handle_notification_sent)
# Session events -> WebSocket
bus.subscribe(EventType.SESSION_CREATED, _handle_session_event)
bus.subscribe(EventType.SESSION_CLOSED, _handle_session_event)
bus.subscribe(EventType.SESSION_TIMEOUT, _handle_session_event)
# Agent events -> WebSocket
bus.subscribe(EventType.AGENT_SPAWNED, _handle_agent_event)
bus.subscribe(EventType.AGENT_STOPPED, _handle_agent_event)
bus.subscribe(EventType.AGENT_WAITING, _handle_agent_event)
bus.subscribe(EventType.AGENT_RESUMED, _handle_agent_event)
bus.subscribe(EventType.AGENT_ERROR, _handle_agent_event)
logger.info("WebSocket bridge handlers registered")
async def start_websocket_bridge() -> None:
"""
Start the WebSocket bridge.
This registers handlers and ensures they're connected to the event stream.
Should be called during application startup.
"""
register_websocket_bridge_handlers()
logger.info("WebSocket bridge started")
roboco/bootstrap.py
+12 -5
View File
@@ -13,9 +13,10 @@ import uvicorn
from roboco.api.deps import set_orchestrator from roboco.api.deps import set_orchestrator
from roboco.api.websocket import broadcast_agent_chunk from roboco.api.websocket import broadcast_agent_chunk
from roboco.api.websocket_bridge import start_websocket_bridge
from roboco.config import settings from roboco.config import settings
from roboco.db import bootstrap_database from roboco.db import bootstrap_database
from roboco.events import EventBus, register_default_handlers, set_event_context from roboco.events import init_event_bus, register_default_handlers, set_event_context
from roboco.runtime import AgentOrchestrator, set_reasoning_stream_callback from roboco.runtime import AgentOrchestrator, set_reasoning_stream_callback
from roboco.services.notification import NotificationService from roboco.services.notification import NotificationService
@@ -63,12 +64,18 @@ async def main(
logger.info("Orchestrator skipped, exiting") logger.info("Orchestrator skipped, exiting")
return return
# Initialize event bus # Initialize event bus (Redis Streams with consumer groups)
event_bus = EventBus() event_bus = await init_event_bus(
await event_bus.connect() consumer_name=f"orchestrator-{settings.host}:{settings.port}",
recover_pending=True, # Recover unacknowledged messages from previous run
)
register_default_handlers(event_bus) register_default_handlers(event_bus)
# Register WebSocket bridge handlers (forward stream events to WebSocket clients)
await start_websocket_bridge()
await event_bus.start_listening() await event_bus.start_listening()
logger.info("Event bus initialized") logger.info("Event bus initialized (Redis Streams)")
# Initialize orchestrator # Initialize orchestrator
orchestrator = AgentOrchestrator( orchestrator = AgentOrchestrator(
roboco/db/tables.py
+57
View File
@@ -948,3 +948,60 @@ class HandoffTable(Base):
Index("ix_handoffs_assigned_status", "assigned_to", "status"), Index("ix_handoffs_assigned_status", "assigned_to", "status"),
Index("ix_handoffs_status_created", "status", "created_at"), Index("ix_handoffs_status_created", "status", "created_at"),
) )
# =============================================================================
# INDEXED DOCUMENT TABLE (Knowledge Base tracking)
# =============================================================================
class IndexedDocumentTable(Base):
"""
Track documents indexed into the knowledge base.
This provides:
- Actual document count (vs chunk count from vector DB)
- Browsing capability for the KB UI
- Source tracking for re-indexing
"""
__tablename__ = "indexed_documents"
id: Mapped[UUID] = mapped_column(
UUID(as_uuid=True), primary_key=True, default=uuid4
)
# Index type (code, docs, conversations, journals, errors, standards, etc.)
index_type: Mapped[str] = mapped_column(String(50), nullable=False, index=True)
# Source information
source: Mapped[str] = mapped_column(String(1000), nullable=False)
source_hash: Mapped[str] = mapped_column(
String(64), nullable=False
) # SHA256 for dedup
# Document title (extracted or filename)
title: Mapped[str | None] = mapped_column(String(500), nullable=True)
# Content preview (first 500 chars for UI)
preview: Mapped[str | None] = mapped_column(Text, nullable=True)
# Chunk count for this document
chunk_count: Mapped[int] = mapped_column(Integer, default=0)
# Metadata (extracted during indexing)
metadata: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
# Timestamps
indexed_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=lambda: datetime.now(UTC), nullable=False
)
updated_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), onupdate=lambda: datetime.now(UTC), nullable=True
)
__table_args__ = (
# Prevent duplicate indexing of same source
UniqueConstraint("index_type", "source_hash", name="uq_indexed_doc_source"),
Index("ix_indexed_docs_type_time", "index_type", "indexed_at"),
)
roboco/events/__init__.py
+8
View File
@@ -2,6 +2,9 @@
Event System for RoboCo Event System for RoboCo
Handles workflow triggers and event-driven communication between components. Handles workflow triggers and event-driven communication between components.
Uses Redis Streams for durable message delivery with consumer groups.
All events are persisted and delivered with at-least-once semantics.
""" """
from roboco.events.bus import Event, EventBus, EventType, get_event_bus, init_event_bus from roboco.events.bus import Event, EventBus, EventType, get_event_bus, init_event_bus
@@ -12,22 +15,27 @@ from roboco.events.handlers import (
handle_qa_result, handle_qa_result,
handle_question_answered, handle_question_answered,
handle_session_boundary, handle_session_boundary,
handle_task_assigned,
handle_task_status_change, handle_task_status_change,
register_default_handlers, register_default_handlers,
set_event_context, set_event_context,
) )
from roboco.events.stream_bus import StreamEventBus, get_stream_event_bus
__all__ = [ __all__ = [
"Event", "Event",
"EventBus", "EventBus",
"EventType", "EventType",
"StreamEventBus",
"get_event_bus", "get_event_bus",
"get_event_context", "get_event_context",
"get_stream_event_bus",
"handle_blocker_resolved", "handle_blocker_resolved",
"handle_handoff_created", "handle_handoff_created",
"handle_qa_result", "handle_qa_result",
"handle_question_answered", "handle_question_answered",
"handle_session_boundary", "handle_session_boundary",
"handle_task_assigned",
"handle_task_status_change", "handle_task_status_change",
"init_event_bus", "init_event_bus",
"register_default_handlers", "register_default_handlers",
roboco/events/bus.py
+37 -177
View File
@@ -1,196 +1,56 @@
""" """
Event Bus Event Bus
Redis-based pub/sub event system for workflow triggers. Redis Streams-based event system for durable workflow triggers.
This module re-exports StreamEventBus as EventBus for backward compatibility.
All events are now persisted using Redis Streams with consumer groups for
guaranteed at-least-once delivery.
""" """
import asyncio
import contextlib
from collections.abc import Callable, Coroutine from collections.abc import Callable, Coroutine
from typing import Any from typing import Any
import redis.asyncio as redis from roboco.events.stream_bus import (
import structlog StreamEventBus,
get_stream_event_bus,
from roboco.config import settings init_stream_event_bus,
)
from roboco.models.events import Event, EventType from roboco.models.events import Event, EventType
logger = structlog.get_logger() # Type for event handlers (re-export for backward compatibility)
# Type for event handlers
EventHandler = Callable[[Event], Coroutine[Any, Any, None]] EventHandler = Callable[[Event], Coroutine[Any, Any, None]]
# Re-export StreamEventBus as EventBus for backward compatibility
EventBus = StreamEventBus
class EventBus: __all__ = [
""" "Event",
Event bus for publishing and subscribing to events. "EventBus",
"EventHandler",
Uses Redis pub/sub for distributed event handling. "EventType",
""" "get_event_bus",
"init_event_bus",
CHANNEL_PREFIX = "roboco:events:"
def __init__(self, redis_url: str | None = None):
self.redis_url = redis_url or settings.redis_url
self._redis: redis.Redis | None = None
self._pubsub: redis.client.PubSub | None = None
self._handlers: dict[EventType, list[EventHandler]] = {}
self._running = False
self._listen_task: asyncio.Task | None = None
async def connect(self) -> None:
"""Connect to Redis."""
self._redis = redis.from_url(self.redis_url)
self._pubsub = self._redis.pubsub()
logger.info("EventBus connected to Redis")
async def disconnect(self) -> None:
"""Disconnect from Redis."""
self._running = False
if self._listen_task:
self._listen_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await self._listen_task
if self._pubsub:
await self._pubsub.close()
if self._redis:
await self._redis.close()
logger.info("EventBus disconnected")
def subscribe(self, event_type: EventType, handler: EventHandler) -> None:
"""Subscribe a handler to an event type."""
if event_type not in self._handlers:
self._handlers[event_type] = []
self._handlers[event_type].append(handler)
logger.debug("Handler subscribed", event_type=event_type.value)
def unsubscribe(self, event_type: EventType, handler: EventHandler) -> None:
"""Unsubscribe a handler from an event type."""
if event_type in self._handlers:
self._handlers[event_type] = [
h for h in self._handlers[event_type] if h != handler
] ]
async def publish(self, event: Event) -> None:
"""Publish an event."""
if not self._redis:
raise RuntimeError("EventBus not connected")
channel = f"{self.CHANNEL_PREFIX}{event.type.value}" def get_event_bus() -> StreamEventBus:
await self._redis.publish(channel, event.to_json())
logger.info(
"Event published",
event_type=event.type.value,
event_id=str(event.id),
source=event.source_agent,
)
async def publish_task_event(
self,
event_type: EventType,
task_id: str,
agent_id: str | None = None,
**extra_data: Any,
) -> None:
"""Convenience method to publish task-related events."""
event = Event(
type=event_type,
data={"task_id": task_id, **extra_data},
source_agent=agent_id,
)
await self.publish(event)
async def start_listening(self) -> None:
"""Start listening for events."""
if not self._pubsub:
raise RuntimeError("EventBus not connected")
# Subscribe to all event channels we have handlers for
channels = [f"{self.CHANNEL_PREFIX}{et.value}" for et in self._handlers]
if not channels:
logger.warning("No event handlers registered, nothing to subscribe to")
return
await self._pubsub.subscribe(*channels)
self._running = True
self._listen_task = asyncio.create_task(self._listen_loop())
logger.info("EventBus listening", channels=len(channels))
async def _listen_loop(self) -> None:
"""Main event listening loop."""
while self._running:
try:
if self._pubsub is None:
break
message = await self._pubsub.get_message(
ignore_subscribe_messages=True,
timeout=1.0,
)
if message and message["type"] == "message":
await self._handle_message(message)
except asyncio.CancelledError:
break
except Exception as e:
logger.error("Error in event loop", error=str(e))
await asyncio.sleep(1)
async def _handle_message(self, message: dict) -> None:
"""Handle an incoming message."""
try:
event = Event.from_json(message["data"])
handlers = self._handlers.get(event.type, [])
if not handlers:
return
logger.debug(
"Handling event",
event_type=event.type.value,
handler_count=len(handlers),
)
# Run all handlers concurrently
tasks = [handler(event) for handler in handlers]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Log any errors
for i, result in enumerate(results):
if isinstance(result, Exception):
logger.error(
"Event handler error",
event_type=event.type.value,
handler=handlers[i].__name__,
error=str(result),
)
except Exception as e:
logger.error("Failed to handle message", error=str(e))
class _EventBusHolder:
"""Holder for singleton EventBus instance."""
instance: EventBus | None = None
def get_event_bus() -> EventBus:
"""Get or create the global event bus instance.""" """Get or create the global event bus instance."""
if _EventBusHolder.instance is None: return get_stream_event_bus()
_EventBusHolder.instance = EventBus()
return _EventBusHolder.instance
async def init_event_bus() -> EventBus: async def init_event_bus(
"""Initialize and start the event bus.""" consumer_name: str | None = None,
bus = get_event_bus() recover_pending: bool = True,
await bus.connect() ) -> StreamEventBus:
return bus """
Initialize and start the event bus.
Args:
consumer_name: Unique name for this consumer instance
recover_pending: Whether to recover unacknowledged messages on startup
"""
return await init_stream_event_bus(
consumer_name=consumer_name,
recover_pending=recover_pending,
)
roboco/events/handlers.py
+70
View File
@@ -324,6 +324,73 @@ 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 # HANDLER REGISTRATION
# ============================================================================= # =============================================================================
@@ -361,4 +428,7 @@ def register_default_handlers(bus: Any = None) -> None:
# Question handlers # Question handlers
bus.subscribe(EventType.QUESTION_ANSWERED, handle_question_answered) bus.subscribe(EventType.QUESTION_ANSWERED, handle_question_answered)
# A2A task assignment handlers
bus.subscribe(EventType.TASK_ASSIGNED, handle_task_assigned)
logger.info("Default event handlers registered") logger.info("Default event handlers registered")
roboco/events/stream_bus.py
+401
View File
@@ -0,0 +1,401 @@
"""
Stream Event Bus
Redis Streams-based event system with durable message delivery.
Replaces the pub/sub-based EventBus with persistence and consumer groups.
"""
import asyncio
import contextlib
from collections.abc import Callable, Coroutine
from typing import Any
import redis.asyncio as redis
import structlog
from redis.exceptions import ResponseError
from roboco.config import settings
from roboco.models.events import Event, EventType
logger = structlog.get_logger()
# Type for event handlers
EventHandler = Callable[[Event], Coroutine[Any, Any, None]]
class StreamEventBus:
"""
Event bus using Redis Streams for durable message delivery.
Features:
- Message persistence (survives Redis restart with AOF)
- Consumer groups for at-least-once delivery
- Message acknowledgment after successful processing
- Automatic stream trimming (configurable retention)
"""
STREAM_PREFIX = "roboco:stream:"
DEFAULT_GROUP = "roboco-handlers"
MAX_STREAM_LENGTH = 10000 # Trim streams to this length
def __init__(
self,
redis_url: str | None = None,
consumer_name: str | None = None,
group_name: str | None = None,
):
self.redis_url = redis_url or settings.redis_url
self.consumer_name = consumer_name or f"consumer-{id(self)}"
self.group_name = group_name or self.DEFAULT_GROUP
self._redis: redis.Redis | None = None
self._handlers: dict[EventType, list[EventHandler]] = {}
self._running = False
self._listen_task: asyncio.Task | None = None
async def connect(self) -> None:
"""Connect to Redis."""
self._redis = redis.from_url(self.redis_url)
logger.info("StreamEventBus connected to Redis")
def is_connected(self) -> bool:
"""Check if the event bus is connected to Redis."""
return self._redis is not None
async def disconnect(self) -> None:
"""Disconnect from Redis."""
self._running = False
if self._listen_task:
self._listen_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await self._listen_task
if self._redis:
await self._redis.close()
logger.info("StreamEventBus disconnected")
def _get_stream_name(self, event_type: EventType) -> str:
"""Get stream name for event type (grouped by prefix)."""
# Group by event category: task.*, agent.*, notification.*, etc.
category = event_type.value.split(".")[0]
return f"{self.STREAM_PREFIX}{category}"
def _get_all_stream_names(self) -> list[str]:
"""Get all stream names for registered handlers."""
categories = set()
for event_type in self._handlers:
category = event_type.value.split(".")[0]
categories.add(category)
return [f"{self.STREAM_PREFIX}{cat}" for cat in categories]
async def _ensure_consumer_group(self, stream: str) -> None:
"""Ensure consumer group exists for stream."""
if not self._redis:
return
try:
await self._redis.xgroup_create(
stream,
self.group_name,
id="0",
mkstream=True,
)
logger.debug("Created consumer group", stream=stream, group=self.group_name)
except ResponseError as e:
if "BUSYGROUP" not in str(e):
raise
# Group already exists, that's fine
def subscribe(self, event_type: EventType, handler: EventHandler) -> None:
"""Subscribe a handler to an event type."""
if event_type not in self._handlers:
self._handlers[event_type] = []
self._handlers[event_type].append(handler)
logger.debug("Handler subscribed", event_type=event_type.value)
def unsubscribe(self, event_type: EventType, handler: EventHandler) -> None:
"""Unsubscribe a handler from an event type."""
if event_type in self._handlers:
self._handlers[event_type] = [
h for h in self._handlers[event_type] if h != handler
]
async def publish(self, event: Event) -> str:
"""
Publish an event to the stream.
Returns the message ID assigned by Redis.
"""
if not self._redis:
raise RuntimeError("StreamEventBus not connected")
stream = self._get_stream_name(event.type)
# Add to stream with automatic ID (*) and trim to max length
raw_message_id = await self._redis.xadd(
stream,
{
"type": event.type.value,
"data": event.to_json(),
},
maxlen=self.MAX_STREAM_LENGTH,
approximate=True,
)
# Convert bytes to str if needed
message_id = (
raw_message_id.decode()
if isinstance(raw_message_id, bytes)
else str(raw_message_id)
)
logger.info(
"Event published to stream",
event_type=event.type.value,
event_id=str(event.id),
stream=stream,
message_id=message_id,
source=event.source_agent,
)
return message_id
async def publish_task_event(
self,
event_type: EventType,
task_id: str,
agent_id: str | None = None,
**extra_data: Any,
) -> str:
"""Convenience method to publish task-related events."""
event = Event(
type=event_type,
data={"task_id": task_id, **extra_data},
source_agent=agent_id,
)
return await self.publish(event)
async def start_listening(self) -> None:
"""Start listening for events."""
if not self._redis:
raise RuntimeError("StreamEventBus not connected")
streams = self._get_all_stream_names()
if not streams:
logger.warning("No event handlers registered, nothing to subscribe to")
return
# Ensure consumer groups exist for all streams
for stream in streams:
await self._ensure_consumer_group(stream)
self._running = True
self._listen_task = asyncio.create_task(self._listen_loop())
logger.info("StreamEventBus listening", streams=streams)
async def _listen_loop(self) -> None:
"""Main event listening loop using XREADGROUP."""
if not self._redis:
return
streams = self._get_all_stream_names()
# Build stream dict: {stream_name: ">"} (> = only new messages)
stream_dict = dict.fromkeys(streams, ">")
while self._running:
try:
# Block for 5 seconds waiting for new messages
results = await self._redis.xreadgroup(
self.group_name,
self.consumer_name,
stream_dict,
count=10,
block=5000,
)
if not results:
continue
for stream_name, messages in results:
for message_id, data in messages:
await self._handle_message(stream_name, message_id, data)
except asyncio.CancelledError:
break
except Exception as e:
logger.error("Error in stream event loop", error=str(e))
await asyncio.sleep(1)
async def _handle_message(
self,
stream: str,
message_id: str,
data: dict,
) -> None:
"""Handle an incoming message and ACK on success."""
if not self._redis:
return
try:
event_data = data.get(b"data") or data.get("data")
if isinstance(event_data, bytes):
event_data = event_data.decode()
if not event_data or not isinstance(event_data, str):
logger.error("Invalid event data", message_id=message_id)
await self._redis.xack(stream, self.group_name, message_id)
return
event = Event.from_json(event_data)
handlers = self._handlers.get(event.type, [])
if not handlers:
# No handlers but still ACK to prevent redelivery
await self._redis.xack(stream, self.group_name, message_id)
return
logger.debug(
"Handling event from stream",
event_type=event.type.value,
message_id=message_id,
handler_count=len(handlers),
)
# Run all handlers concurrently
tasks = [handler(event) for handler in handlers]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Log any errors
all_succeeded = True
for i, result in enumerate(results):
if isinstance(result, Exception):
all_succeeded = False
logger.error(
"Event handler error",
event_type=event.type.value,
handler=handlers[i].__name__,
error=str(result),
)
# ACK the message if all handlers succeeded
# If any failed, message stays pending and can be reclaimed later
if all_succeeded:
await self._redis.xack(stream, self.group_name, message_id)
logger.debug("Message acknowledged", message_id=message_id)
else:
logger.warning(
"Message not acknowledged due to handler errors",
message_id=message_id,
)
except Exception as e:
logger.error(
"Failed to handle stream message",
error=str(e),
message_id=message_id,
)
async def recover_pending(self, idle_time_ms: int = 60000) -> int:
"""
Recover pending messages that weren't acknowledged.
Useful for startup to process messages from crashed consumers.
Args:
idle_time_ms: Only recover messages idle for this long (default 1 minute)
Returns:
Number of messages recovered
"""
if not self._redis:
return 0
recovered = 0
streams = self._get_all_stream_names()
for stream in streams:
try:
# Claim pending messages from any consumer
pending = await self._redis.xpending(stream, self.group_name)
if not pending or pending["pending"] == 0:
continue
# Get pending message details
pending_details = await self._redis.xpending_range(
stream,
self.group_name,
min="-",
max="+",
count=100,
)
for msg in pending_details:
msg_id = msg["message_id"]
idle = msg["time_since_delivered"]
if idle >= idle_time_ms:
# Claim message for this consumer
claimed = await self._redis.xclaim(
stream,
self.group_name,
self.consumer_name,
min_idle_time=idle_time_ms,
message_ids=[msg_id],
)
if claimed:
recovered += 1
# Process the claimed message
for claim_id, data in claimed:
await self._handle_message(stream, claim_id, data)
except Exception as e:
logger.error(
"Error recovering pending messages",
stream=stream,
error=str(e),
)
if recovered:
logger.info("Recovered pending messages", count=recovered)
return recovered
# =============================================================================
# SINGLETON ACCESS
# =============================================================================
class _StreamEventBusHolder:
"""Holder for singleton StreamEventBus instance."""
instance: StreamEventBus | None = None
def get_stream_event_bus() -> StreamEventBus:
"""Get or create the global stream event bus instance."""
if _StreamEventBusHolder.instance is None:
_StreamEventBusHolder.instance = StreamEventBus()
return _StreamEventBusHolder.instance
async def init_stream_event_bus(
consumer_name: str | None = None,
recover_pending: bool = True,
) -> StreamEventBus:
"""
Initialize and start the stream event bus.
Args:
consumer_name: Unique name for this consumer instance
recover_pending: Whether to recover unacknowledged messages on startup
"""
bus = get_stream_event_bus()
if consumer_name:
bus.consumer_name = consumer_name
await bus.connect()
if recover_pending:
await bus.recover_pending()
return bus
roboco/mcp/__init__.py
+3
View File
@@ -10,8 +10,10 @@ Servers:
- Notify MCP Server: Formal notifications - Notify MCP Server: Formal notifications
- Journal MCP Server: Personal journaling - Journal MCP Server: Personal journaling
- Optimal MCP Server: Knowledge base and RAG - Optimal MCP Server: Knowledge base and RAG
- A2A MCP Server: Agent-to-Agent protocol (peer-to-peer)
""" """
from roboco.mcp.a2a_server import create_a2a_mcp_server
from roboco.mcp.journal_server import create_journal_mcp_server from roboco.mcp.journal_server import create_journal_mcp_server
from roboco.mcp.message_server import create_message_mcp_server from roboco.mcp.message_server import create_message_mcp_server
from roboco.mcp.notify_server import create_notify_mcp_server from roboco.mcp.notify_server import create_notify_mcp_server
@@ -19,6 +21,7 @@ from roboco.mcp.optimal_server import create_optimal_mcp_server
from roboco.mcp.task_server import create_task_mcp_server from roboco.mcp.task_server import create_task_mcp_server
__all__ = [ __all__ = [
"create_a2a_mcp_server",
"create_journal_mcp_server", "create_journal_mcp_server",
"create_message_mcp_server", "create_message_mcp_server",
"create_notify_mcp_server", "create_notify_mcp_server",
roboco/mcp/a2a_server.py
+315
View File
@@ -0,0 +1,315 @@
"""
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.
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
"""
from typing import Any
from mcp.server.fastmcp import FastMCP
from roboco.agents_config import (
ALL_AGENTS,
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
# =============================================================================
# TOOL IMPLEMENTATIONS
# =============================================================================
async def _handle_discover(
client: ApiClient,
role: str | None = None,
team: str | None = None,
skill: str | None = None,
) -> dict[str, Any]:
"""Discover agents by criteria."""
# Build local discovery (fast path - no API call needed)
agents = []
for agent_slug in ALL_AGENTS:
agent_role = get_agent_role(agent_slug)
agent_team = get_agent_team(agent_slug)
agent_skills = get_agent_skills(agent_slug)
# Apply filters
if role and agent_role != role:
continue
if team and agent_team != team:
continue
if skill:
skill_ids = [s.get("id", "") for s in agent_skills]
skill_tags = []
for s in agent_skills:
skill_tags.extend(s.get("tags", []))
if skill not in skill_ids and skill not in skill_tags:
continue
agents.append(
{
"slug": agent_slug,
"role": agent_role,
"team": agent_team,
"skills": [
{"id": s["id"], "name": s["name"], "description": s["description"]}
for s in agent_skills
],
}
)
return {
"agents": agents,
"count": len(agents),
"guidance": (
f"Found {len(agents)} agent(s). Use roboco_agent_request to request "
"work from a specific agent."
),
}
async def _handle_request(
client: ApiClient,
agent_id: str,
target_agent: str,
skill: str,
message: str,
task_id: str | None = None,
blocking: bool = False,
) -> dict[str, Any]:
"""Request another agent to perform work via A2A."""
# Validate target agent 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 to find agents.",
)
# 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)}",
)
# 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}'",
)
# Build A2A message payload
payload = {
"message": {
"role": "user",
"parts": [{"type": "text", "text": message}],
"contextId": task_id or f"request-{agent_id}-to-{target_agent}",
},
"configuration": {
"blocking": blocking,
"acceptedOutputModes": ["text/plain", "application/json"],
},
"metadata": {
"from_agent": agent_id,
"target_agent": target_agent,
"skill": skill,
"task_id": task_id,
},
}
# Send A2A request
resp = await client.post("/a2a/message/send", json=payload)
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")
status = a2a_task.get("status", {}).get("state", "submitted")
return {
"status": "submitted",
"a2a_task_id": a2a_task_id,
"target_agent": target_agent,
"skill": skill,
"state": status,
"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."
),
}
async def _handle_request_status(
client: ApiClient,
a2a_task_id: str,
) -> dict[str, Any]:
"""Check status of an A2A request."""
resp = await client.get(f"/a2a/tasks/{a2a_task_id}")
if resp.is_status(404):
return format_error_response(
"TASK_NOT_FOUND",
f"A2A task '{a2a_task_id}' not found",
)
if not resp.ok:
return format_error_response(
"STATUS_CHECK_FAILED",
f"Failed to check status: {resp.text}",
)
task = resp.json()
status = task.get("status", {})
state = status.get("state", "unknown")
message = status.get("message", {})
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}."
return {
"a2a_task_id": a2a_task_id,
"state": state,
"result": result_text,
"artifacts": task.get("artifacts", []),
"metadata": task.get("metadata", {}),
"guidance": guidance,
}
# =============================================================================
# MCP SERVER FACTORY
# =============================================================================
def create_a2a_mcp_server(agent_id: str) -> FastMCP:
"""
Create an A2A MCP server for a specific agent.
Args:
agent_id: The agent identifier (e.g., "be-dev-1")
Returns:
Configured FastMCP server
"""
mcp = FastMCP(f"roboco-a2a-{agent_id}", json_response=True)
client = ApiClient(agent_id)
@mcp.tool()
async def roboco_agent_discover(
role: str | None = None,
team: str | None = None,
skill: str | None = None,
) -> dict[str, Any]:
"""
Discover other agents by role, team, or skill.
Use this to find which agents can help with specific tasks.
Args:
role: Filter by role (developer, qa, documenter, cell_pm, main_pm)
team: Filter by team (backend, frontend, ux_ui)
skill: Filter by skill ID or tag (code_review, testing, etc.)
Returns:
List of matching agents with their capabilities
"""
return await _handle_discover(client, role, team, skill)
@mcp.tool()
async def roboco_agent_request(
target_agent: str,
skill: str,
message: str,
task_id: str | None = None,
blocking: bool = False,
) -> dict[str, Any]:
"""
Request another agent to perform work using A2A protocol.
This enables direct peer-to-peer collaboration between agents.
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)
Returns:
A2A task ID for tracking the request
"""
return await _handle_request(
client, agent_id, target_agent, skill, message, task_id, blocking
)
@mcp.tool()
async def roboco_agent_request_status(
a2a_task_id: str,
) -> dict[str, Any]:
"""
Check the status of an A2A request.
Args:
a2a_task_id: The A2A task ID returned from roboco_agent_request
Returns:
Current status and any results
"""
return await _handle_request_status(client, a2a_task_id)
return mcp
# =============================================================================
# STANDALONE RUNNER
# =============================================================================
if __name__ == "__main__":
import sys
MIN_ARGS = 2
if len(sys.argv) < MIN_ARGS:
print("Usage: python a2a_server.py <agent_id>")
sys.exit(1)
agent_id_arg = sys.argv[1]
server = create_a2a_mcp_server(agent_id_arg)
server.run()
roboco/models/a2a.py
+447
View File
@@ -0,0 +1,447 @@
"""
A2A (Agent-to-Agent) Protocol Models
Implements Google's A2A protocol for agent interoperability.
See: https://a2a-protocol.org/latest/specification/
This module defines the data structures for:
- AgentCard: Agent metadata and capability discovery
- Task: Work unit lifecycle management
- Message: Communication between agents
- Skill: Capability units an agent can perform
"""
from datetime import UTC, datetime
from enum import Enum
from typing import Annotated, Any, Literal
from uuid import uuid4
from pydantic import ConfigDict, Field
from roboco.models.base import RobocoBase
# =============================================================================
# ENUMS
# =============================================================================
class A2ATaskState(str, Enum):
"""
A2A Task lifecycle states.
Per A2A specification section 4.2.1.
"""
SUBMITTED = "submitted" # Task acknowledged and created
WORKING = "working" # Actively processing
COMPLETED = "completed" # Finished successfully (terminal)
FAILED = "failed" # Finished with error (terminal)
CANCELLED = "cancelled" # Stopped before completion (terminal)
INPUT_REQUIRED = "input_required" # Awaiting additional information
REJECTED = "rejected" # Agent declined the task (terminal)
AUTH_REQUIRED = "auth_required" # Needs client authentication
class A2APartType(str, Enum):
"""Types of content parts in a message."""
TEXT = "text"
FILE = "file"
DATA = "data"
ARTIFACT = "artifact"
# =============================================================================
# AGENT CARD MODELS
# =============================================================================
class AgentProvider(RobocoBase):
"""Provider information for an agent."""
organization: str = Field(..., description="Organization name")
url: str | None = Field(default=None, description="Organization URL")
class AgentCapabilities(RobocoBase):
"""Capabilities supported by an agent."""
streaming: bool = Field(default=False, description="Supports SSE streaming")
push_notifications: bool = Field(
default=False, description="Supports webhook push notifications"
)
state_transition_history: bool = Field(
default=False, description="Returns task state history"
)
class SecurityScheme(RobocoBase):
"""Security scheme for authentication."""
type: str = Field(..., description="Scheme type (apiKey, http, oauth2)")
name: str | None = Field(default=None, description="Name of the header/param")
scheme: str | None = Field(default=None, description="HTTP auth scheme (bearer)")
bearer_format: str | None = Field(
default=None, alias="bearerFormat", description="Format hint for tokens"
)
in_location: str | None = Field(
default=None, alias="in", description="Where to send (header, query, cookie)"
)
class AgentSkill(RobocoBase):
"""A capability unit an agent can perform."""
id: str = Field(..., description="Unique skill identifier")
name: str = Field(..., description="Human-readable skill name")
description: str = Field(..., description="What this skill does")
tags: list[str] = Field(default_factory=list, description="Capability categories")
examples: list[str] = Field(
default_factory=list, description="Example invocations"
)
input_modes: list[str] = Field(
default_factory=lambda: ["text/plain"],
alias="inputModes",
description="Supported input MIME types",
)
output_modes: list[str] = Field(
default_factory=lambda: ["text/plain"],
alias="outputModes",
description="Supported output MIME types",
)
class AgentCard(RobocoBase):
"""
A2A Agent Card - The agent's public identity and capabilities.
Published at /.well-known/agent.json per A2A specification.
Acts as the agent's "business card" for discovery.
"""
# Required fields
id: str = Field(..., description="Unique agent identifier")
name: str = Field(..., description="Human-readable agent name")
provider: AgentProvider = Field(..., description="Provider information")
protocol_version: str = Field(
default="1.0",
alias="protocolVersion",
description="Supported A2A protocol version",
)
service_endpoint: str = Field(
..., alias="serviceEndpoint", description="Base URL for A2A operations"
)
capabilities: AgentCapabilities = Field(
default_factory=AgentCapabilities, description="Supported features"
)
security_schemes: dict[str, SecurityScheme] = Field(
default_factory=dict,
alias="securitySchemes",
description="Available auth methods",
)
security: list[dict[str, list[str]]] = Field(
default_factory=list, description="Required security scheme(s)"
)
# Optional fields
description: str | None = Field(
default=None, description="Agent purpose and capabilities"
)
skills: list[AgentSkill] = Field(
default_factory=list, description="Available operations"
)
default_input_modes: list[str] = Field(
default_factory=lambda: ["text/plain", "application/json"],
alias="defaultInputModes",
description="Default accepted input MIME types",
)
default_output_modes: list[str] = Field(
default_factory=lambda: ["text/plain", "application/json"],
alias="defaultOutputModes",
description="Default output MIME types",
)
documentation_url: str | None = Field(
default=None, alias="documentationUrl", description="Agent documentation URL"
)
version: str | None = Field(default=None, description="Agent version")
supports_extended_agent_card: bool = Field(
default=False,
alias="supportsExtendedAgentCard",
description="Whether authenticated card available",
)
metadata: dict[str, Any] = Field(
default_factory=dict, description="Custom metadata"
)
model_config = ConfigDict(
populate_by_name=True,
extra="allow", # A2A cards may have extensions
)
# =============================================================================
# MESSAGE MODELS
# =============================================================================
class TextPart(RobocoBase):
"""Plain text content part."""
type: Literal["text"] = "text"
text: str = Field(..., description="Text content")
class FilePart(RobocoBase):
"""File reference content part."""
type: Literal["file"] = "file"
file: dict[str, Any] = Field(
..., description="File data (uri, mimeType, name, data)"
)
class DataPart(RobocoBase):
"""Structured JSON data content part."""
type: Literal["data"] = "data"
data: dict[str, Any] = Field(..., description="Structured data")
class ArtifactPart(RobocoBase):
"""Reference to a generated artifact."""
type: Literal["artifact"] = "artifact"
artifact: dict[str, Any] = Field(..., description="Artifact reference")
# Union type for message parts
Part = Annotated[
TextPart | FilePart | DataPart | ArtifactPart, Field(discriminator="type")
]
class A2AMessage(RobocoBase):
"""
A2A Message - A communication turn between agents.
Contains one or more parts with content.
"""
role: Literal["user", "agent"] = Field(
..., description="Message sender role"
)
parts: list[Part] = Field(..., description="Content parts")
context_id: str | None = Field(
default=None, alias="contextId", description="Conversation grouping"
)
task_id: str | None = Field(
default=None, alias="taskId", description="Associated task"
)
message_id: str = Field(
default_factory=lambda: str(uuid4()),
alias="messageId",
description="Unique message ID",
)
model_config = ConfigDict(populate_by_name=True)
# =============================================================================
# TASK MODELS
# =============================================================================
class A2ATaskStatus(RobocoBase):
"""Status container for an A2A task."""
state: A2ATaskState = Field(..., description="Current lifecycle state")
message: A2AMessage | None = Field(
default=None, description="Associated status message"
)
timestamp: datetime = Field(
default_factory=lambda: datetime.now(UTC),
description="When status was recorded",
)
class A2AArtifact(RobocoBase):
"""An output artifact produced by a task."""
id: str = Field(default_factory=lambda: str(uuid4()), description="Artifact ID")
name: str = Field(..., description="Artifact name")
parts: list[Part] = Field(..., description="Artifact content parts")
metadata: dict[str, Any] = Field(
default_factory=dict, description="Custom metadata"
)
class A2ATask(RobocoBase):
"""
A2A Task - A unit of work with lifecycle management.
Maps to RoboCo's internal TaskTable but follows A2A semantics.
"""
id: str = Field(
default_factory=lambda: str(uuid4()), description="Server-generated task ID"
)
context_id: str = Field(
..., alias="contextId", description="Groups related interactions"
)
status: A2ATaskStatus = Field(..., description="Current task status")
artifacts: list[A2AArtifact] = Field(
default_factory=list, description="Output artifacts"
)
history: list[A2AMessage] = Field(
default_factory=list, description="Interaction history"
)
metadata: dict[str, Any] = Field(
default_factory=dict, description="Custom metadata"
)
model_config = ConfigDict(populate_by_name=True)
# =============================================================================
# JSON-RPC REQUEST/RESPONSE MODELS
# =============================================================================
class SendMessageConfiguration(RobocoBase):
"""Configuration for SendMessage request."""
accepted_output_modes: list[str] = Field(
default_factory=lambda: ["text/plain", "application/json"],
alias="acceptedOutputModes",
description="Client-accepted output MIME types",
)
history_length: int | None = Field(
default=None,
alias="historyLength",
description="Number of history turns to include in response",
)
blocking: bool = Field(
default=False, description="Wait for task completion before responding"
)
push_notification_config: dict[str, Any] | None = Field(
default=None,
alias="pushNotificationConfig",
description="Webhook config for async updates",
)
model_config = ConfigDict(populate_by_name=True)
class SendMessageRequest(RobocoBase):
"""Request payload for SendMessage JSON-RPC method."""
message: A2AMessage = Field(..., description="Message to send")
configuration: SendMessageConfiguration | None = Field(
default=None, description="Request configuration"
)
metadata: dict[str, Any] = Field(
default_factory=dict, description="Custom request metadata"
)
class SendMessageResponse(RobocoBase):
"""Response payload for SendMessage JSON-RPC method."""
task: A2ATask = Field(..., description="Created or updated task")
class GetTaskRequest(RobocoBase):
"""Request payload for GetTask JSON-RPC method."""
name: str = Field(..., description="Task resource name (tasks/{id})")
history_length: int | None = Field(
default=None,
alias="historyLength",
description="Number of history turns to include",
)
model_config = ConfigDict(populate_by_name=True)
class ListTasksRequest(RobocoBase):
"""Request payload for ListTasks JSON-RPC method."""
page_size: int = Field(
default=20, alias="pageSize", ge=1, le=100, description="Results per page"
)
page_token: str | None = Field(
default=None, alias="pageToken", description="Pagination token"
)
filter: str | None = Field(default=None, description="Filter expression")
order_by: str | None = Field(
default=None, alias="orderBy", description="Sort order"
)
model_config = ConfigDict(populate_by_name=True)
class ListTasksResponse(RobocoBase):
"""Response payload for ListTasks JSON-RPC method."""
tasks: list[A2ATask] = Field(..., description="Task list")
next_page_token: str | None = Field(
default=None, alias="nextPageToken", description="Token for next page"
)
model_config = ConfigDict(populate_by_name=True)
class CancelTaskRequest(RobocoBase):
"""Request payload for CancelTask JSON-RPC method."""
name: str = Field(..., description="Task resource name (tasks/{id})")
reason: str | None = Field(default=None, description="Cancellation reason")
# =============================================================================
# STATE MAPPING UTILITIES
# =============================================================================
def task_status_to_a2a_state(roboco_status: str) -> A2ATaskState:
"""
Map RoboCo TaskStatus to A2A TaskState.
This enables interoperability between RoboCo's internal
task lifecycle and the A2A protocol.
"""
mapping = {
"backlog": A2ATaskState.SUBMITTED,
"pending": A2ATaskState.SUBMITTED,
"claimed": A2ATaskState.WORKING,
"in_progress": A2ATaskState.WORKING,
"blocked": A2ATaskState.INPUT_REQUIRED,
"paused": A2ATaskState.INPUT_REQUIRED,
"verifying": A2ATaskState.WORKING,
"needs_revision": A2ATaskState.INPUT_REQUIRED,
"awaiting_qa": A2ATaskState.WORKING,
"awaiting_documentation": A2ATaskState.WORKING,
"awaiting_pm_review": A2ATaskState.WORKING,
"completed": A2ATaskState.COMPLETED,
"cancelled": A2ATaskState.CANCELLED,
}
return mapping.get(roboco_status, A2ATaskState.WORKING)
def a2a_state_to_task_status(a2a_state: A2ATaskState) -> str:
"""
Map A2A TaskState back to RoboCo TaskStatus.
Used when creating tasks via A2A protocol.
"""
mapping = {
A2ATaskState.SUBMITTED: "pending",
A2ATaskState.WORKING: "in_progress",
A2ATaskState.COMPLETED: "completed",
A2ATaskState.FAILED: "cancelled", # RoboCo uses cancelled for failures
A2ATaskState.CANCELLED: "cancelled",
A2ATaskState.INPUT_REQUIRED: "blocked",
A2ATaskState.REJECTED: "cancelled",
A2ATaskState.AUTH_REQUIRED: "blocked",
}
return mapping.get(a2a_state, "pending")
roboco/models/base.py
+1
View File
@@ -130,6 +130,7 @@ class NotificationType(str, Enum):
ALERT = "alert" ALERT = "alert"
BROADCAST = "broadcast" BROADCAST = "broadcast"
KNOWLEDGE_SHARE = "knowledge_share" # Cross-agent learning notification KNOWLEDGE_SHARE = "knowledge_share" # Cross-agent learning notification
MENTION = "mention" # @mention in chat
class NotificationPriority(str, Enum): class NotificationPriority(str, Enum):
roboco/models/events.py
+1
View File
@@ -20,6 +20,7 @@ class EventType(str, Enum):
# Task lifecycle events # Task lifecycle events
TASK_CREATED = "task.created" TASK_CREATED = "task.created"
TASK_ASSIGNED = "task.assigned" # A2A: task assigned to agent, triggers spawn/notify
TASK_CLAIMED = "task.claimed" TASK_CLAIMED = "task.claimed"
TASK_STARTED = "task.started" TASK_STARTED = "task.started"
TASK_BLOCKED = "task.blocked" TASK_BLOCKED = "task.blocked"
roboco/seeds/initial_data.py
+10 -3
View File
@@ -205,9 +205,16 @@ DEFAULT_AGENTS: list[dict[str, Any]] = [
}, },
# UX/UI Cell # UX/UI Cell
{ {
"id": AGENT_UUIDS["ux-dev"], "id": AGENT_UUIDS["ux-dev-1"],
"slug": "ux-dev", "slug": "ux-dev-1",
"name": "UX/UI Developer", "name": "UX/UI Developer 1",
"role": "developer",
"team": "ux_ui",
},
{
"id": AGENT_UUIDS["ux-dev-2"],
"slug": "ux-dev-2",
"name": "UX/UI Developer 2",
"role": "developer", "role": "developer",
"team": "ux_ui", "team": "ux_ui",
}, },
roboco/services/a2a.py
+480
View File
@@ -0,0 +1,480 @@
"""
A2A (Agent-to-Agent) Protocol Service
Provides business logic for A2A protocol operations including:
- Agent discovery and card generation
- Task lifecycle management via A2A semantics
- Message handling and routing
"""
from uuid import UUID
import structlog
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from roboco.config import settings
from roboco.db.tables import AgentTable, TaskTable
from roboco.models.a2a import (
A2AArtifact,
A2AMessage,
A2ATask,
A2ATaskStatus,
AgentCapabilities,
AgentCard,
AgentProvider,
AgentSkill,
SecurityScheme,
TextPart,
task_status_to_a2a_state,
)
from roboco.models.base import TaskStatus, Team
logger = structlog.get_logger()
class A2AService:
"""
Service layer for A2A protocol operations.
Provides methods for:
- Building Agent Cards for discovery
- Converting between RoboCo tasks and A2A tasks
- Processing A2A messages
"""
def __init__(self, session: AsyncSession):
"""Initialize with database session."""
self.session = session
@staticmethod
def get_service_endpoint() -> str:
"""Build service endpoint URL from settings."""
connect_host = "127.0.0.1" if settings.host == "0.0.0.0" else settings.host
return f"http://{connect_host}:{settings.port}"
@staticmethod
def build_system_agent_card() -> AgentCard:
"""
Build the system-level Agent Card for RoboCo.
This card represents the entire RoboCo system and is served
at /.well-known/agent.json
"""
return AgentCard(
id="roboco-system",
name="RoboCo System",
description=(
"RoboCo is an AI Agentic Company - a virtual organization of "
"AI agents designed to operate as a complete software "
"development workforce."
),
provider=AgentProvider(
organization="RoboCo",
url="https://github.com/roboco",
),
protocol_version="1.0",
service_endpoint=f"{A2AService.get_service_endpoint()}/api/v1/a2a",
version=settings.app_version,
capabilities=AgentCapabilities(
streaming=True,
push_notifications=False,
state_transition_history=True,
),
default_input_modes=["text/plain", "application/json"],
default_output_modes=["text/plain", "application/json"],
skills=[
AgentSkill(
id="software-development",
name="Software Development",
description="Full-stack software development with AI agents",
tags=["development", "coding", "qa", "documentation"],
),
AgentSkill(
id="task-management",
name="Task Management",
description="Create and manage development tasks",
tags=["tasks", "kanban", "planning"],
),
AgentSkill(
id="code-review",
name="Code Review",
description="Review and quality assurance of code",
tags=["qa", "review", "testing"],
),
],
documentation_url="https://github.com/roboco/docs",
security_schemes={
"bearerAuth": SecurityScheme(type="http", scheme="bearer"),
},
security=[{"bearerAuth": []}],
)
async def build_agent_card(self, agent_id: str) -> AgentCard | None:
"""
Build an Agent Card for a specific agent.
Args:
agent_id: Either a UUID string or agent slug
Returns:
AgentCard for the agent, or None if not found
"""
# Try to parse as UUID first
try:
uuid = UUID(agent_id)
result = await self.session.execute(
select(AgentTable).where(AgentTable.id == uuid)
)
except ValueError:
# Not a UUID, try slug lookup
result = await self.session.execute(
select(AgentTable).where(AgentTable.slug == agent_id)
)
agent = result.scalar_one_or_none()
if agent is None:
return None
return self._agent_to_card(agent)
def _agent_to_card(self, agent: AgentTable) -> AgentCard:
"""Convert an AgentTable row to an AgentCard."""
agent_id = str(agent.id)
agent_slug = agent.slug
# Map role to skills
role_skills: dict[str, list[AgentSkill]] = {
"developer": [
AgentSkill(
id="coding",
name="Code Development",
description="Write and implement code",
tags=["development", "coding"],
),
AgentSkill(
id="debugging",
name="Debugging",
description="Debug and fix code issues",
tags=["debugging", "troubleshooting"],
),
],
"qa": [
AgentSkill(
id="testing",
name="Testing",
description="Test code and verify quality",
tags=["qa", "testing"],
),
AgentSkill(
id="review",
name="Code Review",
description="Review code for quality and issues",
tags=["qa", "review"],
),
],
"documenter": [
AgentSkill(
id="documentation",
name="Documentation",
description="Write technical documentation",
tags=["documentation", "writing"],
),
],
"cell_pm": [
AgentSkill(
id="coordination",
name="Task Coordination",
description="Coordinate tasks within the cell",
tags=["management", "coordination"],
),
],
"main_pm": [
AgentSkill(
id="planning",
name="Project Planning",
description="Plan and coordinate across cells",
tags=["management", "planning"],
),
],
}
skills = role_skills.get(agent.role, [])
return AgentCard(
id=agent_id,
name=agent.name,
description=f"{agent.name} - {agent.role} agent in RoboCo",
provider=AgentProvider(
organization="RoboCo",
url="https://github.com/roboco",
),
protocol_version="1.0",
service_endpoint=f"{self.get_service_endpoint()}/api/v1/a2a",
version=settings.app_version,
capabilities=AgentCapabilities(
streaming=True,
push_notifications=False,
state_transition_history=True,
),
default_input_modes=["text/plain", "application/json"],
default_output_modes=["text/plain", "application/json"],
skills=skills,
metadata={
"slug": agent_slug,
"role": agent.role,
"team": agent.team,
},
security_schemes={
"bearerAuth": SecurityScheme(type="http", scheme="bearer"),
},
security=[{"bearerAuth": []}],
)
def task_to_a2a(self, task: TaskTable) -> A2ATask:
"""
Convert a RoboCo TaskTable to A2A Task.
This is the canonical conversion that maintains semantic
mapping between RoboCo's internal task model and A2A.
"""
task_id = str(task.id)
# Get status value as string
if hasattr(task.status, "value"):
status_value = task.status.value
else:
status_value = str(task.status)
a2a_state = task_status_to_a2a_state(status_value)
# Build status message from dev_notes if present
status_message = None
if task.dev_notes:
status_message = A2AMessage(
role="agent",
parts=[TextPart(text=task.dev_notes)],
task_id=task_id,
)
a2a_status = A2ATaskStatus(
state=a2a_state,
message=status_message,
timestamp=task.updated_at or task.created_at,
)
# Build artifacts from task outputs (future: populate from outputs)
artifacts: list[A2AArtifact] = []
# Build metadata
metadata: dict[str, str | int] = {
"roboco_status": status_value,
"priority": task.priority,
"team": str(task.team),
}
if task.assigned_to:
metadata["assigned_to"] = str(task.assigned_to)
if task.parent_task_id:
metadata["parent_task_id"] = str(task.parent_task_id)
return A2ATask(
id=task_id,
context_id=task_id,
status=a2a_status,
artifacts=artifacts,
history=[],
metadata=metadata,
)
async def get_task(self, task_id: str) -> A2ATask | None:
"""
Get a task by ID and return as A2A Task.
Args:
task_id: Task UUID string
Returns:
A2ATask or None if not found
"""
try:
task_uuid = UUID(task_id)
except ValueError:
return None
result = await self.session.execute(
select(TaskTable).where(TaskTable.id == task_uuid)
)
task = result.scalar_one_or_none()
if task is None:
return None
return self.task_to_a2a(task)
async def list_tasks(
self,
page_size: int = 20,
offset: int = 0,
order_by: str | None = None,
) -> tuple[list[A2ATask], bool]:
"""
List tasks with pagination.
Args:
page_size: Number of results to return
offset: Starting offset
order_by: Sort order ("created_at desc" or "created_at asc")
Returns:
Tuple of (tasks, has_more)
"""
query = select(TaskTable)
# Apply ordering
if order_by == "created_at asc":
query = query.order_by(TaskTable.created_at.asc())
else:
query = query.order_by(TaskTable.created_at.desc())
# Apply pagination (fetch one extra to detect more)
query = query.offset(offset).limit(page_size + 1)
result = await self.session.execute(query)
tasks = list(result.scalars().all())
has_more = len(tasks) > page_size
if has_more:
tasks = tasks[:page_size]
return [self.task_to_a2a(t) for t in tasks], has_more
async def create_task_from_message(
self,
title: str,
description: str,
created_by: UUID,
team: Team = Team.BACKEND,
) -> A2ATask:
"""
Create a new task from an A2A message.
Args:
title: Task title
description: Task description
created_by: Agent ID creating the task
team: Team assignment
Returns:
Created A2ATask
"""
task = TaskTable(
title=title,
description=description,
acceptance_criteria=["Task completed as specified"],
status=TaskStatus.PENDING,
priority=5,
team=team,
created_by=created_by,
)
self.session.add(task)
await self.session.flush()
await self.session.refresh(task)
logger.info(
"Created task from A2A message",
task_id=str(task.id),
title=title,
)
return self.task_to_a2a(task)
async def cancel_task(self, task_id: str, reason: str | None = None) -> A2ATask:
"""
Cancel a task.
Args:
task_id: Task UUID string
reason: Optional cancellation reason
Returns:
Updated A2ATask
Raises:
ValueError: If task not found or already in terminal state
"""
try:
task_uuid = UUID(task_id)
except ValueError as e:
raise ValueError(f"Invalid task ID: {task_id}") from e
result = await self.session.execute(
select(TaskTable).where(TaskTable.id == task_uuid)
)
task = result.scalar_one_or_none()
if task is None:
raise ValueError(f"Task not found: {task_id}")
# Check if cancellable
if hasattr(task.status, "value"):
status_value = task.status.value
else:
status_value = str(task.status)
if status_value in ["completed", "cancelled"]:
raise ValueError(f"Task already in terminal state: {status_value}")
# Cancel the task
task.status = TaskStatus.CANCELLED
if reason:
reason_text = f"Cancellation reason: {reason}"
if task.dev_notes:
task.dev_notes = f"{task.dev_notes}\n\n{reason_text}"
else:
task.dev_notes = reason_text
await self.session.flush()
await self.session.refresh(task)
logger.info("Cancelled task via A2A", task_id=task_id, reason=reason)
return self.task_to_a2a(task)
async def discover_agents(
self,
role: str | None = None,
team: str | None = None,
skill_tag: str | None = None,
) -> list[AgentCard]:
"""
Discover agents matching criteria.
Args:
role: Filter by agent role
team: Filter by team
skill_tag: Filter by skill tag (future)
Returns:
List of matching AgentCards
"""
query = select(AgentTable)
if role:
query = query.where(AgentTable.role == role)
if team:
query = query.where(AgentTable.team == team)
result = await self.session.execute(query)
agents = result.scalars().all()
cards = [self._agent_to_card(agent) for agent in agents]
# Filter by skill tag if specified
if skill_tag:
cards = [
card
for card in cards
if any(skill_tag in skill.tags for skill in card.skills)
]
return cards
roboco/services/messaging.py
+53
View File
@@ -22,6 +22,7 @@ from roboco.db.tables import (
ChannelTable, ChannelTable,
GroupTable, GroupTable,
MessageTable, MessageTable,
NotificationTable,
SessionTable, SessionTable,
SessionTaskTable, SessionTaskTable,
) )
@@ -29,6 +30,8 @@ from roboco.enforcement import validate_channel_access
from roboco.events import Event, EventType, get_event_bus from roboco.events import Event, EventType, get_event_bus
from roboco.models.base import ( from roboco.models.base import (
MessageType, MessageType,
NotificationPriority,
NotificationType,
SessionStatus, SessionStatus,
) )
from roboco.models.messaging import ( from roboco.models.messaging import (
@@ -765,6 +768,53 @@ class MessagingService(BaseService):
channel.message_count += 1 channel.message_count += 1
channel.last_activity = now channel.last_activity = now
async def _notify_mentions(
self,
message: MessageTable,
sender_id: UUID,
channel_slug: str,
) -> None:
"""
Create and deliver notifications for mentioned agents.
Publishes NOTIFICATION_SENT events to Redis Streams for real-time
delivery via the WebSocket bridge.
"""
if not message.mentions:
return
# Lazy import to avoid circular dependency
from roboco.services.notification_delivery import get_notification_delivery_service
delivery_service = get_notification_delivery_service(self.session)
for mentioned_id in message.mentions:
# Don't notify yourself
if mentioned_id == sender_id:
continue
# Create mention notification
notification = NotificationTable(
type=NotificationType.MENTION,
priority=NotificationPriority.NORMAL,
from_agent=sender_id,
to_agents=[mentioned_id],
subject=f"You were mentioned in #{channel_slug}",
body=message.content[:500], # Truncate for notification
related_task_id=message.task_id,
)
self.session.add(notification)
await self.session.flush()
# Deliver via Redis Streams -> WebSocket
await delivery_service.deliver(notification.id)
self.log.debug(
"Mention notification sent",
mentioned_id=str(mentioned_id),
message_id=str(message.id),
)
async def send_message( async def send_message(
self, self,
req: MessageCreateRequest, req: MessageCreateRequest,
@@ -810,6 +860,9 @@ class MessagingService(BaseService):
self._update_message_stats(session, group, channel, content_length) self._update_message_stats(session, group, channel, content_length)
await self.session.flush() await self.session.flush()
# Notify mentioned agents via Redis Streams
await self._notify_mentions(message, req.agent_id, channel.slug)
if self._check_session_boundaries(session): if self._check_session_boundaries(session):
await self.close_session(cast("UUID", session.id), "Boundary exceeded") await self.close_session(cast("UUID", session.id), "Boundary exceeded")
roboco/services/notification.py
+12 -2
View File
@@ -171,7 +171,7 @@ class NotificationService:
) )
async def _create_notification(self, params: CreateNotificationParams) -> None: async def _create_notification(self, params: CreateNotificationParams) -> None:
"""Create a notification via the database.""" """Create a notification via the database and deliver it."""
async with get_db_context() as db: async with get_db_context() as db:
notification = NotificationTable( notification = NotificationTable(
type=params.notification_type, type=params.notification_type,
@@ -183,10 +183,20 @@ class NotificationService:
related_task_id=params.related_task_id, related_task_id=params.related_task_id,
) )
db.add(notification) db.add(notification)
await db.flush()
# Deliver via Redis Streams for real-time push
from roboco.services.notification_delivery import (
get_notification_delivery_service,
)
delivery_service = get_notification_delivery_service(db)
await delivery_service.deliver(notification.id)
await db.commit() await db.commit()
logger.info( logger.info(
"Notification created", "Notification created and delivered",
notification_id=str(notification.id), notification_id=str(notification.id),
type=params.notification_type.value, type=params.notification_type.value,
) )
roboco/services/optimal.py
+203 -6
View File
@@ -239,6 +239,53 @@ class OptimalService:
return await plugin.index_sources(sources, project) return await plugin.index_sources(sources, project)
return await plugin.add_sources(sources) return await plugin.add_sources(sources)
async def _track_indexed_document(
self,
index_type: IndexType,
source: str,
title: str | None = None,
preview: str | None = None,
metadata: dict | None = None,
) -> None:
"""Track an indexed document in the database for browsing/stats."""
import hashlib
from roboco.db import get_db_context
from roboco.db.tables import IndexedDocumentTable
source_hash = hashlib.sha256(source.encode()).hexdigest()
async with get_db_context() as db:
from sqlalchemy import select
existing = await db.execute(
select(IndexedDocumentTable).where(
IndexedDocumentTable.index_type == index_type.value,
IndexedDocumentTable.source_hash == source_hash,
)
)
doc = existing.scalar_one_or_none()
if doc:
if title:
doc.title = title
if preview:
doc.preview = preview[:500] if preview else None
if metadata:
doc.metadata = {**(doc.metadata or {}), **metadata}
else:
doc = IndexedDocumentTable(
index_type=index_type.value,
source=source,
source_hash=source_hash,
title=title,
preview=preview[:500] if preview else None,
metadata=metadata or {},
)
db.add(doc)
await db.commit()
async def index_conversation(self, params: IndexConversationParams) -> None: async def index_conversation(self, params: IndexConversationParams) -> None:
"""Index a conversation message.""" """Index a conversation message."""
plugin = self._get_plugin(IndexType.CONVERSATIONS) plugin = self._get_plugin(IndexType.CONVERSATIONS)
@@ -254,6 +301,20 @@ class OptimalService:
message_type=params.message_type, message_type=params.message_type,
) )
# Track in database
source = f"roboco://conversations/{params.session_id or 'unknown'}"
await self._track_indexed_document(
IndexType.CONVERSATIONS,
source=source,
title=f"Message in {params.channel_id or 'channel'}",
preview=params.content[:500] if params.content else None,
metadata={
"channel_id": params.channel_id,
"session_id": params.session_id,
"agent_id": str(params.agent_id) if params.agent_id else None,
},
)
async def index_journal_entry(self, params: IndexJournalEntryParams) -> None: async def index_journal_entry(self, params: IndexJournalEntryParams) -> None:
"""Index a journal entry.""" """Index a journal entry."""
plugin = self._get_plugin(IndexType.JOURNALS) plugin = self._get_plugin(IndexType.JOURNALS)
@@ -269,6 +330,21 @@ class OptimalService:
tags=params.tags, tags=params.tags,
) )
# Track in database
source = f"roboco://journals/{params.entry_id or 'unknown'}"
await self._track_indexed_document(
IndexType.JOURNALS,
source=source,
title=f"Journal: {params.entry_type or 'entry'}",
preview=params.content[:500] if params.content else None,
metadata={
"entry_id": params.entry_id,
"agent_id": str(params.agent_id) if params.agent_id else None,
"entry_type": params.entry_type,
"tags": params.tags,
},
)
# ========================================================================= # =========================================================================
# INDEXING OPERATIONS (New - Optimal Brain) # INDEXING OPERATIONS (New - Optimal Brain)
# ========================================================================= # =========================================================================
@@ -279,21 +355,71 @@ class OptimalService:
if isinstance(plugin, ErrorsIndexPlugin): if isinstance(plugin, ErrorsIndexPlugin):
await plugin.record_error(params) await plugin.record_error(params)
# Track in database
import hashlib
error_hash = hashlib.md5(params.error_message.encode()).hexdigest()[:12]
source = f"roboco://errors/err-{error_hash}"
await self._track_indexed_document(
IndexType.ERRORS,
source=source,
title=f"Error: {params.error_message[:100]}",
preview=f"{params.error_message}\n\nSolution: {params.solution}",
metadata={
"context": params.context,
"worked": params.worked,
"tags": params.tags,
},
)
async def index_standard(self, params: IndexStandardParams) -> None: async def index_standard(self, params: IndexStandardParams) -> None:
"""Index a coding/security/workflow standard.""" """Index a coding/security/workflow standard."""
plugin = self._get_plugin(IndexType.STANDARDS) plugin = self._get_plugin(IndexType.STANDARDS)
if isinstance(plugin, StandardsIndexPlugin): if isinstance(plugin, StandardsIndexPlugin):
await plugin.index_standard(params) await plugin.index_standard(params)
# Track in database
source = f"roboco://standards/{params.domain or 'general'}"
await self._track_indexed_document(
IndexType.STANDARDS,
source=source,
title=f"Standard: {params.domain or 'General'}",
preview=params.content[:500] if params.content else None,
metadata={
"domain": params.domain,
"language": params.language,
"scope": params.scope,
"severity": params.severity,
},
)
async def index_decision(self, params: IndexDecisionParams) -> None: async def index_decision(self, params: IndexDecisionParams) -> None:
"""Index an architectural/design decision.""" """Index an architectural/design decision."""
plugin = self._get_plugin(IndexType.DECISIONS) plugin = self._get_plugin(IndexType.DECISIONS)
if isinstance(plugin, DecisionsIndexPlugin): if isinstance(plugin, DecisionsIndexPlugin):
await plugin.record_decision(params) await plugin.record_decision(params)
# Track in database
import hashlib
topic_hash = hashlib.md5(params.topic.encode()).hexdigest()[:12]
source = f"roboco://decisions/dec-{topic_hash}"
await self._track_indexed_document(
IndexType.DECISIONS,
source=source,
title=f"Decision: {params.topic[:100]}",
preview=f"{params.topic}\n\nDecision: {params.decision}\n\nRationale: {params.rationale}",
metadata={
"scope": params.scope,
"tags": params.tags,
"alternatives": params.alternatives,
},
)
async def record_review(self, params: IndexReviewParams) -> str: async def record_review(self, params: IndexReviewParams) -> str:
"""Record a code review for future reference.""" """Record a code review for future reference."""
plugin = self._get_plugin(IndexType.REVIEWS) plugin = self._get_plugin(IndexType.REVIEWS)
doc_id = ""
if isinstance(plugin, ReviewsIndexPlugin): if isinstance(plugin, ReviewsIndexPlugin):
review_params = ReviewParams( review_params = ReviewParams(
comment=params.summary, comment=params.summary,
@@ -304,24 +430,74 @@ class OptimalService:
severity="info", severity="info",
) )
result = await plugin.record_review(review_params) result = await plugin.record_review(review_params)
return result.doc_id doc_id = result.doc_id
return ""
# Track in database
source = f"roboco://reviews/{params.file_path or 'unknown'}"
await self._track_indexed_document(
IndexType.REVIEWS,
source=source,
title=f"Review: {params.file_path or 'Code'}",
preview=params.summary[:500] if params.summary else None,
metadata={
"file_path": params.file_path,
"reviewer_id": str(params.reviewer_id) if params.reviewer_id else None,
"task_id": str(params.task_id) if params.task_id else None,
},
)
return doc_id
async def record_learning(self, params: LearningParams) -> str: async def record_learning(self, params: LearningParams) -> str:
"""Record a learning for cross-agent knowledge sharing.""" """Record a learning for cross-agent knowledge sharing."""
plugin = self._get_plugin(IndexType.LEARNINGS) plugin = self._get_plugin(IndexType.LEARNINGS)
doc_id = ""
if isinstance(plugin, LearningsIndexPlugin): if isinstance(plugin, LearningsIndexPlugin):
result = await plugin.record_learning(params) result = await plugin.record_learning(params)
return result.doc_id doc_id = result.doc_id
return ""
# Track in database
import hashlib
content_hash = hashlib.md5(params.content.encode()).hexdigest()[:12]
source = f"roboco://learnings/learn-{content_hash}"
await self._track_indexed_document(
IndexType.LEARNINGS,
source=source,
title=f"Learning: {params.category or 'General'}",
preview=params.content[:500] if params.content else None,
metadata={
"category": params.category,
"team": params.team,
"shareable": params.shareable,
"tags": params.tags,
},
)
return doc_id
async def index_standards_file(self, file_path: str) -> int: async def index_standards_file(self, file_path: str) -> int:
"""Index a markdown standards file.""" """Index a markdown standards file."""
from pathlib import Path
plugin = self._get_plugin(IndexType.STANDARDS) plugin = self._get_plugin(IndexType.STANDARDS)
count = 0
if isinstance(plugin, StandardsIndexPlugin): if isinstance(plugin, StandardsIndexPlugin):
results = await plugin.index_markdown_file(file_path) results = await plugin.index_markdown_file(file_path)
return len([r for r in results if r.success]) count = len([r for r in results if r.success])
return 0
# Track in database
path = Path(file_path)
if path.exists():
await self._track_indexed_document(
IndexType.STANDARDS,
source=str(path.absolute()),
title=path.stem.replace("-", " ").replace("_", " ").title(),
preview=path.read_text(errors="ignore")[:500],
metadata={"file_path": file_path},
)
return count
# ========================================================================= # =========================================================================
# SEARCH OPERATIONS # SEARCH OPERATIONS
@@ -504,6 +680,27 @@ class OptimalService:
await plugin.clear() await plugin.clear()
logger.info("Cleared index", index_type=index_type.value) logger.info("Cleared index", index_type=index_type.value)
async def list_documents(
self, index_type: IndexType, limit: int = 50, offset: int = 0
) -> list[dict[str, Any]]:
"""
List documents in a specific index.
Returns list of documents with id, source, indexed_at, and metadata.
"""
plugin = self._get_plugin(index_type)
# Check if plugin has list_documents method
if hasattr(plugin, "list_documents"):
return await plugin.list_documents(limit=limit, offset=offset)
# Fallback: return empty list if plugin doesn't support listing
logger.warning(
"Plugin does not support list_documents",
index_type=index_type.value,
)
return []
async def refresh_index(self, index_type: IndexType, sources: list[str]) -> None: async def refresh_index(self, index_type: IndexType, sources: list[str]) -> None:
"""Refresh an index with new sources.""" """Refresh an index with new sources."""
plugin = self._get_plugin(index_type) plugin = self._get_plugin(index_type)
roboco/services/optimal_brain/indexes/base.py
+96
View File
@@ -495,6 +495,35 @@ class BaseIndexPlugin(ABC):
await self.ragi.clear() await self.ragi.clear()
logger.info(f"Cleared {self.index_type.value} index") logger.info(f"Cleared {self.index_type.value} index")
async def list_documents(
self, limit: int = 50, offset: int = 0
) -> list[dict[str, Any]]:
"""
List documents in the index.
Returns list of documents with id, source, indexed_at, and metadata.
"""
try:
# Use piragi's list method if available
if hasattr(self.ragi, "list"):
docs = await self.ragi.list(limit=limit, offset=offset)
return [
{
"id": str(doc.get("id", "")),
"source": doc.get("source", ""),
"indexed_at": doc.get("indexed_at", ""),
"metadata": doc.get("metadata", {}),
}
for doc in docs
]
# Fallback: return empty list
return []
except Exception as e:
logger.warning(
f"Failed to list documents in {self.index_type.value}: {e}"
)
return []
async def add_sources(self, sources: list[str]) -> int: async def add_sources(self, sources: list[str]) -> int:
""" """
Add file/directory sources to the index. Add file/directory sources to the index.
@@ -507,5 +536,72 @@ class BaseIndexPlugin(ABC):
Returns: Returns:
Number of documents indexed Number of documents indexed
""" """
import hashlib
from pathlib import Path
from roboco.db import get_db_context
from roboco.db.tables import IndexedDocumentTable
await self.ragi.add(sources) await self.ragi.add(sources)
# Track indexed documents in database
async with get_db_context() as db:
for source in sources:
source_path = Path(source)
# Handle glob patterns and directories
if "*" in source or source_path.is_dir():
if source_path.is_dir():
files = list(source_path.rglob("*"))
else:
files = list(Path(".").glob(source))
else:
files = [source_path] if source_path.exists() else []
for file_path in files:
if not file_path.is_file():
continue
# Generate hash for dedup
source_str = str(file_path.absolute())
source_hash = hashlib.sha256(source_str.encode()).hexdigest()
# Extract title from filename or first line
title = file_path.stem.replace("-", " ").replace("_", " ").title()
# Get preview (first 500 chars)
preview = None
try:
content = file_path.read_text(errors="ignore")[:500]
preview = content.strip()
except Exception:
pass
# Upsert document record
from sqlalchemy import select
existing = await db.execute(
select(IndexedDocumentTable).where(
IndexedDocumentTable.index_type == self.index_type.value,
IndexedDocumentTable.source_hash == source_hash,
)
)
doc = existing.scalar_one_or_none()
if doc:
doc.title = title
doc.preview = preview
else:
doc = IndexedDocumentTable(
index_type=self.index_type.value,
source=source_str,
source_hash=source_hash,
title=title,
preview=preview,
chunk_count=0, # Could be calculated later
)
db.add(doc)
await db.commit()
return await self.count() return await self.count()
uv.lock
Generated
+8 -6
View File
@@ -850,7 +850,7 @@ wheels = [
[[package]] [[package]]
name = "fastapi" name = "fastapi"
version = "0.127.1" version = "0.128.0"
source = { registry = "https://pypi.org/simple" } source = { registry = "https://pypi.org/simple" }
dependencies = [ dependencies = [
{ name = "annotated-doc" }, { name = "annotated-doc" },
@@ -858,9 +858,9 @@ dependencies = [
{ name = "starlette" }, { name = "starlette" },
{ name = "typing-extensions" }, { name = "typing-extensions" },
] ]
sdist = { url = "https://files.pythonhosted.org/packages/96/8a/6b9ba6eb8ff3817caae83120495965d9e70afb4d6348cb120e464ee199f4/fastapi-0.127.1.tar.gz", hash = "sha256:946a87ee5d931883b562b6bada787d6c8178becee2683cb3f9b980d593206359", size = 391876, upload-time = "2025-12-26T13:04:47.075Z" } sdist = { url = "https://files.pythonhosted.org/packages/52/08/8c8508db6c7b9aae8f7175046af41baad690771c9bcde676419965e338c7/fastapi-0.128.0.tar.gz", hash = "sha256:1cc179e1cef10a6be60ffe429f79b829dce99d8de32d7acb7e6c8dfdf7f2645a", size = 365682, upload-time = "2025-12-27T15:21:13.714Z" }
wheels = [ wheels = [
{ url = "https://files.pythonhosted.org/packages/d2/f3/a6858d147ed2645c095d11dc2440f94a5f1cd8f4df888e3377e6b5281a0f/fastapi-0.127.1-py3-none-any.whl", hash = "sha256:31d670a4f9373cc6d7994420f98e4dc46ea693145207abc39696746c83a44430", size = 112332, upload-time = "2025-12-26T13:04:45.329Z" }, { url = "https://files.pythonhosted.org/packages/5c/05/5cbb59154b093548acd0f4c7c474a118eda06da25aa75c616b72d8fcd92a/fastapi-0.128.0-py3-none-any.whl", hash = "sha256:aebd93f9716ee3b4f4fcfe13ffb7cf308d99c9f3ab5622d8877441072561582d", size = 103094, upload-time = "2025-12-27T15:21:12.154Z" },
] ]
[[package]] [[package]]
@@ -2447,15 +2447,15 @@ wheels = [
[[package]] [[package]]
name = "pdfminer-six" name = "pdfminer-six"
version = "20251107" version = "20251227"
source = { registry = "https://pypi.org/simple" } source = { registry = "https://pypi.org/simple" }
dependencies = [ dependencies = [
{ name = "charset-normalizer" }, { name = "charset-normalizer" },
{ name = "cryptography" }, { name = "cryptography" },
] ]
sdist = { url = "https://files.pythonhosted.org/packages/1d/50/5315f381a25dc80a8d2ea7c62d9a28c0137f10ccc263623a0db8b49fcced/pdfminer_six-20251107.tar.gz", hash = "sha256:5fb0c553799c591777f22c0c72b77fc2522d7d10c70654e25f4c5f1fd996e008", size = 7387104, upload-time = "2025-11-07T20:01:10.286Z" } sdist = { url = "https://files.pythonhosted.org/packages/fb/ef/543d0d44c9b03ac08316d31afd8e52b8ab8a86d03620ee0d61d6006bba9c/pdfminer_six-20251227.tar.gz", hash = "sha256:1d98781cf884d7274c694e742a94ffe6326ee11ee7ea79e73e1e75c4c74d91eb", size = 7388054, upload-time = "2025-12-27T20:16:22.969Z" }
wheels = [ wheels = [
{ url = "https://files.pythonhosted.org/packages/64/29/d1d9f6b900191288b77613ddefb73ed35b48fb35e44aaf8b01b0422b759d/pdfminer_six-20251107-py3-none-any.whl", hash = "sha256:c09df33e4cbe6b26b2a79248a4ffcccafaa5c5d39c9fff0e6e81567f165b5401", size = 5620299, upload-time = "2025-11-07T20:01:08.722Z" }, { url = "https://files.pythonhosted.org/packages/ea/d3/90bc3ec25e5f33e8e1dea5fd67eae0ef2e8666e7721156fb0091c7098a42/pdfminer_six-20251227-py3-none-any.whl", hash = "sha256:7a41f668a74fdde9e3a6c9265e400992a8312224fb6dbc13ee8861111e330d36", size = 5620874, upload-time = "2025-12-27T20:16:21.09Z" },
] ]
[[package]] [[package]]
@@ -3300,6 +3300,7 @@ dependencies = [
{ name = "python-toon" }, { name = "python-toon" },
{ name = "redis" }, { name = "redis" },
{ name = "sqlalchemy", extra = ["asyncio"] }, { name = "sqlalchemy", extra = ["asyncio"] },
{ name = "sse-starlette" },
{ name = "structlog" }, { name = "structlog" },
{ name = "tenacity" }, { name = "tenacity" },
{ name = "tiktoken" }, { name = "tiktoken" },
@@ -3383,6 +3384,7 @@ requires-dist = [
{ name = "safety", marker = "extra == 'dev'" }, { name = "safety", marker = "extra == 'dev'" },
{ name = "semgrep", marker = "extra == 'dev'" }, { name = "semgrep", marker = "extra == 'dev'" },
{ name = "sqlalchemy", extras = ["asyncio"] }, { name = "sqlalchemy", extras = ["asyncio"] },
{ name = "sse-starlette" },
{ name = "structlog" }, { name = "structlog" },
{ name = "tenacity" }, { name = "tenacity" },
{ name = "tiktoken" }, { name = "tiktoken" },