Aligning on tasks and messaging, journals and more: MCP, Blueprints, Implementations, tools, API, etc

roboco/mcp/message_server.py

@@ -311,23 +311,55 @@ async def _handle_channel_history(
     }
 
 
-async def _get_task_primary_session(client: ApiClient, task_id: str) -> str | None:
-    """Get the primary session ID for a task, if one exists."""
-    resp = await client.get(f"/sessions/for-task/{task_id}")
-    if not resp.ok:
-        return None
+async def _get_task_primary_session(
+    client: ApiClient, task_id: str, max_depth: int = 5
+) -> str | None:
+    """Get the primary session ID for a task.
 
-    sessions = resp.json()
-    if not sessions:
-        return None
+    If the task has no session, traverses up the parent hierarchy
+    to find the parent's session. Subtasks inherit their parent's session.
 
-    # Find primary session
-    for session in sessions:
-        if session.get("is_primary"):
-            return str(session.get("session_id"))
+    Args:
+        client: API client
+        task_id: The task to find session for
+        max_depth: Maximum parent levels to traverse (prevents infinite loops)
 
-    # Fall back to first session if no primary marked
-    return str(sessions[0].get("session_id")) if sessions else None
+    Returns:
+        Session ID or None if no session found in hierarchy
+    """
+    current_task_id = task_id
+    depth = 0
+
+    while current_task_id and depth < max_depth:
+        # Check if this task has a session
+        resp = await client.get(f"/sessions/for-task/{current_task_id}")
+        if resp.ok:
+            sessions = resp.json()
+            if sessions:
+                # Find primary session
+                for session in sessions:
+                    if session.get("is_primary"):
+                        return str(session.get("session_id"))
+                # Fall back to first session if no primary marked
+                return str(sessions[0].get("session_id"))
+
+        # No session found - check if this is a subtask with a parent
+        task_resp = await client.get(f"/tasks/{current_task_id}")
+        if not task_resp.ok:
+            return None
+
+        task_data = task_resp.json()
+        parent_id = task_data.get("parent_task_id")
+
+        if not parent_id:
+            # No parent - we've reached the top without finding a session
+            return None
+
+        # Traverse up to parent
+        current_task_id = parent_id
+        depth += 1
+
+    return None
 
 
 async def _handle_message_send(
@@ -345,19 +377,19 @@ async def _handle_message_send(
     ):
         return validation_error
 
-    # task_id is required - use task's linked session
+    # task_id is required - use task's linked session (or parent's session for subtasks)
     session_id = await _get_task_primary_session(client, data.task_id)
     if not session_id:
-        # Task has no linked session - PM setup issue
+        # Task has no linked session and no parent with session - PM setup issue
         return format_error_response(
             "NO_TASK_SESSION",
-            f"Task {data.task_id} has no linked session.",
+            f"Task {data.task_id} has no linked session (checked parent hierarchy).",
             {
                 "guidance": (
-                    "This task doesn't have a work session yet.\n"
-                    "Cell PM must create one using "
-                    "roboco_session_create_for_tasks.\n"
-                    "Escalate to your PM if you need a session for this task."
+                    "Neither this task nor its parent have a work session.\n"
+                    "Cell PM must create one with roboco_session_create_for_tasks\n"
+                    "for the PARENT task before subtasks can be worked on.\n"
+                    "Escalate to your PM using roboco_task_escalate."
                 ),
                 "task_id": data.task_id,
             },

roboco/mcp/task_server.py

@@ -219,6 +219,37 @@ def _register_core_tools(mcp: FastMCP, client: ApiClient, agent_id: str) -> None
         """
         return await handle_agent_idle(client, agent_id)
 
+    @mcp.tool()
+    async def roboco_task_escalate(
+        task_id: str, reason: str, escalate_to: str | None = None
+    ) -> dict[str, Any]:
+        """
+        Escalate a task up the management hierarchy.
+
+        Use this when:
+        - Task is blocked by something outside your control
+        - You need PM guidance or decision
+        - Task scope has grown beyond your authority
+        - Cross-team coordination is needed
+
+        Escalation chain:
+        - Developer/QA/Doc -> Cell PM
+        - Cell PM -> Main PM
+        - Main PM -> Product Owner
+
+        Args:
+            task_id: The task UUID to escalate
+            reason: Why this task needs escalation (be specific)
+            escalate_to: Optional specific target (overrides default chain)
+
+        Returns:
+            Task with escalation confirmation
+        """
+        input_data = TaskEscalateInput(
+            task_id=task_id, reason=reason, escalate_to=escalate_to
+        )
+        return await handle_task_escalate(client, input_data, agent_id)
+
 
 def _register_blocking_tools(mcp: FastMCP, client: ApiClient, agent_id: str) -> None:
     """Register blocking/unblocking/pause tools."""
@@ -297,15 +328,18 @@ def _register_blocking_tools(mcp: FastMCP, client: ApiClient, agent_id: str) ->
         return await handle_task_pause(client, data, agent_id)
 
 
-def _register_qa_tools(mcp: FastMCP, client: ApiClient, agent_id: str) -> None:
-    """Register QA and verification tools."""
+def _register_developer_submit_tools(
+    mcp: FastMCP, client: ApiClient, agent_id: str
+) -> None:
+    """Register developer-only submission tools (submit_verification, submit_qa)."""
 
     @mcp.tool()
     async def roboco_task_submit_verification(task_id: str) -> dict[str, Any]:
         """
-        Submit task for self-verification.
+        Submit task for self-verification (developer only).
 
         ENFORCEMENT:
+        - Only developers can use this tool
         - Task must be in 'in_progress' status
         - At least one commit should exist
 
@@ -322,9 +356,10 @@ def _register_qa_tools(mcp: FastMCP, client: ApiClient, agent_id: str) -> None:
         task_id: str, dev_notes: str, handoff_summary: str
     ) -> dict[str, Any]:
         """
-        Submit task for QA review.
+        Submit task for QA review (developer only).
 
         ENFORCEMENT:
+        - Only developers can use this tool
         - Task must be in 'verifying' status
         - Dev notes and handoff summary required
 
@@ -340,6 +375,12 @@ def _register_qa_tools(mcp: FastMCP, client: ApiClient, agent_id: str) -> None:
             client, task_id, dev_notes, handoff_summary, agent_id
         )
 
+
+def _register_qa_verdict_tools(
+    mcp: FastMCP, client: ApiClient, agent_id: str
+) -> None:
+    """Register QA-only verdict tools (qa_pass, qa_fail)."""
+
     @mcp.tool()
     async def roboco_task_qa_pass(task_id: str, qa_notes: str) -> dict[str, Any]:
         """
@@ -381,6 +422,12 @@ def _register_qa_tools(mcp: FastMCP, client: ApiClient, agent_id: str) -> None:
         """
         return await handle_task_qa_fail(client, task_id, qa_notes, issues, agent_id)
 
+
+def _register_documenter_tools(
+    mcp: FastMCP, client: ApiClient, agent_id: str
+) -> None:
+    """Register documenter-only tools (docs_complete)."""
+
     @mcp.tool()
     async def roboco_task_docs_complete(
         task_id: str, doc_notes: str | None = None
@@ -404,6 +451,12 @@ def _register_qa_tools(mcp: FastMCP, client: ApiClient, agent_id: str) -> None:
         """
         return await handle_docs_complete(client, task_id, agent_id, doc_notes)
 
+
+def _register_pm_completion_tools(
+    mcp: FastMCP, client: ApiClient, agent_id: str
+) -> None:
+    """Register PM-only task completion tools."""
+
     @mcp.tool()
     async def roboco_task_complete(task_id: str) -> dict[str, Any]:
         """
@@ -501,37 +554,6 @@ def _register_pm_tools(mcp: FastMCP, client: ApiClient, agent_id: str) -> None:
         """
         return await handle_task_cancel(client, task_id, agent_id, reason)
 
-    @mcp.tool()
-    async def roboco_task_escalate(
-        task_id: str, reason: str, escalate_to: str | None = None
-    ) -> dict[str, Any]:
-        """
-        Escalate a task up the management hierarchy.
-
-        Use this when:
-        - Task is blocked by something outside your control
-        - You need PM guidance or decision
-        - Task scope has grown beyond your authority
-        - Cross-team coordination is needed
-
-        Escalation chain:
-        - Developer/QA/Doc -> Cell PM
-        - Cell PM -> Main PM
-        - Main PM -> Product Owner
-
-        Args:
-            task_id: The task UUID to escalate
-            reason: Why this task needs escalation (be specific)
-            escalate_to: Optional specific target (overrides default chain)
-
-        Returns:
-            Task with escalation confirmation
-        """
-        input_data = TaskEscalateInput(
-            task_id=task_id, reason=reason, escalate_to=escalate_to
-        )
-        return await handle_task_escalate(client, input_data, agent_id)
-
     @mcp.tool()
     async def roboco_task_activate(task_id: str) -> dict[str, Any]:
         """
@@ -687,22 +709,51 @@ def create_task_mcp_server(agent_id: str) -> FastMCP:
     Create a Task MCP server for a specific agent.
 
     The agent_id is embedded in the server to enforce ownership rules.
+    Tools are registered based on role - agents only see tools they can use.
 
     Args:
         agent_id: The agent identifier (e.g., "be-dev-1")
 
     Returns:
-        Configured FastMCP server
+        Configured FastMCP server with role-appropriate tools
     """
+    from roboco.agents_config import get_agent_role
+
     mcp = FastMCP(f"roboco-task-{agent_id}", json_response=True)
     client = ApiClient(agent_id)
+    role = get_agent_role(agent_id)
 
-    # Register all tools via helper functions
+    # Core tools available to ALL agents
     _register_core_tools(mcp, client, agent_id)
-    _register_blocking_tools(mcp, client, agent_id)
-    _register_qa_tools(mcp, client, agent_id)
-    _register_pm_tools(mcp, client, agent_id)
-    _register_session_tools(mcp, client, agent_id)
+
+    # Role-specific tool registration
+    if role == "developer":
+        # Developers: submit workflow + blocking
+        _register_developer_submit_tools(mcp, client, agent_id)
+        _register_blocking_tools(mcp, client, agent_id)
+
+    elif role == "qa":
+        # QA: verdict tools only
+        _register_qa_verdict_tools(mcp, client, agent_id)
+
+    elif role == "documenter":
+        # Documenters: docs completion only
+        _register_documenter_tools(mcp, client, agent_id)
+
+    elif role in ("cell_pm", "main_pm"):
+        # PMs: full management capabilities
+        _register_pm_completion_tools(mcp, client, agent_id)
+        _register_pm_tools(mcp, client, agent_id)
+        _register_session_tools(mcp, client, agent_id)
+        _register_blocking_tools(mcp, client, agent_id)
+
+    elif role in ("product_owner", "head_marketing", "auditor", "ceo"):
+        # Board/Management: PM tools + completion
+        _register_pm_completion_tools(mcp, client, agent_id)
+        _register_pm_tools(mcp, client, agent_id)
+        _register_session_tools(mcp, client, agent_id)
+
+    # Unknown role: only core tools (scan, get, claim, etc.)
 
     return mcp
 

roboco/mcp/tasks/handlers/_helpers.py

@@ -96,7 +96,8 @@ def validate_task_claimable(task: dict, agent_role: str) -> dict[str, Any] | Non
     task_status = task.get("status")
     claimable_statuses = {
         "qa": ["awaiting_qa"],
-        "documenter": ["awaiting_documentation"],
+        # Documenters: pending (direct docs tasks) or awaiting_documentation (workflow)
+        "documenter": ["pending", "awaiting_documentation"],
     }
     allowed = claimable_statuses.get(agent_role, ["pending"])
 

roboco/mcp/tasks/handlers/lifecycle.py

@@ -74,10 +74,24 @@ async def handle_docs_complete(
     )
 
 
+def _is_pm_own_task(task: dict[str, Any], agent_id: str) -> bool:
+    """Check if this is the PM's own task (assigned to them)."""
+    assigned_to = task.get("assigned_to")
+    # Could be UUID or slug - check both patterns
+    return assigned_to == agent_id or (
+        isinstance(assigned_to, str) and agent_id in assigned_to
+    )
+
+
 async def handle_task_complete(
     client: ApiClient, task_id: str, agent_id: str
 ) -> dict[str, Any]:
-    """Handle task completion (PM only)."""
+    """Handle task completion (PM only).
+
+    Two completion paths:
+    1. Completing developer work: task must be in 'awaiting_pm_review'
+    2. Completing PM's own task: task can be in 'in_progress' if assigned to PM
+    """
     if error := _validate_pm_role(agent_id, "complete tasks"):
         return error
 
@@ -86,8 +100,19 @@ async def handle_task_complete(
         return error
     assert task is not None
 
-    if error := validate_task_status(task, "awaiting_pm_review", "complete"):
-        return error
+    current_status = task.get("status")
+
+    # PM completing their own task - allow from in_progress
+    if current_status == "in_progress" and _is_pm_own_task(task, agent_id):
+        pass  # Valid - PM completing their own work
+    # Normal path - developer work went through QA/Docs
+    elif current_status != "awaiting_pm_review":
+        return format_error_response(
+            "INVALID_STATE",
+            f"Cannot complete task in '{current_status}' status. "
+            "Expected 'awaiting_pm_review' (dev work) or 'in_progress' (own task).",
+            {"current_status": current_status},
+        )
 
     complete_resp = await client.post(f"/tasks/{task_id}/complete")
     if not complete_resp.ok:

roboco/mcp/tasks/handlers/review.py

@@ -17,6 +17,19 @@ from roboco.mcp.utils import ApiClient, format_error_response, resolve_agent_uui
 from roboco.services.task import extract_original_developer
 
 
+def _validate_developer_role(agent_id: str) -> dict[str, Any] | None:
+    """Validate agent is a developer (not PM/QA/Documenter). Returns error or None."""
+    agent_role = get_agent_role(agent_id)
+    if agent_role != "developer":
+        return format_error_response(
+            "NOT_DEVELOPER",
+            "Only developers can submit work for verification/QA. "
+            "PMs should use roboco_task_complete() directly.",
+            {"your_role": agent_role, "allowed_roles": ["developer"]},
+        )
+    return None
+
+
 def _has_work_evidence(task: dict[str, Any]) -> bool:
     """Check if task has evidence of work done."""
     return bool(
@@ -30,28 +43,44 @@ def _build_verification_checklist(task: dict[str, Any]) -> str:
     return "\n".join(f"- [ ] {c}" for c in criteria)
 
 
-async def handle_task_submit_verification(
+async def _validate_verification_submission(
     client: ApiClient, task_id: str, agent_id: str
-) -> dict[str, Any]:
-    """Handle task verification submission."""
+) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
+    """Validate task for verification. Returns (task, None) or (None, error)."""
+    # Only developers can submit for verification
+    if error := _validate_developer_role(agent_id):
+        return None, error
+
     task, error = await fetch_task_or_error(client, task_id)
     if error:
-        return error
+        return None, error
     assert task is not None
 
     if error := await validate_task_ownership(task, agent_id, client):
-        return error
+        return None, error
 
     if error := validate_task_status(task, "in_progress", "submit for verification"):
-        return error
+        return None, error
 
     if not _has_work_evidence(task):
-        return format_error_response(
+        return None, format_error_response(
             "NO_WORK_EVIDENCE",
             "No evidence of work found. Add commits with roboco_task_add_commit "
             "or update progress with roboco_task_progress before verification.",
         )
 
+    return task, None
+
+
+async def handle_task_submit_verification(
+    client: ApiClient, task_id: str, agent_id: str
+) -> dict[str, Any]:
+    """Handle task verification submission."""
+    task, error = await _validate_verification_submission(client, task_id, agent_id)
+    if error:
+        return error
+    assert task is not None
+
     verify_resp = await client.post(f"/tasks/{task_id}/verify")
     if not verify_resp.ok:
         return format_error_response(
@@ -132,6 +161,10 @@ async def handle_task_submit_qa(
     agent_id: str,
 ) -> dict[str, Any]:
     """Handle task QA submission."""
+    # Only developers can submit for QA
+    if error := _validate_developer_role(agent_id):
+        return error
+
     if error := _validate_qa_notes(dev_notes, handoff_summary):
         return error