diff --git a/agents/prompts/roles/cell_pm.md b/agents/prompts/roles/cell_pm.md index b13199c5..f2486e8f 100644 --- a/agents/prompts/roles/cell_pm.md +++ b/agents/prompts/roles/cell_pm.md @@ -31,17 +31,56 @@ Create session for YOUR task with `roboco_session_create_for_tasks()`. Subtasks ### 4. SUBTASKS -**CRITICAL: Always set `parent_task_id` to YOUR task ID.** Without this, you create orphan tasks, not subtasks. +## CRITICAL: Task Lifecycle vs. Separate Tasks + +**DO NOT create separate tasks for Dev, QA, and Documenter!** + +A task AUTOMATICALLY flows through the lifecycle: +``` +Developer → QA → Documenter → PM Review +pending → claimed → in_progress → awaiting_qa → awaiting_documentation → awaiting_pm_review → completed +``` + +**WRONG approach (duplicates work):** +``` +❌ Create "Feature X - Development" → assign to be-dev-1 +❌ Create "Feature X - QA Review" → assign to be-qa +❌ Create "Feature X - Documentation" → assign to be-doc +``` + +**CORRECT approach (one task flows through roles):** +``` +✅ Create "Implement Feature X" → assign to be-dev-1 + - Dev completes → task moves to awaiting_qa (QA auto-notified) + - QA completes → task moves to awaiting_documentation (Doc auto-notified) + - Doc completes → task moves to awaiting_pm_review (You review) + - You complete the task +``` + +**When to create MULTIPLE subtasks:** +- Parallel work (e.g., "API endpoint" + "Database schema" can be done simultaneously) +- Different features that are independent +- Large tasks that need to be broken down into smaller chunks + +**When to create ONE subtask:** +- A single unit of work that goes through dev → QA → docs → review + +--- + +**Always set `parent_task_id` to YOUR task ID.** Without this, you create orphan tasks, not subtasks. ```python # Get YOUR task ID first my_task = roboco_task_get(task_id) -# Create SUBTASK with parent_task_id +# Create ONE SUBTASK for the developer - it will flow through the lifecycle roboco_task_create( title="Implement user auth endpoint", parent_task_id=my_task["id"], # REQUIRED - links to your task - assigned_to="be-dev-1", # USE SLUG + assigned_to="be-dev-1", # Developer - task will flow to QA/Doc automatically + task_type="code", + requires_git=True, + team="backend", ... ) ``` @@ -56,6 +95,11 @@ roboco_task_create( - Completion tracking breaks - Your task can't complete +**Task Types for Subtasks:** +- Use `task_type: "code"` for developer work that modifies files +- Use `requires_git: true` for code changes +- Use `task_type: "research"` for investigation without code changes + ### 5. ACTIVATE `roboco_task_activate()` moves backlog → pending. Now visible to devs. @@ -65,9 +109,38 @@ roboco_task_create( ### 7. PAUSE + IDLE `roboco_task_pause()` with checkpoint, then `roboco_agent_idle()`. -### 8. MONITOR +### 8. MONITOR + HANDLE BLOCKERS When respawned: scan, read journals, update progress, handle blockers. +**CRITICAL: When you resolve a blocker, you MUST call `roboco_task_unblock()`!** + +Blocker resolution workflow: +1. Developer calls `roboco_task_block()` → task status becomes `blocked` +2. Developer escalates to you with `roboco_task_escalate()` +3. You receive notification and investigate +4. You fix the issue (create branch, resolve dependency, etc.) +5. **YOU MUST CALL `roboco_task_unblock(task_id, resolution_notes)`** +6. Task returns to `in_progress`, developer is notified and respawned + +```python +# After resolving a blocker: +roboco_task_unblock( + task_id="...", + resolution="Created missing branch manually. Developer can now proceed." +) +``` + +**DO NOT:** +- ❌ Just message the developer and hope they figure it out +- ❌ Create new duplicate tasks instead of unblocking +- ❌ Move tasks to random statuses manually +- ❌ Claim the blocked task yourself (it's assigned to the developer!) + +**DO:** +- ✅ Fix the root cause +- ✅ Call `roboco_task_unblock()` with clear resolution notes +- ✅ The system will notify and respawn the developer automatically + ### 9. REVIEW PR (Git Tasks) When subtasks reach `awaiting_pm_review`: 1. Review the PR: `roboco_git_diff(project_slug)` to see changes diff --git a/agents/prompts/roles/developer.md b/agents/prompts/roles/developer.md index c09f1e32..1c98fbfc 100644 --- a/agents/prompts/roles/developer.md +++ b/agents/prompts/roles/developer.md @@ -4,6 +4,27 @@ You implement features, fix bugs, and write code. For communication structure: `roboco_kb_search("communication hierarchy")` +## CRITICAL: Tool Availability Check + +**Before starting work, verify your MCP tools are available.** + +At session start, you receive an `init` message with `mcp_servers` status. Check it: +- If `roboco-task` shows `"status":"failed"` → Task tools unavailable +- If `roboco-message` shows `"status":"failed"` → Messaging tools unavailable + +**If critical tools are unavailable:** +1. Check `roboco_notify_list()` - notifications should still work +2. Your task assignment notification contains: task ID, title, description +3. Use the notification body to understand your assignment +4. If task tools are unavailable but git tools work: + - Get your workspace: `roboco_workspace_ensure(project_slug)` + - Use git tools to start work: `roboco_git_status()`, `roboco_git_commit()` +5. **Report the issue via notification** - the system needs to know tools failed + +**DO NOT spin endlessly if tools are missing.** Report and request help. + +--- + ## Workflow **Phase 1: Development** diff --git a/agents/prompts/roles/main_pm.md b/agents/prompts/roles/main_pm.md index 6474e9a1..a27c9b33 100644 --- a/agents/prompts/roles/main_pm.md +++ b/agents/prompts/roles/main_pm.md @@ -85,8 +85,16 @@ roboco_task_create( ### 7. PAUSE + IDLE `roboco_task_pause()` with checkpoint, then `roboco_agent_idle()`. -### 8. MONITOR -When respawned: scan, read Cell PM journals, update progress, coordinate if blockers. +### 8. MONITOR + HANDLE ESCALATIONS +When respawned: scan, read Cell PM journals, update progress, handle escalated blockers. + +**When Cell PM escalates a blocker to you:** +1. Investigate the root cause +2. Fix the issue (infrastructure, permissions, cross-cell coordination, etc.) +3. **Call `roboco_task_unblock(task_id, resolution_notes)` on the blocked task** +4. The system will notify and respawn the affected agents + +**DO NOT just send a message and hope they figure it out. CALL UNBLOCK.** ### 9. REVIEW PR (Git Tasks) When cell tasks reach `awaiting_pm_review` and all subtasks are merged: diff --git a/alembic/versions/001_initial_schema.py b/alembic/versions/001_initial_schema.py index c5a7303a..5c36bc47 100644 --- a/alembic/versions/001_initial_schema.py +++ b/alembic/versions/001_initial_schema.py @@ -338,6 +338,12 @@ def upgrade() -> None: ), sa.Column("updated_at", sa.DateTime(), nullable=True), sa.Column("claimed_at", sa.DateTime(), nullable=True), + sa.Column( + "claimed_by", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("agents.id", ondelete="SET NULL"), + nullable=True, + ), sa.Column("started_at", sa.DateTime(), nullable=True), sa.Column("completed_at", sa.DateTime(), nullable=True), sa.Column("target_date", sa.DateTime(), nullable=True), @@ -865,6 +871,128 @@ def upgrade() -> None: sa.UniqueConstraint("task_id", name="uq_handoffs_task_id"), ) + # ========================================================================== + # A2A CONVERSATIONS TABLE + # ========================================================================== + op.create_table( + "a2a_conversations", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + # The pair (canonical order: lexically smaller first) + sa.Column("agent_a", sa.String(50), nullable=False, index=True), + sa.Column("agent_b", sa.String(50), nullable=False, index=True), + # Context + sa.Column("topic", sa.String(255), nullable=True), + sa.Column( + "task_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("tasks.id", ondelete="SET NULL"), + nullable=True, + index=True, + ), + # Status + sa.Column( + "status", + sa.Enum("active", "paused", "closed", name="a2aconversationstatus"), + nullable=False, + server_default="active", + index=True, + ), + sa.Column("resolution", sa.Text(), nullable=True), + # Stats + sa.Column("message_count", sa.Integer(), nullable=False, server_default="0"), + sa.Column("unread_by_a", sa.Integer(), nullable=False, server_default="0"), + sa.Column("unread_by_b", sa.Integer(), nullable=False, server_default="0"), + # Timestamps + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.func.now(), + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.func.now(), + onupdate=sa.func.now(), + ), + sa.Column("last_message_at", sa.DateTime(timezone=True), nullable=True), + ) + + # Unique constraint for pair + topic + op.create_unique_constraint( + "uq_a2a_pair_topic", "a2a_conversations", ["agent_a", "agent_b", "topic"] + ) + + # Composite indexes + op.create_index("ix_a2a_conv_pair", "a2a_conversations", ["agent_a", "agent_b"]) + op.create_index( + "ix_a2a_conv_status_updated", "a2a_conversations", ["status", "updated_at"] + ) + + # ========================================================================== + # A2A MESSAGES TABLE + # ========================================================================== + op.create_table( + "a2a_messages", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + sa.Column( + "conversation_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("a2a_conversations.id", ondelete="CASCADE"), + nullable=False, + index=True, + ), + # Sender + sa.Column("from_agent", sa.String(50), nullable=False, index=True), + # Content + sa.Column("content", sa.Text(), nullable=False), + sa.Column( + "message_kind", + sa.Enum("message", "request", "response", "system", name="a2amessagekind"), + nullable=False, + server_default="message", + ), + # Threading + sa.Column( + "response_to_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("a2a_messages.id", ondelete="SET NULL"), + nullable=True, + ), + sa.Column( + "requires_response", sa.Boolean(), nullable=False, server_default="false" + ), + # Read tracking + sa.Column("read_at", sa.DateTime(timezone=True), nullable=True), + # Timestamps + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.func.now(), + index=True, + ), + # Edit support + sa.Column("edited_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("edit_history", postgresql.JSON(), server_default="[]"), + ) + + # Composite indexes + op.create_index( + "ix_a2a_msg_conv_created", "a2a_messages", ["conversation_id", "created_at"] + ) + op.create_index( + "ix_a2a_msg_from_created", "a2a_messages", ["from_agent", "created_at"] + ) + # Partial index for pending responses + op.create_index( + "ix_a2a_msg_pending", + "a2a_messages", + ["conversation_id"], + postgresql_where=sa.text("requires_response = true"), + ) + # ========================================================================== # PERFORMANCE INDEXES # ========================================================================== @@ -911,21 +1039,34 @@ def upgrade() -> None: op.create_index("ix_tasks_project_status", "tasks", ["project_id", "status"]) -def downgrade() -> None: - # Drop new indexes +def _downgrade_indexes() -> None: + """Drop all indexes.""" + # Performance indexes op.drop_index("ix_tasks_project_status", table_name="tasks") op.drop_index("ix_work_sessions_agent_status", table_name="work_sessions") op.drop_index("ix_work_sessions_project_status", table_name="work_sessions") op.drop_index("ix_projects_active", table_name="projects") op.drop_index("ix_projects_cell", table_name="projects") - - # Drop original indexes op.drop_index("ix_notifications_to_timestamp", table_name="notifications") op.drop_index("ix_messages_channel_timestamp", table_name="messages") op.drop_index("ix_tasks_assigned_status", table_name="tasks") op.drop_index("ix_tasks_team_status", table_name="tasks") - # Drop tables in reverse order + +def _downgrade_a2a() -> None: + """Drop A2A tables and indexes.""" + op.drop_index("ix_a2a_msg_pending", table_name="a2a_messages") + op.drop_index("ix_a2a_msg_from_created", table_name="a2a_messages") + op.drop_index("ix_a2a_msg_conv_created", table_name="a2a_messages") + op.drop_table("a2a_messages") + op.drop_index("ix_a2a_conv_status_updated", table_name="a2a_conversations") + op.drop_index("ix_a2a_conv_pair", table_name="a2a_conversations") + op.drop_constraint("uq_a2a_pair_topic", "a2a_conversations", type_="unique") + op.drop_table("a2a_conversations") + + +def _downgrade_tables() -> None: + """Drop all tables in reverse order.""" op.drop_table("handoffs") op.drop_table("journal_entries") op.drop_table("journals") @@ -936,17 +1077,17 @@ def downgrade() -> None: op.drop_table("sessions") op.drop_table("groups") op.drop_table("channels") - # Drop FKs before dropping tasks op.drop_constraint("fk_agents_current_task", "agents", type_="foreignkey") op.drop_constraint("fk_work_sessions_task", "work_sessions", type_="foreignkey") - op.drop_table("tasks") op.drop_table("work_sessions") op.drop_table("projects") op.drop_table("agents") - # Drop enums + +def _downgrade_enums() -> None: + """Drop all enum types.""" op.execute("DROP TYPE IF EXISTS worksessionstatus") op.execute("DROP TYPE IF EXISTS tasktype") op.execute("DROP TYPE IF EXISTS tasknature") @@ -963,3 +1104,12 @@ def downgrade() -> None: op.execute("DROP TYPE IF EXISTS agentstatus") op.execute("DROP TYPE IF EXISTS team") op.execute("DROP TYPE IF EXISTS agentrole") + op.execute("DROP TYPE IF EXISTS a2amessagekind") + op.execute("DROP TYPE IF EXISTS a2aconversationstatus") + + +def downgrade() -> None: + _downgrade_indexes() + _downgrade_a2a() + _downgrade_tables() + _downgrade_enums() diff --git a/roboco/agent_sdk/server.py b/roboco/agent_sdk/server.py index 7ca640a0..e6bbe3c6 100644 --- a/roboco/agent_sdk/server.py +++ b/roboco/agent_sdk/server.py @@ -69,7 +69,9 @@ async def receive_message(msg: A2AMessage) -> dict[str, str]: Receive A2A message from another agent. Messages are queued by priority for Claude Code to poll. + Also persists to database for conversation history. """ + # Queue for immediate polling if msg.priority == MessagePriority.URGENT: urgent_inbox.append(msg) logger.info( @@ -87,9 +89,42 @@ async def receive_message(msg: A2AMessage) -> dict[str, str]: skill=msg.skill, ) + # Also persist to database for conversation history + # This enables resuming conversations across agent spawns + await _persist_received_message(msg) + return {"status": "queued", "message_id": str(msg.id)} +async def _persist_received_message(msg: A2AMessage) -> None: + """Persist received message to database via main API.""" + try: + async with httpx.AsyncClient() as client: + # First, ensure conversation exists + conv_resp = await client.post( + f"{MAIN_API_URL}/api/v1/a2a/chat/conversations", + json={ + "target_agent": msg.from_agent, + "topic": msg.skill, # Use skill as topic + "task_id": msg.task_id, + "initial_message": msg.content, + "requires_response": False, + }, + headers={"X-Agent-ID": AGENT_ID}, + timeout=5.0, + ) + # Note: 409 conflict is ok - conversation already exists + if conv_resp.status_code not in (200, 201, 409): + logger.warning( + "Failed to persist A2A message", + status_code=conv_resp.status_code, + from_agent=msg.from_agent, + ) + except Exception as e: + # Don't fail receive if persistence fails + logger.warning("Failed to persist A2A message", error=str(e)) + + # ============================================================================= # SEND (to other agents) # ============================================================================= diff --git a/roboco/api/deps.py b/roboco/api/deps.py index c850b6e0..08d9c7e0 100644 --- a/roboco/api/deps.py +++ b/roboco/api/deps.py @@ -121,6 +121,36 @@ async def get_current_agent_id( CurrentAgentId = Annotated[UUID, Depends(get_current_agent_id)] +async def get_current_agent_slug( + x_agent_id: Annotated[str | None, Header()] = None, +) -> str: + """ + Get the current agent slug from request headers. + + Unlike get_current_agent_id, this returns the slug directly without + resolving to UUID. Useful for A2A where we work with agent slugs. + + Args: + x_agent_id: Agent slug from X-Agent-ID header + + Returns: + Agent slug string + + Raises: + HTTPException: If agent ID header is missing + """ + if not x_agent_id: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Missing X-Agent-ID header", + ) + return x_agent_id + + +# Type alias for agent slug dependency +CurrentAgentSlug = Annotated[str, Depends(get_current_agent_slug)] + + async def get_optional_agent_id( db: DbSession, x_agent_id: Annotated[str | None, Header()] = None, diff --git a/roboco/api/routes/a2a.py b/roboco/api/routes/a2a.py index ff9e28d6..41f26291 100644 --- a/roboco/api/routes/a2a.py +++ b/roboco/api/routes/a2a.py @@ -17,14 +17,31 @@ Endpoints: import asyncio import contextlib from collections.abc import AsyncGenerator -from typing import Any +from typing import Annotated, Any -from fastapi import APIRouter, HTTPException, Query, Request, status +from fastapi import APIRouter, Depends, HTTPException, Query, Request, status from fastapi.responses import JSONResponse from sse_starlette import EventSourceResponse -from roboco.api.deps import DbSession +from roboco.api.deps import CurrentAgentSlug, DbSession +from roboco.api.schemas.a2a_chat import ( + ConversationCloseRequest, + ConversationCreateRequest, + ConversationListResponse, + ConversationResponse, + ConversationSummaryResponse, + InboxSummaryResponse, + ListConversationsParams, + ListMessagesParams, + MessageCreateRequest, + MessageListResponse, + MessageResponse, + PairListResponse, + PairResponse, +) +from roboco.enforcement import A2AAccessDeniedError from roboco.models.a2a import ( + A2AConversationStatus, A2ATask, AgentCard, CancelTaskRequest, @@ -32,6 +49,7 @@ from roboco.models.a2a import ( SendMessageRequest, ) from roboco.services.a2a import A2AService +from roboco.utils.converters import require_uuid # Router for A2A API endpoints (mounted at /api/v1/a2a) router = APIRouter() @@ -462,3 +480,342 @@ async def get_agent_card_by_id( ) return card + + +# ============================================================================= +# PERSISTENT A2A CHAT ENDPOINTS +# ============================================================================= +# These endpoints manage persistent conversations stored in the database. + + +@router.get("/chat/inbox") +async def get_inbox_summary( + db: DbSession, + agent_slug: CurrentAgentSlug, +) -> InboxSummaryResponse: + """Get A2A inbox summary for current agent.""" + service = A2AService(db) + summary = await service.get_inbox_summary(agent_slug) + return InboxSummaryResponse( + total_unread=summary.total_unread, + conversations_with_unread=summary.conversations_with_unread, + pending_responses=summary.pending_responses, + unanswered_requests=summary.unanswered_requests, + ) + + +@router.get("/chat/pairs") +async def list_pairs( + db: DbSession, + agent_slug: CurrentAgentSlug, +) -> PairListResponse: + """List unique agent pairs for current agent.""" + service = A2AService(db) + pairs = await service.list_pairs(agent_slug) + return PairListResponse( + items=[ + PairResponse( + agent_a=p.agent_a, + agent_b=p.agent_b, + conversation_count=p.conversation_count, + total_unread=p.total_unread, + last_activity=p.last_activity, + ) + for p in pairs + ], + total=len(pairs), + ) + + +@router.get("/chat/conversations") +async def list_chat_conversations( + db: DbSession, + agent_slug: CurrentAgentSlug, + params: Annotated[ListConversationsParams, Depends()], +) -> ConversationListResponse: + """List A2A conversations for current agent.""" + service = A2AService(db) + + status_filter = None + if params.status: + status_filter = A2AConversationStatus(params.status) + + conversations = await service.list_conversations( + agent_slug=agent_slug, + status=status_filter, + with_agent=params.with_agent, + task_id=params.task_id, + limit=params.limit, + ) + + return ConversationListResponse( + items=[ + ConversationSummaryResponse( + id=require_uuid(c.id), + other_agent=c.other_agent, + topic=c.topic, + task_id=require_uuid(c.task_id) if c.task_id else None, + status=c.status, + message_count=c.message_count, + unread_count=c.unread_count, + last_message_at=c.last_message_at, + last_message_preview=c.last_message_preview, + ) + for c in conversations + ], + total=len(conversations), + ) + + +@router.post("/chat/conversations", status_code=status.HTTP_201_CREATED) +async def create_conversation( + db: DbSession, + agent_slug: CurrentAgentSlug, + data: ConversationCreateRequest, +) -> ConversationResponse: + """Start a new A2A conversation.""" + service = A2AService(db) + + try: + conv = await service.get_or_create_conversation( + agent_a=agent_slug, + agent_b=data.target_agent, + topic=data.topic, + task_id=data.task_id, + ) + except A2AAccessDeniedError as e: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail={ + "error": "A2A_ACCESS_DENIED", + "message": e.message, + "route_hint": e.route_hint, + }, + ) from None + + # Send initial message + await service.send_chat_message( + conversation_id=require_uuid(conv.id), + from_agent=agent_slug, + content=data.initial_message, + options={"requires_response": data.requires_response}, + ) + + # Refresh conversation to get updated stats + refreshed = await service.get_conversation(require_uuid(conv.id), agent_slug) + if refreshed is None: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to retrieve created conversation", + ) + + await db.commit() + + return ConversationResponse( + id=require_uuid(refreshed.id), + agent_a=refreshed.agent_a, + agent_b=refreshed.agent_b, + topic=refreshed.topic, + task_id=require_uuid(refreshed.task_id) if refreshed.task_id else None, + status=refreshed.status, + resolution=refreshed.resolution, + message_count=refreshed.message_count, + unread_by_a=refreshed.unread_by_a, + unread_by_b=refreshed.unread_by_b, + created_at=refreshed.created_at, + updated_at=refreshed.updated_at, + last_message_at=refreshed.last_message_at, + ) + + +@router.get("/chat/conversations/{conversation_id}") +async def get_conversation( + conversation_id: str, + db: DbSession, + agent_slug: CurrentAgentSlug, +) -> ConversationResponse: + """Get a specific conversation.""" + service = A2AService(db) + conv = await service.get_conversation(require_uuid(conversation_id), agent_slug) + + if conv is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Conversation not found: {conversation_id}", + ) + + return ConversationResponse( + id=require_uuid(conv.id), + agent_a=conv.agent_a, + agent_b=conv.agent_b, + topic=conv.topic, + task_id=require_uuid(conv.task_id) if conv.task_id else None, + status=conv.status, + resolution=conv.resolution, + message_count=conv.message_count, + unread_by_a=conv.unread_by_a, + unread_by_b=conv.unread_by_b, + created_at=conv.created_at, + updated_at=conv.updated_at, + last_message_at=conv.last_message_at, + ) + + +@router.post("/chat/conversations/{conversation_id}/close") +async def close_conversation( + conversation_id: str, + db: DbSession, + agent_slug: CurrentAgentSlug, + data: ConversationCloseRequest | None = None, +) -> None: + """Close a conversation.""" + service = A2AService(db) + + try: + await service.close_conversation( + conversation_id=require_uuid(conversation_id), + agent_slug=agent_slug, + resolution=data.resolution if data else None, + ) + except ValueError as e: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(e), + ) from None + + await db.commit() + + +@router.get("/chat/conversations/{conversation_id}/messages") +async def list_chat_messages( + conversation_id: str, + db: DbSession, + agent_slug: CurrentAgentSlug, + params: Annotated[ListMessagesParams, Depends()], +) -> MessageListResponse: + """Get messages in a conversation.""" + service = A2AService(db) + + messages = await service.get_messages( + conversation_id=require_uuid(conversation_id), + agent_slug=agent_slug, + limit=params.limit + 1, # +1 to detect has_more + before=params.before, + ) + + has_more = len(messages) > params.limit + if has_more: + messages = messages[: params.limit] + + return MessageListResponse( + items=[ + MessageResponse( + id=require_uuid(m.id), + conversation_id=require_uuid(m.conversation_id), + from_agent=m.from_agent, + content=m.content, + message_kind=m.message_kind, + response_to_id=( + require_uuid(m.response_to_id) if m.response_to_id else None + ), + requires_response=m.requires_response, + read_at=m.read_at, + created_at=m.created_at, + edited_at=m.edited_at, + ) + for m in messages + ], + total=len(messages), + has_more=has_more, + ) + + +@router.post( + "/chat/conversations/{conversation_id}/messages", + status_code=status.HTTP_201_CREATED, +) +async def send_chat_message( + conversation_id: str, + db: DbSession, + agent_slug: CurrentAgentSlug, + data: MessageCreateRequest, +) -> MessageResponse: + """Send a message in a conversation.""" + service = A2AService(db) + + try: + msg = await service.send_chat_message( + conversation_id=require_uuid(conversation_id), + from_agent=agent_slug, + content=data.content, + options={ + "message_kind": data.message_kind, + "response_to_id": data.response_to_id, + "requires_response": data.requires_response, + }, + ) + except ValueError as e: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(e), + ) from None + + await db.commit() + + return MessageResponse( + id=require_uuid(msg.id), + conversation_id=require_uuid(msg.conversation_id), + from_agent=msg.from_agent, + content=msg.content, + message_kind=msg.message_kind, + response_to_id=require_uuid(msg.response_to_id) if msg.response_to_id else None, + requires_response=msg.requires_response, + read_at=msg.read_at, + created_at=msg.created_at, + edited_at=msg.edited_at, + ) + + +@router.post("/chat/conversations/{conversation_id}/read", status_code=204) +async def mark_read( + conversation_id: str, + db: DbSession, + agent_slug: CurrentAgentSlug, +) -> None: + """Mark all messages in conversation as read.""" + service = A2AService(db) + await service.mark_read(require_uuid(conversation_id), agent_slug) + await db.commit() + + +@router.get("/chat/tasks/{task_id}/conversations") +async def get_task_conversations( + task_id: str, + db: DbSession, + agent_slug: CurrentAgentSlug, +) -> ConversationListResponse: + """Get A2A conversations linked to a specific task.""" + service = A2AService(db) + + conversations = await service.list_conversations( + agent_slug=agent_slug, + task_id=require_uuid(task_id), + ) + + return ConversationListResponse( + items=[ + ConversationSummaryResponse( + id=require_uuid(c.id), + other_agent=c.other_agent, + topic=c.topic, + task_id=require_uuid(c.task_id) if c.task_id else None, + status=c.status, + message_count=c.message_count, + unread_count=c.unread_count, + last_message_at=c.last_message_at, + last_message_preview=c.last_message_preview, + ) + for c in conversations + ], + total=len(conversations), + ) diff --git a/roboco/api/routes/git.py b/roboco/api/routes/git.py index 07cff8ac..b1a175bc 100644 --- a/roboco/api/routes/git.py +++ b/roboco/api/routes/git.py @@ -48,7 +48,6 @@ from roboco.api.schemas.git import ( GitStatusResponse, ) from roboco.exceptions import GitCommandError, GitTimeoutError -from roboco.models.base import TaskStatus from roboco.services.base import NotFoundError, ServiceError, ValidationError from roboco.services.git import get_git_service from roboco.services.project import get_project_service @@ -397,15 +396,9 @@ async def create_branch( await task_service.update(task_uuid, branch_name=branch_name) - children = await task_service.get_subtasks(task_uuid) - for child in children: - if ( - child.status == TaskStatus.BACKLOG - and child.requires_git - and not child.branch_name - ): - child_uuid = UUID(str(child.id)) - await task_service.update(child_uuid, branch_name=branch_name) + # NOTE: Do NOT propagate branch_name to children. + # Each task creates its OWN branch when claimed, forking from parent's branch. + # Children's branches follow hierarchy: parent--child--grandchild await db.commit() except ServiceError as e: diff --git a/roboco/api/schemas/a2a_chat.py b/roboco/api/schemas/a2a_chat.py new file mode 100644 index 00000000..6bd23bc6 --- /dev/null +++ b/roboco/api/schemas/a2a_chat.py @@ -0,0 +1,160 @@ +""" +A2A Chat API Schemas + +Request/response models for persistent A2A conversation endpoints. +""" + +from datetime import datetime +from uuid import UUID + +from pydantic import BaseModel, Field + +from roboco.models.a2a import A2AConversationStatus, A2AMessageKind + +# ============================================================================= +# CONVERSATION SCHEMAS +# ============================================================================= + + +class ConversationCreateRequest(BaseModel): + """Request to create/start a conversation.""" + + target_agent: str = Field(..., description="Agent slug to chat with") + topic: str | None = Field(default=None, description="Optional topic") + task_id: UUID | None = Field(default=None, description="Optional task link") + initial_message: str = Field(..., min_length=1, max_length=10000) + requires_response: bool = Field(default=False) + + +class ConversationCloseRequest(BaseModel): + """Request to close a conversation.""" + + resolution: str | None = Field(default=None, description="Why closing") + + +class ConversationResponse(BaseModel): + """Conversation response.""" + + id: UUID + agent_a: str + agent_b: str + topic: str | None + task_id: UUID | None + status: A2AConversationStatus + resolution: str | None + message_count: int + unread_by_a: int + unread_by_b: int + created_at: datetime + updated_at: datetime + last_message_at: datetime | None + + +class ConversationSummaryResponse(BaseModel): + """Summary for list views.""" + + id: UUID + other_agent: str + topic: str | None + task_id: UUID | None + status: A2AConversationStatus + message_count: int + unread_count: int + last_message_at: datetime | None + last_message_preview: str | None + + +class ConversationListResponse(BaseModel): + """List of conversation summaries.""" + + items: list[ConversationSummaryResponse] + total: int + + +class ListConversationsParams(BaseModel): + """Query params for listing conversations.""" + + status: A2AConversationStatus | None = None + with_agent: str | None = None + task_id: UUID | None = None + limit: int = Field(50, ge=1, le=100) + + +# ============================================================================= +# MESSAGE SCHEMAS +# ============================================================================= + + +class MessageCreateRequest(BaseModel): + """Request to send a message.""" + + content: str = Field(..., min_length=1, max_length=10000) + message_kind: A2AMessageKind = A2AMessageKind.MESSAGE + response_to_id: UUID | None = None + requires_response: bool = False + + +class MessageResponse(BaseModel): + """A2A chat message response.""" + + id: UUID + conversation_id: UUID + from_agent: str + content: str + message_kind: A2AMessageKind + response_to_id: UUID | None + requires_response: bool + read_at: datetime | None + created_at: datetime + edited_at: datetime | None + + +class MessageListResponse(BaseModel): + """List of messages.""" + + items: list[MessageResponse] + total: int + has_more: bool + + +class ListMessagesParams(BaseModel): + """Query params for listing messages.""" + + limit: int = Field(100, ge=1, le=500) + before: datetime | None = None + + +# ============================================================================= +# INBOX SCHEMAS +# ============================================================================= + + +class InboxSummaryResponse(BaseModel): + """Inbox summary.""" + + total_unread: int + conversations_with_unread: int + pending_responses: int + unanswered_requests: int + + +# ============================================================================= +# PAIRS SCHEMAS +# ============================================================================= + + +class PairResponse(BaseModel): + """Agent pair for frontend.""" + + agent_a: str + agent_b: str + conversation_count: int + total_unread: int + last_activity: datetime | None + + +class PairListResponse(BaseModel): + """List of pairs.""" + + items: list[PairResponse] + total: int diff --git a/roboco/api/schemas/tasks.py b/roboco/api/schemas/tasks.py index c8ee6f96..db6c049c 100644 --- a/roboco/api/schemas/tasks.py +++ b/roboco/api/schemas/tasks.py @@ -252,6 +252,7 @@ class TaskResponse(BaseModel): created_at: datetime updated_at: datetime | None claimed_at: datetime | None + claimed_by: UUID | None started_at: datetime | None completed_at: datetime | None target_date: datetime | None @@ -563,6 +564,7 @@ def task_to_response(task: "TaskTable") -> TaskResponse: created_at=task.created_at, updated_at=task.updated_at, claimed_at=task.claimed_at, + claimed_by=to_python_uuid(task.claimed_by), started_at=task.started_at, completed_at=task.completed_at, target_date=task.target_date, diff --git a/roboco/db/tables.py b/roboco/db/tables.py index 3a74edff..e9e7481e 100644 --- a/roboco/db/tables.py +++ b/roboco/db/tables.py @@ -27,6 +27,7 @@ from sqlalchemy.dialects.postgresql import ARRAY, UUID from sqlalchemy.orm import Mapped, mapped_column, relationship from roboco.db.base import Base +from roboco.models.a2a import A2AConversationStatus, A2AMessageKind from roboco.models.base import ( AgentRole, AgentStatus, @@ -206,6 +207,11 @@ class TaskTable(Base): claimed_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True ) + claimed_by: Mapped[UUID | None] = mapped_column( + UUID(as_uuid=True), + ForeignKey("agents.id", ondelete="SET NULL"), + nullable=True, + ) started_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True ) @@ -1225,3 +1231,167 @@ class IndexedDocumentTable(Base): UniqueConstraint("index_type", "source_hash", name="uq_indexed_doc_source"), Index("ix_indexed_docs_type_time", "index_type", "indexed_at"), ) + + +# ============================================================================= +# A2A CONVERSATION TABLE +# ============================================================================= + + +class A2AConversationTable(Base): + """ + Persistent A2A conversation between two agents. + + Uses canonical ordering (agent_a < agent_b) for unique pair identification. + This enables persistent chat history across agent spawns. + """ + + __tablename__ = "a2a_conversations" + + # Identity + id: Mapped[UUID] = mapped_column( + UUID(as_uuid=True), primary_key=True, default=uuid4 + ) + + # The pair (canonical order: lexically smaller first) + agent_a: Mapped[str] = mapped_column(String(50), nullable=False, index=True) + agent_b: Mapped[str] = mapped_column(String(50), nullable=False, index=True) + + # Context + topic: Mapped[str | None] = mapped_column(String(255), nullable=True) + task_id: Mapped[UUID | None] = mapped_column( + UUID(as_uuid=True), + ForeignKey("tasks.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + + # Status + status: Mapped[A2AConversationStatus] = mapped_column( + Enum(A2AConversationStatus), + nullable=False, + default=A2AConversationStatus.ACTIVE, + index=True, + ) + resolution: Mapped[str | None] = mapped_column(Text, nullable=True) + + # Stats + message_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + unread_by_a: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + unread_by_b: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + + # Timestamps + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), default=lambda: datetime.now(UTC), nullable=False + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + default=lambda: datetime.now(UTC), + onupdate=lambda: datetime.now(UTC), + nullable=False, + ) + last_message_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + + # Relationships + task: Mapped["TaskTable | None"] = relationship("TaskTable", lazy="select") + messages: Mapped[list["A2AMessageTable"]] = relationship( + "A2AMessageTable", back_populates="conversation", lazy="select" + ) + + __table_args__ = ( + # Unique pair + topic combination + UniqueConstraint("agent_a", "agent_b", "topic", name="uq_a2a_pair_topic"), + # Ensure canonical ordering + # CheckConstraint("agent_a < agent_b", name="ck_a2a_agent_order"), + # Composite indexes + Index("ix_a2a_conv_pair", "agent_a", "agent_b"), + Index("ix_a2a_conv_status_updated", "status", "updated_at"), + ) + + +# ============================================================================= +# A2A MESSAGE TABLE +# ============================================================================= + + +class A2AMessageTable(Base): + """ + Individual message in persistent A2A conversation. + + Supports threading via response_to_id and read tracking. + """ + + __tablename__ = "a2a_messages" + + # Identity + id: Mapped[UUID] = mapped_column( + UUID(as_uuid=True), primary_key=True, default=uuid4 + ) + conversation_id: Mapped[UUID] = mapped_column( + UUID(as_uuid=True), + ForeignKey("a2a_conversations.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + + # Sender (must be agent_a or agent_b from conversation) + from_agent: Mapped[str] = mapped_column(String(50), nullable=False, index=True) + + # Content + content: Mapped[str] = mapped_column(Text, nullable=False) + message_kind: Mapped[A2AMessageKind] = mapped_column( + Enum(A2AMessageKind), + nullable=False, + default=A2AMessageKind.MESSAGE, + ) + + # Threading + response_to_id: Mapped[UUID | None] = mapped_column( + UUID(as_uuid=True), + ForeignKey("a2a_messages.id", ondelete="SET NULL"), + nullable=True, + ) + requires_response: Mapped[bool] = mapped_column( + Boolean, nullable=False, default=False + ) + + # Read tracking + read_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + + # Timestamps + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + default=lambda: datetime.now(UTC), + nullable=False, + index=True, + ) + + # Edit support + edited_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + edit_history: Mapped[list[dict[str, Any]]] = mapped_column(JSON, default=list) + + # Relationships + conversation: Mapped["A2AConversationTable"] = relationship( + "A2AConversationTable", back_populates="messages" + ) + response_to: Mapped["A2AMessageTable | None"] = relationship( + "A2AMessageTable", remote_side=[id], lazy="select" + ) + + __table_args__ = ( + # Composite indexes for message queries + Index("ix_a2a_msg_conv_created", "conversation_id", "created_at"), + Index("ix_a2a_msg_from_created", "from_agent", "created_at"), + # Partial index for pending responses + Index( + "ix_a2a_msg_pending", + "conversation_id", + postgresql_where=(requires_response.is_(True)), + ), + ) diff --git a/roboco/enforcement/__init__.py b/roboco/enforcement/__init__.py index 3665f87e..c896668e 100644 --- a/roboco/enforcement/__init__.py +++ b/roboco/enforcement/__init__.py @@ -20,6 +20,11 @@ All functions are designed to be imported as needed. Some are internal helpers used by the primary validate_* functions. """ +from roboco.enforcement.a2a_access import ( + A2AAccessDeniedError, + get_a2a_allowed_targets, + validate_a2a_access, +) from roboco.enforcement.channel_access import ( CHANNEL_ACCESS, ChannelAccessDeniedError, @@ -63,6 +68,7 @@ __all__ = [ "CHANNEL_ACCESS", "ROLE_RESTRICTED_TRANSITIONS", "VALID_TRANSITIONS", + "A2AAccessDeniedError", "ChannelAccessDeniedError", "GitContext", "GitRequirementError", @@ -74,6 +80,7 @@ __all__ = [ "can_agent_transition", "can_read_journal", "can_review_task", + "get_a2a_allowed_targets", "get_agent_channels", "get_notification_scope", "get_readable_journals", @@ -81,6 +88,7 @@ __all__ = [ "is_active_state", "is_terminal_state", "is_waiting_state", + "validate_a2a_access", "validate_channel_access", "validate_git_requirements", "validate_journal_access", diff --git a/roboco/enforcement/a2a_access.py b/roboco/enforcement/a2a_access.py new file mode 100644 index 00000000..17a61f0b --- /dev/null +++ b/roboco/enforcement/a2a_access.py @@ -0,0 +1,84 @@ +""" +A2A Access Enforcement + +Validates agent-to-agent communication permissions. +Uses the same communication matrix as channels/notifications. +""" + +from roboco.agents_config import can_a2a_direct, get_a2a_route_hint +from roboco.exceptions import RobocoError + + +class A2AAccessDeniedError(RobocoError): + """Raised when A2A communication is not permitted.""" + + def __init__( + self, + from_agent: str, + to_agent: str, + reason: str, + route_hint: str | None = None, + ): + self.from_agent = from_agent + self.to_agent = to_agent + self.reason = reason + self.route_hint = route_hint + super().__init__( + code="A2A_ACCESS_DENIED", + message=f"{from_agent} cannot A2A with {to_agent}: {reason}", + details={ + "from_agent": from_agent, + "to_agent": to_agent, + "reason": reason, + "route_hint": route_hint, + }, + ) + + +def validate_a2a_access(from_agent: str, to_agent: str) -> bool: + """ + Validate that from_agent can initiate A2A with to_agent. + + Args: + from_agent: The agent initiating the A2A (slug) + to_agent: The target agent (slug) + + Returns: + True if allowed + + Raises: + A2AAccessDeniedError: If A2A not permitted + """ + allowed, error = can_a2a_direct(from_agent, to_agent) + + if not allowed: + route_hint = get_a2a_route_hint(from_agent, to_agent) + raise A2AAccessDeniedError( + from_agent=from_agent, + to_agent=to_agent, + reason=error or "A2A not permitted", + route_hint=route_hint, + ) + + return True + + +def get_a2a_allowed_targets(from_agent: str, all_agents: list[str]) -> list[str]: + """ + Get list of agents that from_agent can A2A with. + + Args: + from_agent: The agent to check permissions for + all_agents: List of all agent slugs to check against + + Returns: + List of agent slugs that from_agent can A2A with + """ + allowed = [] + for target in all_agents: + if target == from_agent: + continue # Can't A2A with yourself + is_allowed, _ = can_a2a_direct(from_agent, target) + if is_allowed: + allowed.append(target) + return allowed diff --git a/roboco/mcp/a2a_server.py b/roboco/mcp/a2a_server.py index 17a7b637..7765e651 100644 --- a/roboco/mcp/a2a_server.py +++ b/roboco/mcp/a2a_server.py @@ -12,6 +12,7 @@ Tools available to ALL agents: import contextlib import os +from dataclasses import dataclass from typing import Any import httpx @@ -278,6 +279,219 @@ async def _handle_check() -> dict[str, Any]: ) +# ============================================================================= +# PERSISTENT CONVERSATION HANDLERS +# ============================================================================= + + +@dataclass +class StartConversationParams: + """Parameters for starting a conversation.""" + + agent_id: str + target_agent: str + message: str + topic: str | None = None + task_id: str | None = None + requires_response: bool = False + + +async def _handle_start_conversation(params: StartConversationParams) -> dict[str, Any]: + """Start or continue a persistent A2A conversation.""" + # Validate target + if params.target_agent not in ALL_AGENTS: + return format_error_response( + "AGENT_NOT_FOUND", + f"Agent '{params.target_agent}' not found.", + ) + + # Check A2A permissions + allowed, error_msg = can_a2a_direct(params.agent_id, params.target_agent) + if not allowed: + return format_error_response( + "A2A_NOT_PERMITTED", + error_msg or f"Cannot A2A {params.target_agent} directly.", + hint=get_a2a_route_hint(params.agent_id, params.target_agent), + ) + + # Call main API to create conversation + async with httpx.AsyncClient() as client: + try: + resp = await client.post( + f"{API_URL}/api/v1/a2a/chat/conversations", + json={ + "target_agent": params.target_agent, + "topic": params.topic, + "task_id": params.task_id, + "initial_message": params.message, + "requires_response": params.requires_response, + }, + headers={"X-Agent-ID": params.agent_id}, + timeout=10.0, + ) + resp.raise_for_status() + data = resp.json() + + return { + "status": "success", + "conversation_id": data.get("id"), + "target_agent": params.target_agent, + "topic": params.topic, + "task_id": params.task_id, + "guidance": ( + f"Started conversation with {params.target_agent}. " + f"Conversation ID: {data.get('id')}" + ), + } + + except httpx.HTTPStatusError as e: + error_detail = e.response.json() if e.response.content else {} + return format_error_response( + error_detail.get("error", "CONVERSATION_FAILED"), + error_detail.get("message", str(e)), + hint=error_detail.get("route_hint"), + ) + except httpx.ConnectError: + return format_error_response( + "API_UNAVAILABLE", + "Main API not available.", + ) + + +async def _handle_list_conversations( + agent_id: str, + status: str | None = None, + with_agent: str | None = None, + task_id: str | None = None, +) -> dict[str, Any]: + """List persistent A2A conversations.""" + params: dict[str, Any] = {} + if status: + params["status"] = status + if with_agent: + params["with_agent"] = with_agent + if task_id: + params["task_id"] = task_id + + async with httpx.AsyncClient() as client: + try: + resp = await client.get( + f"{API_URL}/api/v1/a2a/chat/conversations", + params=params, + headers={"X-Agent-ID": agent_id}, + timeout=10.0, + ) + resp.raise_for_status() + data = resp.json() + + conversations = data.get("items", []) + return { + "conversations": conversations, + "count": len(conversations), + "guidance": ( + f"You have {len(conversations)} conversation(s)." + if conversations + else "No conversations found." + ), + } + + except httpx.ConnectError: + return format_error_response("API_UNAVAILABLE", "Main API not available.") + except httpx.HTTPStatusError as e: + return format_error_response("LIST_FAILED", str(e)) + + +async def _handle_send_chat_message( + agent_id: str, + conversation_id: str, + message: str, + requires_response: bool = False, +) -> dict[str, Any]: + """Send message in existing conversation.""" + async with httpx.AsyncClient() as client: + try: + resp = await client.post( + f"{API_URL}/api/v1/a2a/chat/conversations/{conversation_id}/messages", + json={ + "content": message, + "requires_response": requires_response, + }, + headers={"X-Agent-ID": agent_id}, + timeout=10.0, + ) + resp.raise_for_status() + data = resp.json() + + return { + "status": "success", + "message_id": data.get("id"), + "conversation_id": conversation_id, + "guidance": "Message sent.", + } + + except httpx.HTTPStatusError as e: + return format_error_response("SEND_FAILED", str(e)) + except httpx.ConnectError: + return format_error_response("API_UNAVAILABLE", "Main API not available.") + + +async def _handle_get_inbox(agent_id: str) -> dict[str, Any]: + """Get persistent A2A inbox summary.""" + async with httpx.AsyncClient() as client: + try: + resp = await client.get( + f"{API_URL}/api/v1/a2a/chat/inbox", + headers={"X-Agent-ID": agent_id}, + timeout=10.0, + ) + resp.raise_for_status() + data = resp.json() + + return { + "total_unread": data.get("total_unread", 0), + "conversations_with_unread": data.get("conversations_with_unread", 0), + "pending_responses": data.get("pending_responses", 0), + "unanswered_requests": data.get("unanswered_requests", 0), + "guidance": ( + f"You have {data.get('total_unread', 0)} unread message(s) " + f"in {data.get('conversations_with_unread', 0)} conversation(s)." + ), + } + + except httpx.ConnectError: + return format_error_response("API_UNAVAILABLE", "Main API not available.") + except httpx.HTTPStatusError as e: + return format_error_response("INBOX_FAILED", str(e)) + + +async def _handle_close_conversation( + agent_id: str, + conversation_id: str, + resolution: str | None = None, +) -> dict[str, Any]: + """Close a conversation.""" + async with httpx.AsyncClient() as client: + try: + resp = await client.post( + f"{API_URL}/api/v1/a2a/chat/conversations/{conversation_id}/close", + json={"resolution": resolution} if resolution else {}, + headers={"X-Agent-ID": agent_id}, + timeout=10.0, + ) + resp.raise_for_status() + + return { + "status": "success", + "conversation_id": conversation_id, + "guidance": "Conversation closed.", + } + + except httpx.HTTPStatusError as e: + return format_error_response("CLOSE_FAILED", str(e)) + except httpx.ConnectError: + return format_error_response("API_UNAVAILABLE", "Main API not available.") + + # ============================================================================= # MCP SERVER FACTORY # ============================================================================= @@ -385,6 +599,126 @@ def create_a2a_mcp_server(agent_id: str) -> FastMCP: """ return await _handle_check() + # ========================================================================= + # PERSISTENT CONVERSATION TOOLS + # ========================================================================= + + @mcp.tool() + async def roboco_a2a_start_conversation( + target_agent: str, + message: str, + topic: str | None = None, + task_id: str | None = None, + requires_response: bool = False, + ) -> dict[str, Any]: + """ + Start or continue a persistent A2A conversation. + + Creates a conversation thread that persists across agent spawns. + Messages are stored in the database and can be retrieved later. + + Args: + target_agent: Agent slug to chat with (e.g., "be-pm", "fe-dev-1") + message: Your initial message + topic: Optional topic/subject for the conversation + task_id: Optional task to link this conversation to + requires_response: Set true if you need a response + + Returns: + Conversation details including conversation_id + """ + return await _handle_start_conversation( + StartConversationParams( + agent_id=agent_id, + target_agent=target_agent, + message=message, + topic=topic, + task_id=task_id, + requires_response=requires_response, + ) + ) + + @mcp.tool() + async def roboco_a2a_list_conversations( + status: str | None = None, + with_agent: str | None = None, + task_id: str | None = None, + ) -> dict[str, Any]: + """ + List your A2A conversations. + + Args: + status: Filter by status (active, paused, closed) + with_agent: Filter by specific agent + task_id: Filter by linked task + + Returns: + List of conversation summaries + """ + return await _handle_list_conversations( + agent_id=agent_id, + status=status, + with_agent=with_agent, + task_id=task_id, + ) + + @mcp.tool() + async def roboco_a2a_send( + conversation_id: str, + message: str, + requires_response: bool = False, + ) -> dict[str, Any]: + """ + Send a message in an existing conversation. + + Args: + conversation_id: The conversation to send to + message: Your message content + requires_response: Set true if you need a response + + Returns: + Message details including message_id + """ + return await _handle_send_chat_message( + agent_id=agent_id, + conversation_id=conversation_id, + message=message, + requires_response=requires_response, + ) + + @mcp.tool() + async def roboco_a2a_inbox() -> dict[str, Any]: + """ + Get your A2A inbox summary. + + Shows unread counts, pending responses, and unanswered requests. + + Returns: + Inbox summary with counts + """ + return await _handle_get_inbox(agent_id) + + @mcp.tool() + async def roboco_a2a_close_conversation( + conversation_id: str, + resolution: str | None = None, + ) -> dict[str, Any]: + """ + Close a conversation. + + Args: + conversation_id: The conversation to close + resolution: Optional note about why/how it was resolved + + Returns: + Status confirmation + """ + return await _handle_close_conversation( + agent_id=agent_id, + conversation_id=conversation_id, + resolution=resolution, + ) + return mcp diff --git a/roboco/mcp/project_server.py b/roboco/mcp/project_server.py index 0f059f43..71e435b8 100644 --- a/roboco/mcp/project_server.py +++ b/roboco/mcp/project_server.py @@ -301,7 +301,7 @@ async def _handle_project_update( async def _handle_workspace_ensure( - client: ApiClient, project_slug: str + client: ApiClient, project_slug: str, agent_id: str ) -> dict[str, Any]: """Handle workspace ensure request.""" # First check project exists @@ -337,23 +337,29 @@ async def _handle_workspace_ensure( {"status": git_resp.status_code, "detail": git_resp.text}, ) + # Compute full workspace path for agent + agent_team = get_agent_team(agent_id) + workspace_path = f"/data/workspaces/{project_slug}/{agent_team}/{agent_id}" + git_status = git_resp.json() + guidance = ( + f"Workspace ready at {workspace_path}. " + "Use roboco_git_* MCP tools for git operations." + ) return { "status": "success", "data": { "exists": True, "branch": git_status.get("current_branch"), "has_uncommitted": git_status.get("has_changes", False), + "path": workspace_path, }, - "guidance": ( - "Workspace ready. Use roboco_git_* MCP tools for git operations. " - f"Direct filesystem access at /data/workspaces/{project_slug}/..." - ), + "guidance": guidance, } async def _handle_workspace_status( - client: ApiClient, project_slug: str + client: ApiClient, project_slug: str, agent_id: str ) -> dict[str, Any]: """Handle workspace status request.""" try: @@ -363,11 +369,19 @@ async def _handle_workspace_status( "CONNECTION_ERROR", f"Failed to connect: {type(e).__name__}" ) + # Compute full workspace path for agent + agent_team = get_agent_team(agent_id) + workspace_path = f"/data/workspaces/{project_slug}/{agent_team}/{agent_id}" + if resp.is_status(status.HTTP_404_NOT_FOUND): + not_found_guidance = ( + f"Workspace not found at {workspace_path}. " + "Use roboco_workspace_ensure() to create." + ) return { "status": "success", - "data": {"exists": False}, - "guidance": "Workspace not found. Use roboco_workspace_ensure() to create.", + "data": {"exists": False, "path": workspace_path}, + "guidance": not_found_guidance, } if not resp.ok: @@ -378,6 +392,10 @@ async def _handle_workspace_status( ) git_status = resp.json() + ready_guidance = ( + f"Workspace ready at {workspace_path}. " + "Use roboco_git_* MCP tools for git operations." + ) return { "status": "success", "data": { @@ -386,12 +404,9 @@ async def _handle_workspace_status( "has_uncommitted": git_status.get("has_changes", False), "staged_files": git_status.get("staged_files", []), "unstaged_files": git_status.get("unstaged_files", []), + "path": workspace_path, }, - "guidance": ( - "Workspace ready. Use roboco_git_* MCP tools for git operations " - "(commit, push, branch, etc). Direct filesystem access is available " - f"at /data/workspaces/{project_slug}/... for your agent." - ), + "guidance": ready_guidance, } @@ -499,24 +514,25 @@ def create_project_mcp_server(agent_id: str) -> FastMCP: project_slug: Project to create workspace for Returns: - Workspace path and status + Workspace path and status including full filesystem path """ - return await _handle_workspace_ensure(client, project_slug) + return await _handle_workspace_ensure(client, project_slug, agent_id) @mcp.tool() async def roboco_workspace_status(project_slug: str) -> dict[str, Any]: """ Get your workspace status for a project. - Returns whether workspace exists, current branch, and changes. + Returns whether workspace exists, current branch, changes, and your + full filesystem path for direct file access. Args: project_slug: Project to check workspace for Returns: - Workspace status + Workspace status including path """ - return await _handle_workspace_status(client, project_slug) + return await _handle_workspace_status(client, project_slug, agent_id) # ------------------------------------------------------------------------- # Tools available ONLY to PM/Board/CEO diff --git a/roboco/mcp/tasks/handlers/_helpers.py b/roboco/mcp/tasks/handlers/_helpers.py index a9b7ae2f..28377429 100644 --- a/roboco/mcp/tasks/handlers/_helpers.py +++ b/roboco/mcp/tasks/handlers/_helpers.py @@ -139,18 +139,28 @@ async def validate_task_claimable( Special case: If an agent is already assigned to a pending task (PM assigned it directly to them), they can claim it to transition to 'claimed' status. """ - # ENFORCEMENT: Main PM must delegate code tasks, not execute them + # ENFORCEMENT: ALL PMs must delegate code tasks, not execute them task_type = task.get("task_type", "code") - if agent_role == "main_pm" and task_type == "code": - return format_error_response( - "PM_CANNOT_EXECUTE_CODE", - "Main PM cannot claim code tasks. You coordinate, not execute.", - {"task_type": task_type, "your_role": agent_role}, - hint=( + pm_roles = ("main_pm", "cell_pm") + if agent_role in pm_roles and task_type == "code": + if agent_role == "main_pm": + hint = ( "Create a subtask for the appropriate Cell PM (be-pm, fe-pm, ux-pm) " "using roboco_task_create(parent_task_id=this_task_id, team='backend', " "assigned_to='be-pm'). Then activate it with roboco_task_activate()." - ), + ) + else: # cell_pm + hint = ( + "You cannot claim code tasks. Create a subtask for a developer " + "(be-dev-1, be-dev-2, etc.) using roboco_task_create(parent_task_id=" + "this_task_id, assigned_to='be-dev-1'). Then roboco_task_activate()." + ) + return format_error_response( + "PM_CANNOT_EXECUTE_CODE", + f"{agent_role.replace('_', ' ').title()} cannot claim code tasks. " + "PMs coordinate, developers execute.", + {"task_type": task_type, "your_role": agent_role}, + hint=hint, ) task_status = task.get("status") diff --git a/roboco/mcp/tasks/handlers/management.py b/roboco/mcp/tasks/handlers/management.py index fb59d2f2..11183cd0 100644 --- a/roboco/mcp/tasks/handlers/management.py +++ b/roboco/mcp/tasks/handlers/management.py @@ -317,22 +317,36 @@ def _build_task_payload( ) -> dict[str, Any]: """Build task creation payload from input data. - For subtasks, inherits task_type from parent if not explicitly set to non-default. - If assigning to a PM, task_type defaults to 'planning' (PMs coordinate, not code). + Task type logic: + - If assigning to a PM, task_type defaults to 'planning' (PMs coordinate). + - If assigning to a cell member, task_type stays as specified (default=code). + - For subtasks WITHOUT an assignee, inherits task_type from parent. + - Cell members assigned to subtasks do NOT inherit 'planning' from PM. """ # Determine task_type task_type = input_data.task_type + assignee_role: str | None = None + + if input_data.assigned_to: + assignee_role = get_agent_role(input_data.assigned_to) # If assigning to a PM, default to 'planning' (PMs don't code) - if input_data.assigned_to and task_type == "code": - assignee_role = get_agent_role(input_data.assigned_to) - if assignee_role in ("cell_pm", "main_pm"): - task_type = "planning" + if assignee_role in ("cell_pm", "main_pm") and task_type == "code": + task_type = "planning" - # For subtasks, inherit from parent if still at default - if parent_task and task_type == "code": - parent_type = parent_task.get("task_type", "code") - task_type = parent_type + # For subtasks, inherit from parent ONLY if: + # 1. Has parent task + # 2. task_type is still default ("code") + # 3. NOT assigned to a cell member (they do code work, not planning) + # This prevents developers from incorrectly getting task_type="planning" + # when assigned to subtasks under PM coordination tasks. + should_inherit = ( + parent_task + and task_type == "code" + and assignee_role not in ("developer", "qa", "documenter") + ) + if should_inherit and parent_task is not None: + task_type = parent_task.get("task_type", "code") payload: dict[str, Any] = { "title": input_data.title, @@ -429,6 +443,24 @@ async def _validate_task_create_inputs( if error: return None, error + # ENFORCEMENT: PMs cannot assign code tasks to PMs (including themselves) + caller_role = get_agent_role(agent_id) + pm_roles = ("main_pm", "cell_pm") + is_pm_assigning_code = ( + caller_role in pm_roles + and input_data.task_type == "code" + and assignee + ) + if is_pm_assigning_code and assignee: # assignee guaranteed by condition above + assignee_role = get_agent_role(assignee) + if assignee_role in pm_roles: + return None, format_error_response( + "PM_CANNOT_OWN_CODE_TASKS", + "PMs cannot be assigned code tasks. Assign to a developer.", + {"assignee": assignee, "assignee_role": assignee_role}, + hint="Assign to: be-dev-1, be-dev-2, fe-dev-1, etc.", + ) + # Validate project for git-enabled tasks return await _validate_project(client, input_data) @@ -448,6 +480,24 @@ async def handle_task_create( if parent_resp.ok: parent_task = parent_resp.json() + # HARD ENFORCEMENT: You must CLAIM a task before creating subtasks for it. + # Workflow: SCAN → CLAIM → PLAN → SUBTASKS. No skipping CLAIM. + # Check claimed_by directly - status can be misleading (paused without claim, etc.) + if parent_task: + parent_claimed_by = parent_task.get("claimed_by") + if not parent_claimed_by: + return format_error_response( + "CLAIM_REQUIRED", + "You must CLAIM this task before creating subtasks.", + { + "parent_task_id": parent_task["id"], + "parent_status": parent_task.get("status"), + "parent_claimed_by": parent_claimed_by, + "workflow": "SCAN → CLAIM → PLAN → SUBTASKS", + }, + hint=f"Call roboco_task_claim('{parent_task['id']}') first.", + ) + # GUARDRAIL: If parent requires git, child must also require git (hierarchy) if parent_task and parent_task.get("requires_git") and not input_data.requires_git: return format_error_response( @@ -461,20 +511,6 @@ async def handle_task_create( hint="Set requires_git=True (or omit it, defaults to True).", ) - # GUARDRAIL: Git subtasks need parent to have branch (for forking) - if parent_task and input_data.requires_git: - parent_branch = parent_task.get("branch_name") - if not parent_branch: - return format_error_response( - "PARENT_BRANCH_REQUIRED", - "Parent task must have a branch before creating git subtasks.", - { - "parent_task_id": parent_task["id"], - "parent_status": parent_task.get("status"), - }, - hint="Claim the parent task first. Branch is auto-created on claim.", - ) - payload = _build_task_payload(input_data, project_id, parent_task) try: @@ -494,15 +530,32 @@ async def handle_task_create( task = create_resp.json() + # CRITICAL: Handle assignment failures explicitly - DO NOT silently discard errors + assignment_failed = False + assignment_error_msg = "" if input_data.assigned_to: - assigned_task, _ = await assign_task_to_agent( + assigned_task, assign_error = await assign_task_to_agent( client, task["id"], input_data.assigned_to ) if assigned_task: task = assigned_task + elif assign_error: + # Assignment FAILED - this is critical, don't silently ignore! + assignment_failed = True + assignment_error_msg = assign_error.get("guidance", "Assignment failed") + # Log but continue - task was created but assignment failed guidance = _format_create_guidance(task, input_data.assigned_to) + # Add CRITICAL WARNING if assignment failed + if assignment_failed: + guidance += ( + f"\n\n🚨 CRITICAL: Assignment to '{input_data.assigned_to}' FAILED! " + f"Error: {assignment_error_msg}. " + "Task was created but NOT assigned to intended agent. " + "You may need to manually assign using roboco_task_assign()." + ) + # Check for role mismatch and add warning if found if input_data.assigned_to: role_warning = get_role_mismatch_warning(task, input_data.assigned_to) @@ -512,6 +565,98 @@ async def handle_task_create( return format_task_response(task, "CREATED", guidance) +async def _check_assignment_guardrails( + client: ApiClient, + task: dict[str, Any], + roles: tuple[str, str], + caller_uuid: str | None = None, +) -> dict[str, Any] | None: + """Check guardrails for task assignment. Returns error or None. + + Args: + client: API client + task: Task dict with id, assigned_to, etc. + roles: Tuple of (caller_role, assignee_role) + caller_uuid: UUID of the caller for ownership check + """ + caller_role, assignee_role = roles + task_id = task.get("id") + + # ENFORCEMENT: You cannot reassign a task assigned TO YOU. + # If you need to delegate, create subtasks instead. + task_assigned_to = task.get("assigned_to") + # Normalize to string for comparison (handles UUID objects vs strings) + task_assigned_str = str(task_assigned_to).lower() if task_assigned_to else None + caller_uuid_str = str(caller_uuid).lower() if caller_uuid else None + if task_assigned_str and caller_uuid_str and task_assigned_str == caller_uuid_str: + return format_error_response( + "CANNOT_REASSIGN_OWN_TASK", + "You cannot reassign a task that was assigned to you. " + "Create subtasks to delegate work.", + { + "task_id": task_id, + "assigned_to": task_assigned_to, + "your_id": caller_uuid, + }, + hint=( + f"Use roboco_task_create(parent_task_id='{task_id}', " + "assigned_to='be-dev-1', ...) to create a subtask." + ), + ) + + # ENFORCEMENT: Block Cell PM from directly assigning devs on complex tasks + if caller_role == "cell_pm" and assignee_role == "developer": + complexity = task.get("estimated_complexity", "low") + if complexity in ("medium", "high", "critical"): + # Check if task already has subtasks + try: + subtasks_resp = await client.get(f"/tasks/{task_id}/subtasks") + subtasks = subtasks_resp.json() if subtasks_resp.ok else [] + except Exception: + subtasks = [] + + # Also allow if this task IS a subtask (has parent) + is_subtask = task.get("parent_task_id") is not None + + if not subtasks and not is_subtask: + return format_error_response( + "SUBTASK_REQUIRED", + f"Cannot assign {complexity} complexity task directly to dev. " + "Cell PM must break down the work into subtasks first.", + { + "task_id": task_id, + "complexity": complexity, + "guidance": ( + f"Create subtasks with: roboco_task_create(" + f"parent_task_id='{task_id}', ...) " + "Then assign each subtask to developers." + ), + }, + ) + + # GUARDRAIL: Git tasks need branch before assigning to developers + if ( + task.get("requires_git") + and not task.get("branch_name") + and assignee_role == "developer" + ): + return format_error_response( + "NO_BRANCH_FOR_GIT_TASK", + "Git task must have a branch before assigning to developer.", + { + "task_id": task_id, + "requires_git": True, + "has_branch": False, + }, + hint=( + "Either claim the task first (creates branch), " + "or create subtasks for developers." + ), + ) + + return None + + async def handle_task_assign( client: ApiClient, input_data: TaskAssignInput, agent_id: str ) -> dict[str, Any]: @@ -536,56 +681,14 @@ async def handle_task_assign( if validation_error: return validation_error - # ENFORCEMENT: Block Cell PM from directly assigning devs on complex tasks - # Cell PM must create subtasks first for medium+ complexity work + # Resolve caller UUID for ownership check + caller_uuid = await resolve_agent_uuid_cached(agent_id, client) assignee_role = get_agent_role(input_data.assignee) - if role == "cell_pm" and assignee_role == "developer": - complexity = task.get("estimated_complexity", "low") - if complexity in ("medium", "high", "critical"): - # Check if task already has subtasks - try: - subtasks_resp = await client.get( - f"/tasks/{input_data.task_id}/subtasks" - ) - subtasks = subtasks_resp.json() if subtasks_resp.ok else [] - except Exception: - subtasks = [] - - # Also allow if this task IS a subtask (has parent) - is_subtask = task.get("parent_task_id") is not None - - if not subtasks and not is_subtask: - return format_error_response( - "SUBTASK_REQUIRED", - f"Cannot assign {complexity} complexity task directly to dev. " - "Cell PM must break down the work into subtasks first.", - { - "task_id": task.get("id"), - "complexity": complexity, - "guidance": ( - f"Create subtasks with: roboco_task_create(" - f"parent_task_id='{task.get('id')}', ...) " - "Then assign each subtask to developers." - ), - }, - ) - - # GUARDRAIL: Git tasks need branch before assigning to developers - if task.get("requires_git") and not task.get("branch_name"): - if assignee_role == "developer": - return format_error_response( - "NO_BRANCH_FOR_GIT_TASK", - "Git task must have a branch before assigning to developer.", - { - "task_id": task.get("id"), - "requires_git": True, - "has_branch": False, - }, - hint=( - "Either claim the task first (creates branch), " - "or create subtasks for developers." - ), - ) + guardrail_error = await _check_assignment_guardrails( + client, task, (role, assignee_role), caller_uuid + ) + if guardrail_error: + return guardrail_error assigned_task, assign_error = await assign_task_to_agent( client, input_data.task_id, input_data.assignee diff --git a/roboco/models/a2a.py b/roboco/models/a2a.py index ffe6e855..86f0d20d 100644 --- a/roboco/models/a2a.py +++ b/roboco/models/a2a.py @@ -446,3 +446,155 @@ def a2a_state_to_task_status(a2a_state: A2ATaskState) -> str: A2ATaskState.AUTH_REQUIRED: "blocked", } return mapping.get(a2a_state, "pending") + + +# ============================================================================= +# PERSISTENT A2A CONVERSATION MODELS +# ============================================================================= +# These models represent persistent storage for agent-to-agent conversations. +# Unlike the A2A protocol models above (which are for wire format), these are +# stored in the database to maintain conversation history across agent spawns. + + +class A2AConversationStatus(str, Enum): + """A2A conversation states.""" + + ACTIVE = "active" + PAUSED = "paused" + CLOSED = "closed" + + +class A2AMessageKind(str, Enum): + """Types of persistent A2A messages.""" + + MESSAGE = "message" # Regular message + REQUEST = "request" # Explicit request for help + RESPONSE = "response" # Reply to request + SYSTEM = "system" # System-generated (e.g., "conversation closed") + + +class A2AConversation(RobocoBase): + """ + Persistent A2A conversation between two agents. + + Uses canonical ordering (agent_a < agent_b) for unique pair identification. + Stored in database to maintain history across agent spawns. + """ + + id: str = Field(default_factory=lambda: str(uuid4())) + # The pair (canonical order: lexically smaller first) + agent_a: str = Field(..., description="First agent slug (lexically smaller)") + agent_b: str = Field(..., description="Second agent slug (lexically larger)") + + # Context + topic: str | None = Field(default=None, description="Optional conversation topic") + task_id: str | None = Field(default=None, description="Optional linked task UUID") + + # Status + status: A2AConversationStatus = A2AConversationStatus.ACTIVE + resolution: str | None = Field(default=None, description="Why conversation closed") + + # Stats + message_count: int = Field(default=0, description="Total messages in conversation") + unread_by_a: int = Field(default=0, description="Unread count for agent_a") + unread_by_b: int = Field(default=0, description="Unread count for agent_b") + + # Timestamps + created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) + updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) + last_message_at: datetime | None = None + + def other_agent(self, my_slug: str) -> str: + """Get the other agent's slug.""" + return self.agent_b if my_slug == self.agent_a else self.agent_a + + def my_unread(self, my_slug: str) -> int: + """Get my unread count.""" + return self.unread_by_a if my_slug == self.agent_a else self.unread_by_b + + def is_participant(self, agent_slug: str) -> bool: + """Check if agent is a participant in this conversation.""" + return agent_slug in (self.agent_a, self.agent_b) + + +class A2AChatMessage(RobocoBase): + """ + Individual message in persistent A2A conversation. + + Distinct from A2AMessage (protocol format) - this is for database storage. + """ + + id: str = Field(default_factory=lambda: str(uuid4())) + conversation_id: str = Field(..., description="Parent conversation ID") + + # Sender (must be agent_a or agent_b from conversation) + from_agent: str = Field(..., description="Agent slug who sent this message") + + # Content + content: str = Field(..., description="Message content") + message_kind: A2AMessageKind = Field( + default=A2AMessageKind.MESSAGE, description="Type of message" + ) + + # Threading + response_to_id: str | None = Field( + default=None, description="ID of message this replies to" + ) + requires_response: bool = Field( + default=False, description="Whether sender is waiting for response" + ) + + # Read tracking + read_at: datetime | None = Field( + default=None, description="When other party read this message" + ) + + # Timestamps + created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) + + # Edit support (like existing messages) + edited_at: datetime | None = None + edit_history: list[dict[str, Any]] = Field(default_factory=list) + + +class A2AInboxSummary(RobocoBase): + """Summary of pending A2A for an agent.""" + + total_unread: int = Field(default=0, description="Total unread messages") + conversations_with_unread: int = Field( + default=0, description="Number of conversations with unread messages" + ) + pending_responses: int = Field( + default=0, description="Messages where I'm waiting for response" + ) + unanswered_requests: int = Field( + default=0, description="Messages where they're waiting for my response" + ) + + +class A2AConversationSummary(RobocoBase): + """Summary of an A2A conversation for list views.""" + + id: str + other_agent: str = Field(..., description="The other agent's slug") + topic: str | None = None + task_id: str | None = None + status: A2AConversationStatus + message_count: int + unread_count: int = Field(description="Unread messages for requesting agent") + last_message_at: datetime | None + last_message_preview: str | None = Field( + default=None, description="Truncated last message content" + ) + + +class A2APair(RobocoBase): + """Unique agent pair for frontend display.""" + + agent_a: str + agent_b: str + conversation_count: int = Field( + default=0, description="Number of conversations between this pair" + ) + total_unread: int = Field(default=0, description="Total unread across all convos") + last_activity: datetime | None = None diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 2ac97282..4b3fd8b4 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -2389,17 +2389,21 @@ Begin with step 1: roboco_task_get("{task_id}") async def _dispatch_dev_work(self, client: httpx.AsyncClient) -> None: """ - Dispatch assigned pending work to the assigned agent. + Dispatch assigned work to the assigned agent. - NOTE: This handles PRE-ASSIGNED tasks (assigned by PM) and - needs_revision tasks. New unassigned pending tasks are handled by + NOTE: This handles PRE-ASSIGNED tasks (assigned by PM), + needs_revision tasks, and in_progress tasks where agent is not active + (e.g., after unblock). New unassigned pending tasks are handled by _dispatch_pm_work() which routes them through the PM hierarchy. - Monitors: assigned pending tasks, needs_revision tasks + Monitors: assigned pending tasks, needs_revision tasks, orphaned in_progress Spawns: Any assigned agent (dev, doc, qa) with appropriate prompt """ # Get tasks needing attention - tasks = await self._fetch_tasks(client, ["pending", "needs_revision"]) + # Include in_progress to catch unblocked tasks that need agent respawn + tasks = await self._fetch_tasks( + client, ["pending", "needs_revision", "in_progress"] + ) for task in tasks: team = task.get("team") @@ -2420,6 +2424,22 @@ Begin with step 1: roboco_task_get("{task_id}") ) continue + # For in_progress tasks where agent is NOT active (e.g., after unblock), + # respawn them to continue their work + if task.get("status") == "in_progress" and agent_slug: + if not self._is_agent_active(agent_slug): + logger.info( + "Respawning agent for orphaned in_progress task", + task_id=task["id"], + agent=agent_slug, + ) + await self.spawn_agent( + agent_id=agent_slug, + task_id=task["id"], + initial_prompt=self._build_dev_prompt(task), + ) + continue + # For pending tasks that ARE already assigned (by PM), # spawn the assigned agent with the appropriate prompt if agent_slug and not self._is_agent_active(agent_slug): diff --git a/roboco/services/a2a.py b/roboco/services/a2a.py index 88ac8e48..f96d866a 100644 --- a/roboco/services/a2a.py +++ b/roboco/services/a2a.py @@ -7,6 +7,7 @@ Provides business logic for A2A protocol operations including: - Message handling and routing """ +from datetime import datetime from typing import Any, cast from uuid import UUID @@ -16,11 +17,24 @@ from sqlalchemy.ext.asyncio import AsyncSession from roboco.agents_config import ALL_AGENTS, get_agent_skills, get_agent_team from roboco.config import settings -from roboco.db.tables import AgentTable, TaskTable +from roboco.db.tables import ( + A2AConversationTable, + A2AMessageTable, + AgentTable, + TaskTable, +) +from roboco.enforcement import validate_a2a_access from roboco.events import Event, EventType, get_event_bus from roboco.models.a2a import ( A2AArtifact, + A2AChatMessage, + A2AConversation, + A2AConversationStatus, + A2AConversationSummary, + A2AInboxSummary, A2AMessage, + A2AMessageKind, + A2APair, A2ATask, A2ATaskStatus, AgentCapabilities, @@ -756,3 +770,521 @@ class A2AService: ) except Exception: pass # Don't fail if event bus unavailable + + # ========================================================================= + # PERSISTENT CONVERSATION MANAGEMENT + # ========================================================================= + # These methods handle persistent A2A conversations stored in the database. + # They complement the existing A2A protocol methods above. + + @staticmethod + def _canonical_pair(agent_a: str, agent_b: str) -> tuple[str, str]: + """Return agents in canonical order (lexically smaller first).""" + return (agent_a, agent_b) if agent_a < agent_b else (agent_b, agent_a) + + async def get_or_create_conversation( + self, + agent_a: str, + agent_b: str, + topic: str | None = None, + task_id: UUID | None = None, + ) -> A2AConversation: + """ + Get existing conversation or create new one. + + Args: + agent_a: First agent slug + agent_b: Second agent slug + topic: Optional conversation topic + task_id: Optional task to link + + Returns: + A2AConversation model + + Raises: + A2AAccessDeniedError: If A2A not permitted between agents + """ + # Validate permissions + validate_a2a_access(agent_a, agent_b) + + # Canonical ordering + a, b = self._canonical_pair(agent_a, agent_b) + + # Try to find existing + query = select(A2AConversationTable).where( + A2AConversationTable.agent_a == a, + A2AConversationTable.agent_b == b, + ) + if topic: + query = query.where(A2AConversationTable.topic == topic) + else: + query = query.where(A2AConversationTable.topic.is_(None)) + + result = await self.session.execute(query) + existing = result.scalar_one_or_none() + + if existing: + return self._conv_to_model(existing) + + # Create new conversation + conv = A2AConversationTable( + agent_a=a, + agent_b=b, + topic=topic, + task_id=task_id, + status=A2AConversationStatus.ACTIVE, + ) + self.session.add(conv) + await self.session.flush() + await self.session.refresh(conv) + + logger.info( + "Created A2A conversation", + conversation_id=str(conv.id), + agent_a=a, + agent_b=b, + topic=topic, + ) + + return self._conv_to_model(conv) + + async def get_conversation( + self, + conversation_id: UUID, + agent_slug: str, + ) -> A2AConversation | None: + """ + Get conversation by ID if agent is a participant. + + Args: + conversation_id: Conversation UUID + agent_slug: Agent requesting (must be participant) + + Returns: + A2AConversation or None if not found/not authorized + """ + result = await self.session.execute( + select(A2AConversationTable).where( + A2AConversationTable.id == conversation_id + ) + ) + conv = result.scalar_one_or_none() + + if conv is None: + return None + + # Verify agent is participant + if agent_slug not in (conv.agent_a, conv.agent_b): + return None + + return self._conv_to_model(conv) + + async def list_conversations( + self, + agent_slug: str, + status: A2AConversationStatus | None = None, + with_agent: str | None = None, + task_id: UUID | None = None, + limit: int = 50, + ) -> list[A2AConversationSummary]: + """ + List conversations for an agent. + + Args: + agent_slug: Agent to list for + status: Filter by status + with_agent: Filter by other participant + task_id: Filter by linked task + limit: Max results + + Returns: + List of conversation summaries + """ + from sqlalchemy import or_ + + query = select(A2AConversationTable).where( + or_( + A2AConversationTable.agent_a == agent_slug, + A2AConversationTable.agent_b == agent_slug, + ) + ) + + if status: + query = query.where(A2AConversationTable.status == status) + + if with_agent: + a, b = self._canonical_pair(agent_slug, with_agent) + query = query.where( + A2AConversationTable.agent_a == a, + A2AConversationTable.agent_b == b, + ) + + if task_id: + query = query.where(A2AConversationTable.task_id == task_id) + + query = query.order_by(A2AConversationTable.updated_at.desc()).limit(limit) + + result = await self.session.execute(query) + conversations = result.scalars().all() + + summaries = [] + for conv in conversations: + # Get last message preview + msg_query = ( + select(A2AMessageTable) + .where(A2AMessageTable.conversation_id == conv.id) + .order_by(A2AMessageTable.created_at.desc()) + .limit(1) + ) + msg_result = await self.session.execute(msg_query) + last_msg = msg_result.scalar_one_or_none() + + other = conv.agent_b if agent_slug == conv.agent_a else conv.agent_a + unread = ( + conv.unread_by_a if agent_slug == conv.agent_a else conv.unread_by_b + ) + + summaries.append( + A2AConversationSummary( + id=str(conv.id), + other_agent=other, + topic=conv.topic, + task_id=str(conv.task_id) if conv.task_id else None, + status=conv.status, + message_count=conv.message_count, + unread_count=unread, + last_message_at=conv.last_message_at, + last_message_preview=(last_msg.content[:100] if last_msg else None), + ) + ) + + return summaries + + async def close_conversation( + self, + conversation_id: UUID, + agent_slug: str, + resolution: str | None = None, + ) -> None: + """Close a conversation.""" + result = await self.session.execute( + select(A2AConversationTable).where( + A2AConversationTable.id == conversation_id + ) + ) + conv = result.scalar_one_or_none() + + if conv is None: + raise ValueError(f"Conversation not found: {conversation_id}") + + if agent_slug not in (conv.agent_a, conv.agent_b): + raise ValueError("Not a participant in this conversation") + + conv.status = A2AConversationStatus.CLOSED + conv.resolution = resolution + await self.session.flush() + + logger.info( + "Closed A2A conversation", + conversation_id=str(conversation_id), + by_agent=agent_slug, + ) + + async def send_chat_message( + self, + conversation_id: UUID, + from_agent: str, + content: str, + options: dict[str, Any] | None = None, + ) -> A2AChatMessage: + """ + Send message in conversation. + + Args: + conversation_id: Target conversation + from_agent: Sender slug + content: Message content + options: Optional dict with message_kind, response_to_id, requires_response + + Returns: + Created A2AChatMessage + + Raises: + ValueError: If conversation not found or sender not participant + """ + from datetime import UTC, datetime + + opts = options or {} + message_kind = opts.get("message_kind", A2AMessageKind.MESSAGE) + response_to_id = opts.get("response_to_id") + requires_response = opts.get("requires_response", False) + + result = await self.session.execute( + select(A2AConversationTable).where( + A2AConversationTable.id == conversation_id + ) + ) + conv = result.scalar_one_or_none() + + if conv is None: + raise ValueError(f"Conversation not found: {conversation_id}") + + if from_agent not in (conv.agent_a, conv.agent_b): + raise ValueError("Not a participant in this conversation") + + # Create message + msg = A2AMessageTable( + conversation_id=conversation_id, + from_agent=from_agent, + content=content, + message_kind=message_kind, + response_to_id=response_to_id, + requires_response=requires_response, + ) + self.session.add(msg) + + # Update conversation stats + conv.message_count += 1 + conv.last_message_at = datetime.now(UTC) + + # Update unread count for the OTHER agent + if from_agent == conv.agent_a: + conv.unread_by_b += 1 + else: + conv.unread_by_a += 1 + + await self.session.flush() + await self.session.refresh(msg) + + logger.info( + "Sent A2A chat message", + conversation_id=str(conversation_id), + message_id=str(msg.id), + from_agent=from_agent, + ) + + return self._msg_to_model(msg) + + async def get_messages( + self, + conversation_id: UUID, + agent_slug: str, + limit: int = 100, + before: datetime | None = None, + ) -> list[A2AChatMessage]: + """Get messages in conversation.""" + # Verify access + conv_result = await self.session.execute( + select(A2AConversationTable).where( + A2AConversationTable.id == conversation_id + ) + ) + conv = conv_result.scalar_one_or_none() + + if conv is None: + return [] + + if agent_slug not in (conv.agent_a, conv.agent_b): + return [] + + query = ( + select(A2AMessageTable) + .where(A2AMessageTable.conversation_id == conversation_id) + .order_by(A2AMessageTable.created_at.desc()) + .limit(limit) + ) + + if before: + query = query.where(A2AMessageTable.created_at < before) + + result = await self.session.execute(query) + messages = result.scalars().all() + + # Return in chronological order + return [self._msg_to_model(m) for m in reversed(list(messages))] + + async def mark_read( + self, + conversation_id: UUID, + agent_slug: str, + ) -> None: + """Mark all messages in conversation as read by agent.""" + from datetime import UTC, datetime + + result = await self.session.execute( + select(A2AConversationTable).where( + A2AConversationTable.id == conversation_id + ) + ) + conv = result.scalar_one_or_none() + + if conv is None: + return + + if agent_slug not in (conv.agent_a, conv.agent_b): + return + + # Reset unread count + if agent_slug == conv.agent_a: + conv.unread_by_a = 0 + else: + conv.unread_by_b = 0 + + # Mark messages as read using SQLAlchemy update statement + from sqlalchemy import update + + stmt = ( + update(A2AMessageTable) + .where(A2AMessageTable.conversation_id == conversation_id) + .where(A2AMessageTable.from_agent != agent_slug) + .where(A2AMessageTable.read_at.is_(None)) + .values(read_at=datetime.now(UTC)) + ) + await self.session.execute(stmt) + + await self.session.flush() + + async def get_inbox_summary(self, agent_slug: str) -> A2AInboxSummary: + """Get summary of pending A2A for agent.""" + from sqlalchemy import func, or_ + + # Get conversations with unread + conv_query = select(A2AConversationTable).where( + or_( + A2AConversationTable.agent_a == agent_slug, + A2AConversationTable.agent_b == agent_slug, + ) + ) + conv_result = await self.session.execute(conv_query) + conversations = conv_result.scalars().all() + + total_unread = 0 + conversations_with_unread = 0 + + for conv in conversations: + unread = ( + conv.unread_by_a if agent_slug == conv.agent_a else conv.unread_by_b + ) + if unread > 0: + conversations_with_unread += 1 + total_unread += unread + + # Count pending responses (messages I sent that require response) + pending_query = ( + select(func.count()) + .select_from(A2AMessageTable) + .where( + A2AMessageTable.from_agent == agent_slug, + A2AMessageTable.requires_response.is_(True), + ) + ) + pending_result = await self.session.execute(pending_query) + pending_responses = pending_result.scalar() or 0 + + # Count unanswered requests (messages to me that require response) + unanswered_query = ( + select(func.count()) + .select_from(A2AMessageTable) + .join(A2AConversationTable) + .where( + or_( + A2AConversationTable.agent_a == agent_slug, + A2AConversationTable.agent_b == agent_slug, + ), + A2AMessageTable.from_agent != agent_slug, + A2AMessageTable.requires_response.is_(True), + ) + ) + unanswered_result = await self.session.execute(unanswered_query) + unanswered_requests = unanswered_result.scalar() or 0 + + return A2AInboxSummary( + total_unread=total_unread, + conversations_with_unread=conversations_with_unread, + pending_responses=pending_responses, + unanswered_requests=unanswered_requests, + ) + + async def list_pairs(self, agent_slug: str) -> list[A2APair]: + """List unique agent pairs for frontend display.""" + from sqlalchemy import or_ + + query = ( + select(A2AConversationTable) + .where( + or_( + A2AConversationTable.agent_a == agent_slug, + A2AConversationTable.agent_b == agent_slug, + ) + ) + .order_by(A2AConversationTable.updated_at.desc()) + ) + + result = await self.session.execute(query) + conversations = result.scalars().all() + + # Group by pair + pairs: dict[tuple[str, str], A2APair] = {} + for conv in conversations: + pair_key = (conv.agent_a, conv.agent_b) + if pair_key not in pairs: + pairs[pair_key] = A2APair( + agent_a=conv.agent_a, + agent_b=conv.agent_b, + conversation_count=0, + total_unread=0, + last_activity=None, + ) + + pairs[pair_key].conversation_count += 1 + + unread = ( + conv.unread_by_a if agent_slug == conv.agent_a else conv.unread_by_b + ) + pairs[pair_key].total_unread += unread + + current_activity = pairs[pair_key].last_activity + if current_activity is None or ( + conv.updated_at is not None and conv.updated_at > current_activity + ): + pairs[pair_key].last_activity = conv.updated_at + + return list(pairs.values()) + + # ========================================================================= + # MODEL CONVERSIONS + # ========================================================================= + + def _conv_to_model(self, conv: A2AConversationTable) -> A2AConversation: + """Convert table row to Pydantic model.""" + return A2AConversation( + id=str(conv.id), + agent_a=conv.agent_a, + agent_b=conv.agent_b, + topic=conv.topic, + task_id=str(conv.task_id) if conv.task_id else None, + status=conv.status, + resolution=conv.resolution, + message_count=conv.message_count, + unread_by_a=conv.unread_by_a, + unread_by_b=conv.unread_by_b, + created_at=conv.created_at, + updated_at=conv.updated_at, + last_message_at=conv.last_message_at, + ) + + def _msg_to_model(self, msg: A2AMessageTable) -> A2AChatMessage: + """Convert table row to Pydantic model.""" + return A2AChatMessage( + id=str(msg.id), + conversation_id=str(msg.conversation_id), + from_agent=msg.from_agent, + content=msg.content, + message_kind=msg.message_kind, + response_to_id=str(msg.response_to_id) if msg.response_to_id else None, + requires_response=msg.requires_response, + read_at=msg.read_at, + created_at=msg.created_at, + edited_at=msg.edited_at, + edit_history=msg.edit_history or [], + ) diff --git a/roboco/services/git.py b/roboco/services/git.py index 4b50751d..3f4c5ed9 100644 --- a/roboco/services/git.py +++ b/roboco/services/git.py @@ -350,12 +350,20 @@ class GitService(BaseService): if not base_branch: project_service = get_project_service(self.session) project = await project_service.get_by_slug(request.project_slug) - base_branch = str(project.default_branch) if project else "main" + base_branch = ( + str(project.default_branch) + if project and project.default_branch + else "main" + ) # Validate parent branch exists on remote (unless it's the default branch) project_service = get_project_service(self.session) project = await project_service.get_by_slug(request.project_slug) - default_branch = str(project.default_branch) if project else "main" + default_branch = ( + str(project.default_branch) + if project and project.default_branch + else "main" + ) if base_branch != default_branch: # Parent is not the default branch - verify it exists on remote @@ -544,7 +552,11 @@ class GitService(BaseService): # Determine target branch and get project token project_service = get_project_service(self.session) project = await project_service.get_by_slug(request.project_slug) - default_branch = str(project.default_branch) if project else "main" + default_branch = ( + str(project.default_branch) + if project and project.default_branch + else "main" + ) # Get decrypted token from project (required for PR creation) git_token = await project_service.get_decrypted_token_by_slug( @@ -663,7 +675,11 @@ class GitService(BaseService): # Get target branch project = await project_service.get_by_slug(project_slug) - target_branch = str(project.default_branch) if project else "main" + target_branch = ( + str(project.default_branch) + if project and project.default_branch + else "main" + ) # Get merge commit await self._run_git(workspace, ["checkout", target_branch]) diff --git a/roboco/services/messaging.py b/roboco/services/messaging.py index c47eeac9..abcc495d 100644 --- a/roboco/services/messaging.py +++ b/roboco/services/messaging.py @@ -17,7 +17,7 @@ from uuid import UUID from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession -from sqlalchemy.orm import joinedload +from sqlalchemy.orm import joinedload, selectinload from roboco.db.tables import ( ChannelTable, @@ -639,6 +639,61 @@ class MessagingService(BaseService): ) return list(result.scalars().all()) + async def _resolve_group_from_parent_tasks( + self, + task_ids: list[UUID], + ) -> GroupTable | None: + """ + Resolve group by looking at parent tasks' sessions. + + When creating a session for subtasks, inherit the group from + the parent task's session. This maintains proper hierarchy. + + Args: + task_ids: Task IDs to check for parent sessions + + Returns: + Group from parent task's session, or None if no parent has a session + """ + from roboco.db.tables import TaskTable + + for task_id in task_ids: + # Get the task to find its parent + task_result = await self.session.execute( + select(TaskTable).where(TaskTable.id == task_id) + ) + task = task_result.scalar_one_or_none() + + if not task or not task.parent_task_id: + continue + + # Find parent task's primary session + parent_session_link = await self.session.execute( + select(SessionTaskTable) + .where( + SessionTaskTable.task_id == task.parent_task_id, + SessionTaskTable.is_primary.is_(True), + ) + .options( + selectinload(SessionTaskTable.session).selectinload( + SessionTable.group + ) + ) + ) + link = parent_session_link.scalar_one_or_none() + + if link and link.session and link.session.group: + self.log.info( + "Inherited group from parent task's session", + task_id=str(task_id), + parent_task_id=str(task.parent_task_id), + group_id=str(link.session.group.id), + group_name=link.session.group.name, + ) + return link.session.group + + return None + async def create_session_for_tasks( self, req: SessionForTasksCreate, @@ -665,8 +720,11 @@ class MessagingService(BaseService): if not channel: raise NotFoundError(f"Channel '{req.channel_slug}' not found") - # Use provided group_id or fall back to first group in channel + # Resolve group: explicit > inherited from parent > fallback to first + group: GroupTable | None = None + if req.group_id: + # Explicit group_id provided group_result = await self.session.execute( select(GroupTable).where(GroupTable.id == req.group_id) ) @@ -674,10 +732,24 @@ class MessagingService(BaseService): if not group: raise NotFoundError(f"Group '{req.group_id}' not found") else: - groups = await self.list_groups_in_channel(cast("UUID", channel.id)) - if not groups: - raise ValueError(f"No groups found in channel '{req.channel_slug}'") - group = groups[0] + # Try to inherit group from parent task's session + group = await self._resolve_group_from_parent_tasks(req.task_ids) + + if not group: + # Fall back to first group in channel (last resort) + groups = await self.list_groups_in_channel(cast("UUID", channel.id)) + if not groups: + raise ValueError( + f"No groups found in channel '{req.channel_slug}'. " + "Create a group first or specify group_id." + ) + group = groups[0] + self.log.warning( + "Session created without explicit group, using first group", + channel_slug=req.channel_slug, + group_name=group.name, + task_ids=[str(t) for t in req.task_ids], + ) # Create session with config and scope session_req = SessionCreateRequest( diff --git a/roboco/services/task.py b/roboco/services/task.py index 2d25a7a6..62c4caca 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -395,6 +395,46 @@ class TaskService(BaseService): return await self._auto_create_branch(task, agent_id) + async def _find_ancestor_branch(self, task: TaskTable) -> str | None: + """Walk up task hierarchy to find nearest ancestor with a branch. + + This handles cases where immediate parent doesn't have a branch + (e.g., planning tasks created by PMs that don't need branches). + + Returns: + Branch name of nearest ancestor, or None if no ancestor has one. + """ + current_parent_id = task.parent_task_id + visited: set[str] = set() # Prevent infinite loops + + while current_parent_id: + parent_id_str = str(current_parent_id) + if parent_id_str in visited: + self.log.warning( + "Circular reference detected in task hierarchy", + task_id=str(task.id), + cycle_at=parent_id_str, + ) + break + visited.add(parent_id_str) + + parent = await self.get(UUID(parent_id_str)) + if not parent: + break + + if parent.branch_name: + self.log.info( + "Found ancestor branch", + task_id=str(task.id), + ancestor_id=parent_id_str, + branch=str(parent.branch_name), + ) + return str(parent.branch_name) + + current_parent_id = parent.parent_task_id + + return None + async def _auto_create_branch( self, task: TaskTable, @@ -427,16 +467,23 @@ class TaskService(BaseService): if not project: raise ValueError(f"Project {task.project_id} not found") + # Resolve parent branch - walk up hierarchy to find nearest ancestor with branch parent_branch: str | None = None if task.parent_task_id: - parent_task = await self.get(UUID(str(task.parent_task_id))) - if parent_task and parent_task.branch_name: - parent_branch = str(parent_task.branch_name) - else: - raise ValueError( - "Parent task must be claimed first. " - "Subtasks fork from parent branch (auto-created on claim)." - ) + parent_branch = await self._find_ancestor_branch(task) + + # Fallback to project default branch if no ancestor has a branch + # This handles cases like: Root planning task (no branch) → Dev subtask + if not parent_branch: + default_branch = ( + str(project.default_branch) if project.default_branch else "main" + ) + self.log.info( + "No ancestor branch found, using project default", + task_id=str(task.id), + default_branch=default_branch, + ) + parent_branch = default_branch workspace = await git_service.get_workspace(project.slug, agent_id) @@ -672,8 +719,9 @@ class TaskService(BaseService): # Set context for QA/Documenter claims (only if not already set) self._set_original_developer_context(task, agent) - # Update assignment + # Update assignment and claim tracking task.assigned_to = cast("Any", agent_id) + task.claimed_by = cast("Any", agent_id) task.claimed_at = datetime.now(UTC) # Transition to CLAIMED - validated with role for proper enforcement