Live Chat A2A + Lifecycle & Workflow updates

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

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**

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:

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()

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)
 # =============================================================================

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,

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),
+    )

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:

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

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,

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)),
+        ),
+    )

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",

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

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
 
 

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

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")

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"):
+    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

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

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):

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 [],
+        )

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])

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:
+            # 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}'")
+                    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(

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