Huge refactoring but stuff is working again; minus some issues here and there.

roboco/mcp/__init__.py

@@ -9,16 +9,19 @@ Servers:
 - Message MCP Server: Channel messaging
 - Notify MCP Server: Formal notifications
 - Journal MCP Server: Personal journaling
+- Optimal MCP Server: Knowledge base and RAG
 """
 
 from roboco.mcp.journal_server import create_journal_mcp_server
 from roboco.mcp.message_server import create_message_mcp_server
 from roboco.mcp.notify_server import create_notify_mcp_server
+from roboco.mcp.optimal_server import create_optimal_mcp_server
 from roboco.mcp.task_server import create_task_mcp_server
 
 __all__ = [
     "create_journal_mcp_server",
     "create_message_mcp_server",
     "create_notify_mcp_server",
+    "create_optimal_mcp_server",
     "create_task_mcp_server",
 ]

roboco/mcp/journal_server.py

@@ -12,6 +12,8 @@ Tools:
 - roboco_journal_struggle: Log a struggle
 - roboco_journal_search: Search past entries
 - roboco_journal_stats: Get journal statistics
+- roboco_journal_recent: Get recent journal entries
+- roboco_journal_read_team: Read team member journals (cell members can read each other)
 """
 
 from typing import Any

roboco/mcp/message_server.py

@@ -6,10 +6,12 @@ enforcement of channel access rules.
 
 Tools:
 - roboco_message_send: Send a message to a channel
-- roboco_message_list: List recent messages
 - roboco_message_get: Get a specific message
 - roboco_channel_list: List available channels
 - roboco_channel_history: Get channel message history
+- roboco_ask_question: Ask a question with structured response options
+- roboco_report_blocker: Report a blocker with details
+- roboco_session_history_for_task: Get message history for a task's session
 """
 
 from datetime import UTC, datetime, timedelta
@@ -19,7 +21,6 @@ from fastapi import status
 from mcp.server.fastmcp import FastMCP
 
 from roboco.agents_config import CHANNEL_ACCESS, get_agent_role
-from roboco.llm import ToonAdapter
 from roboco.mcp.schemas import (
     AskQuestionInput,
     ReportBlockerInput,
@@ -31,10 +32,6 @@ from roboco.mcp.utils import (
     resolve_agent_uuid_cached,
 )
 
-# Global TOON adapter for encoding message data
-_toon = ToonAdapter()
-
-
 # =============================================================================
 # HELPER FUNCTIONS
 # =============================================================================
@@ -397,12 +394,14 @@ async def _handle_message_send(
 
     # Resolve mentions (slugs) to UUIDs using shared cache
     resolved_mentions: list[str] = []
+    skipped_mentions: list[str] = []
     if data.mentions:
         for mention in data.mentions:
             resolved = await resolve_agent_uuid_cached(mention, client)
             if resolved:
                 resolved_mentions.append(resolved)
-            # Skip unresolved mentions rather than failing
+            else:
+                skipped_mentions.append(mention)
 
     message_data = {
         "session_id": session_id,
@@ -421,7 +420,7 @@ async def _handle_message_send(
             "SEND_FAILED", "Failed to send message", {"api_error": resp.text}
         )
 
-    return {
+    result = {
         "status": "sent",
         "message": resp.json(),
         "channel": data.channel_slug,
@@ -429,6 +428,15 @@ async def _handle_message_send(
         "guidance": f"Message sent to task {data.task_id}'s session.",
     }
 
+    # Warn about failed mention resolution
+    if skipped_mentions:
+        result["warnings"] = [
+            f"Could not resolve mentions: {skipped_mentions}. "
+            "Check agent slug spelling (e.g., 'be-dev-1', 'be-pm')."
+        ]
+
+    return result
+
 
 async def _handle_message_get(client: ApiClient, message_id: str) -> dict[str, Any]:
     """Handle message retrieval."""
@@ -617,14 +625,17 @@ def create_message_mcp_server(agent_id: str) -> FastMCP:
         # Get task's primary session
         session_id = await _get_task_primary_session(client, task_id)
         if not session_id:
-            return {
-                "error": "NO_SESSION",
-                "message": f"Task {task_id} has no linked session.",
-                "guidance": (
-                    "This task doesn't have a work session yet. "
-                    "The PM should create one before work begins."
-                ),
-            }
+            return format_error_response(
+                "NO_SESSION_FOR_TASK",
+                f"Task {task_id} has no linked session.",
+                {
+                    "guidance": (
+                        "This task doesn't have a work session yet. "
+                        "The PM should create one before work begins."
+                    ),
+                    "task_id": task_id,
+                },
+            )
 
         # Get messages from the session
         resp = await client.get(

roboco/mcp/optimal_server.py

@@ -0,0 +1,321 @@
+"""
+Optimal MCP Server
+
+Exposes knowledge base, RAG, and semantic search tools to Claude Code agents.
+
+Tools:
+- roboco_kb_search: Semantic search across indexed content
+- roboco_rag_query: RAG query with answer generation
+- roboco_kb_index_code: Index code files (PM/Developer)
+- roboco_kb_index_docs: Index documentation (PM/Documenter)
+- roboco_kb_stats: Get index statistics
+- roboco_tokens_estimate: Estimate token count for content
+"""
+
+from typing import Any
+
+from fastapi import status as http_status
+from mcp.server.fastmcp import FastMCP
+
+from roboco.mcp.utils import ApiClient, format_error_response
+
+
+def _register_search_tools(mcp: FastMCP, client: ApiClient) -> None:
+    """Register search tools available to all agents."""
+
+    @mcp.tool()
+    async def roboco_kb_search(
+        query: str,
+        top_k: int = 5,
+        project: str | None = None,
+        task_id: str | None = None,
+        index_types: list[str] | None = None,
+    ) -> dict[str, Any]:
+        """
+        Semantic search across indexed knowledge base.
+
+        Use this to find relevant code, documentation, past decisions,
+        or learnings that might help with your current task.
+
+        Args:
+            query: Natural language search query
+            top_k: Number of results to return (1-20, default 5)
+            project: Optional project filter
+            task_id: Optional task filter
+            index_types: Index types to search (code, docs, decisions, learnings)
+
+        Returns:
+            Search results with relevance scores and source info
+        """
+        payload: dict[str, Any] = {
+            "query": query,
+            "top_k": min(max(top_k, 1), 20),
+        }
+        if project:
+            payload["project"] = project
+        if task_id:
+            payload["task_id"] = task_id
+        if index_types:
+            payload["index_types"] = index_types
+
+        resp = await client.post("/optimal/kb/search", json=payload)
+        if not resp.ok:
+            return format_error_response(
+                "SEARCH_FAILED",
+                "Failed to search knowledge base",
+                {"api_error": resp.text},
+            )
+
+        result = resp.json()
+        return {
+            "status": "success",
+            "query": query,
+            "total": result.get("total", 0),
+            "results": result.get("results", []),
+        }
+
+    @mcp.tool()
+    async def roboco_rag_query(
+        query: str,
+        top_k: int = 5,
+        project: str | None = None,
+        task_id: str | None = None,
+    ) -> dict[str, Any]:
+        """
+        RAG query - get an AI-generated answer using knowledge base context.
+
+        Use this when you need an answer synthesized from the knowledge base,
+        not just search results. Good for questions like:
+        - "How does authentication work in this codebase?"
+        - "What's the pattern for error handling?"
+        - "What decisions were made about the database schema?"
+
+        Args:
+            query: Natural language question
+            top_k: Number of context chunks to use (1-20, default 5)
+            project: Optional project filter
+            task_id: Optional task filter
+
+        Returns:
+            Generated answer with citations to sources
+        """
+        payload: dict[str, Any] = {
+            "query": query,
+            "top_k": min(max(top_k, 1), 20),
+        }
+        if project:
+            payload["project"] = project
+        if task_id:
+            payload["task_id"] = task_id
+
+        resp = await client.post("/optimal/rag/query", json=payload)
+        if not resp.ok:
+            return format_error_response(
+                "RAG_FAILED",
+                "Failed to query RAG",
+                {"api_error": resp.text},
+            )
+
+        result = resp.json()
+        return {
+            "status": "success",
+            "query": query,
+            "answer": result.get("answer", ""),
+            "citations": result.get("citations", []),
+            "context_used": result.get("context_used", 0),
+        }
+
+    @mcp.tool()
+    async def roboco_kb_stats() -> dict[str, Any]:
+        """
+        Get knowledge base statistics.
+
+        Shows what's indexed and available for search.
+
+        Returns:
+            Stats about indexed content by type
+        """
+        resp = await client.get("/optimal/stats")
+        if not resp.ok:
+            return format_error_response(
+                "STATS_FAILED",
+                "Failed to get KB stats",
+                {"api_error": resp.text},
+            )
+
+        return {
+            "status": "success",
+            **resp.json(),
+        }
+
+
+def _register_indexing_tools(mcp: FastMCP, client: ApiClient) -> None:
+    """Register indexing tools (permission-controlled at API level)."""
+
+    @mcp.tool()
+    async def roboco_kb_index_code(
+        sources: list[str],
+        project: str | None = None,
+    ) -> dict[str, Any]:
+        """
+        Index code files for semantic search.
+
+        PERMISSION: Requires INDEX_CODE permission (typically PM, Developer).
+
+        Args:
+            sources: List of file paths, directories, or globs (e.g., ["src/**/*.py"])
+            project: Optional project identifier for filtering
+
+        Returns:
+            Count of indexed files
+        """
+        if not sources:
+            return format_error_response(
+                "INVALID_INPUT",
+                "At least one source path required",
+            )
+
+        payload: dict[str, Any] = {"sources": sources}
+        if project:
+            payload["project"] = project
+
+        resp = await client.post("/optimal/kb/index/code", json=payload)
+        if not resp.ok:
+            if resp.status_code == http_status.HTTP_403_FORBIDDEN:
+                return format_error_response(
+                    "NOT_AUTHORIZED",
+                    "You don't have permission to index code",
+                )
+            return format_error_response(
+                "INDEX_FAILED",
+                "Failed to index code",
+                {"api_error": resp.text},
+            )
+
+        result = resp.json()
+        return {
+            "status": "indexed",
+            "indexed": result.get("indexed", 0),
+            "sources": sources,
+            "project": project,
+        }
+
+    @mcp.tool()
+    async def roboco_kb_index_docs(
+        sources: list[str],
+        project: str | None = None,
+    ) -> dict[str, Any]:
+        """
+        Index documentation for semantic search.
+
+        PERMISSION: Requires INDEX_DOCS permission (typically PM, Documenter).
+
+        Args:
+            sources: List of file paths, URLs, or globs (e.g., ["docs/**/*.md"])
+            project: Optional project identifier for filtering
+
+        Returns:
+            Count of indexed documents
+        """
+        if not sources:
+            return format_error_response(
+                "INVALID_INPUT",
+                "At least one source path required",
+            )
+
+        payload: dict[str, Any] = {"sources": sources}
+        if project:
+            payload["project"] = project
+
+        resp = await client.post("/optimal/kb/index/docs", json=payload)
+        if not resp.ok:
+            if resp.status_code == http_status.HTTP_403_FORBIDDEN:
+                return format_error_response(
+                    "NOT_AUTHORIZED",
+                    "You don't have permission to index documentation",
+                )
+            return format_error_response(
+                "INDEX_FAILED",
+                "Failed to index documentation",
+                {"api_error": resp.text},
+            )
+
+        result = resp.json()
+        return {
+            "status": "indexed",
+            "indexed": result.get("indexed", 0),
+            "sources": sources,
+            "project": project,
+        }
+
+
+def _register_utility_tools(mcp: FastMCP, client: ApiClient) -> None:
+    """Register utility tools."""
+
+    @mcp.tool()
+    async def roboco_tokens_estimate(
+        content: str,
+        model: str = "claude-sonnet-4-20250514",
+    ) -> dict[str, Any]:
+        """
+        Estimate token count for content.
+
+        Use this to check if content will fit within context limits.
+
+        Args:
+            content: Text content to estimate
+            model: Model to estimate for (default: claude-sonnet-4)
+
+        Returns:
+            Token count estimate
+        """
+        if not content:
+            return format_error_response(
+                "INVALID_INPUT",
+                "Content cannot be empty",
+            )
+
+        resp = await client.post(
+            "/optimal/tokens/estimate",
+            json={"content": content, "model": model},
+        )
+        if not resp.ok:
+            return format_error_response(
+                "ESTIMATE_FAILED",
+                "Failed to estimate tokens",
+                {"api_error": resp.text},
+            )
+
+        result = resp.json()
+        return {
+            "status": "success",
+            "token_count": result.get("token_count", 0),
+            "model": model,
+            "content_length": len(content),
+        }
+
+
+def create_optimal_mcp_server(agent_id: str) -> FastMCP:
+    """Create an Optimal MCP server for a specific agent."""
+    mcp = FastMCP(f"roboco-optimal-{agent_id}", json_response=True)
+    client = ApiClient(agent_id)
+
+    # Register all tool groups
+    _register_search_tools(mcp, client)
+    _register_indexing_tools(mcp, client)
+    _register_utility_tools(mcp, client)
+
+    return mcp
+
+
+if __name__ == "__main__":
+    import sys
+
+    _MIN_ARGS = 2
+    if len(sys.argv) < _MIN_ARGS:
+        print("Usage: python -m roboco.mcp.optimal_server <agent_id>")
+        sys.exit(1)
+
+    agent_id_cli = sys.argv[1]
+    server = create_optimal_mcp_server(agent_id_cli)
+    server.run()

roboco/mcp/task_server.py

@@ -4,29 +4,47 @@ Task MCP Server
 Exposes task management tools to Claude Code agents with built-in
 enforcement of task lifecycle rules.
 
-Tools:
+Tools (Core - all agents):
 - roboco_task_scan: List available tasks (paused, assigned, available)
 - roboco_task_get: Get task details
 - roboco_task_claim: Claim a task
 - roboco_task_plan: Submit implementation plan
 - roboco_task_start: Start working on task
 - roboco_task_progress: Update progress
+- roboco_task_escalate: Escalate task up hierarchy
+- roboco_task_substitute: Release task gracefully
+- roboco_task_submit_pm_review: Submit non-dev task directly to PM
+- roboco_agent_idle: Signal no work available (triggers shutdown)
+
+Tools (Blocking - Developer/PM):
 - roboco_task_block: Mark task as blocked
 - roboco_task_unblock: Unblock task
 - roboco_task_pause: Pause task
+
+Tools (Developer):
+- roboco_task_submit_verification: Self-verify before QA
 - roboco_task_submit_qa: Submit for QA review
-- roboco_task_qa_pass: Pass QA (QA role only)
-- roboco_task_qa_fail: Fail QA (QA role only)
-- roboco_task_docs_complete: Mark docs complete (Documenter only)
-- roboco_task_complete: Mark task complete (PM only, after docs)
-- roboco_task_create: Create new task (PM only)
-- roboco_task_assign: Assign task to agent (PM only)
-- roboco_task_cancel: Cancel a task (PM/Board only)
-- roboco_task_escalate: Escalate task up hierarchy (all agents)
-- roboco_session_create_for_tasks: Create work session for tasks (PM only)
-- roboco_session_link_task: Link session to task (PM only)
-- roboco_session_unlink_task: Unlink session from task (PM only)
+
+Tools (QA):
+- roboco_task_qa_pass: Pass QA
+- roboco_task_qa_fail: Fail QA with issues
+
+Tools (Documenter):
+- roboco_task_docs_complete: Mark documentation complete
+
+Tools (PM/Board):
+- roboco_task_create: Create new task
+- roboco_task_assign: Assign task to agent
+- roboco_task_activate: Move task from backlog to pending
+- roboco_task_complete: Mark task complete (after full workflow)
+- roboco_task_cancel: Cancel a task
+
+Tools (Sessions - PM/Board):
+- roboco_session_create_for_tasks: Create work session for tasks
+- roboco_session_link_task: Link session to task
+- roboco_session_unlink_task: Unlink session from task
 - roboco_session_get_for_task: Get sessions for a task (all agents)
+- roboco_group_create: Create agent groups (Main PM only)
 """
 
 from typing import Any
@@ -70,6 +88,7 @@ from roboco.mcp.tasks.handlers import (
     handle_task_start,
     handle_task_submit_qa,
     handle_task_submit_verification,
+    handle_task_substitute,
     handle_task_unblock,
 )
 from roboco.mcp.utils import ApiClient
@@ -251,6 +270,54 @@ def _register_core_tools(mcp: FastMCP, client: ApiClient, agent_id: str) -> None
         )
         return await handle_task_escalate(client, input_data, agent_id)
 
+    @mcp.tool()
+    async def roboco_task_substitute(
+        task_id: str,
+        reason: str,
+        details: str,
+        suggested_role: str | None = None,
+        suggested_team: str | None = None,
+    ) -> dict[str, Any]:
+        """
+        Request to be substituted out of a task.
+
+        Use this to gracefully release a task when you cannot or should not
+        continue working on it. This BYPASSES the normal "can't claim while
+        in_progress" rule - that's the whole point.
+
+        REASONS (SubstituteReason enum values):
+        - low_context: Insufficient context to continue safely
+        - out_of_scope_team: Task belongs to different team
+        - out_of_scope_role: Task requires different role (e.g., QA, not dev)
+        - task_complete: Finished work, releasing for next stage
+        - max_retries: Exceeded retry limit, need fresh perspective
+        - blocked_external: Need skills outside your capabilities
+
+        EFFECT:
+        - Task is released and reassigned (or moved to QA/docs/blocked)
+        - You are FREE to claim new work with roboco_task_scan()
+
+        Args:
+            task_id: Task UUID to release
+            reason: One of: low_context, out_of_scope_team, out_of_scope_role,
+                    task_complete, max_retries, blocked_external
+            details: Human-readable explanation
+            suggested_role: Hint for reassignment (developer, qa, pm, documenter)
+            suggested_team: Hint for reassignment (backend, frontend, ux_ui)
+
+        Returns:
+            Confirmation with next steps
+        """
+        return await handle_task_substitute(
+            client,
+            task_id,
+            agent_id,
+            reason,
+            details,
+            suggested_role=suggested_role,
+            suggested_team=suggested_team,
+        )
+
     @mcp.tool()
     async def roboco_task_submit_pm_review(
         task_id: str, notes: str | None = None

roboco/mcp/tasks/handlers/__init__.py

@@ -37,6 +37,7 @@ from roboco.mcp.tasks.handlers.sessions import (
     handle_session_link_task,
     handle_session_unlink_task,
 )
+from roboco.mcp.tasks.handlers.substitute import handle_task_substitute
 from roboco.mcp.tasks.handlers.work import (
     handle_task_plan,
     handle_task_progress,
@@ -70,5 +71,6 @@ __all__ = [
     "handle_task_start",
     "handle_task_submit_qa",
     "handle_task_submit_verification",
+    "handle_task_substitute",
     "handle_task_unblock",
 ]

roboco/mcp/tasks/handlers/_helpers.py

@@ -98,8 +98,13 @@ def get_scan_guidance(
 
 
 def check_blocking_tasks(active_tasks: list[dict]) -> dict[str, Any] | None:
-    """Check for blocking active tasks. Returns error or None."""
-    blocking_statuses = ["pending", "claimed", "in_progress", "verifying"]
+    """Check for blocking active tasks. Returns error or None.
+
+    NOTE: "pending" is NOT blocking. If PM assigned multiple pending tasks,
+    agent should be able to claim any of them. Only tasks being actively
+    worked on (claimed, in_progress, verifying) block new claims.
+    """
+    blocking_statuses = ["claimed", "in_progress", "verifying"]
     blocking = [t for t in active_tasks if t.get("status") in blocking_statuses]
     if blocking:
         status = blocking[0].get("status", "active")

roboco/mcp/tasks/handlers/lifecycle.py

@@ -89,6 +89,55 @@ def _is_pm_own_task(task: dict[str, Any], agent_id: str) -> bool:
     )
 
 
+async def _check_children_completed(
+    client: ApiClient, task_id: str
+) -> dict[str, Any] | None:
+    """Check ALL children of a task are completed.
+
+    Cancelled subtasks also block completion - they must be resolved first.
+    Returns error if any children are not completed, None if OK.
+    """
+    try:
+        # Fetch children/subtasks for this task
+        resp = await client.get(f"/tasks/{task_id}/subtasks")
+        if not resp.ok:
+            # If endpoint doesn't exist or fails, skip check (backwards compat)
+            return None
+
+        subtasks = resp.json()
+        if not subtasks:
+            return None  # No children, OK to complete
+
+        incomplete: list[dict[str, str]] = []
+        for subtask in subtasks:
+            subtask_status = subtask.get("status")
+            # ONLY "completed" is acceptable - cancelled/pending/etc. block completion
+            if subtask_status != "completed":
+                incomplete.append({
+                    "id": str(subtask.get("id", "unknown")),
+                    "title": subtask.get("title", "Untitled"),
+                    "status": subtask_status or "unknown",
+                })
+
+        if incomplete:
+            return format_error_response(
+                "INCOMPLETE_CHILDREN",
+                f"Cannot complete task: {len(incomplete)} subtask(s) not completed.",
+                {
+                    "incomplete_subtasks": incomplete,
+                    "guidance": (
+                        "ALL subtasks must be COMPLETED before completing parent. "
+                        "Cancelled subtasks must be resolved or removed first."
+                    ),
+                },
+            )
+
+        return None
+    except Exception:
+        # If check fails for any reason, allow completion (backwards compat)
+        return None
+
+
 async def handle_task_complete(
     client: ApiClient, task_id: str, agent_id: str
 ) -> dict[str, Any]:
@@ -120,6 +169,10 @@ async def handle_task_complete(
             {"current_status": current_status},
         )
 
+    # Check all children are completed before allowing parent completion
+    if error := await _check_children_completed(client, task_id):
+        return error
+
     complete_resp = await client.post(f"/tasks/{task_id}/complete")
     if not complete_resp.ok:
         return format_error_response(

roboco/mcp/tasks/handlers/management.py

@@ -24,6 +24,58 @@ from roboco.mcp.utils import ApiClient, format_error_response, resolve_agent_uui
 # HELPER FUNCTIONS
 # =============================================================================
 
+# Roles that cannot be assigned to cell-specific work
+BOARD_ROLES = frozenset({"product_owner", "head_marketing", "auditor", "ceo"})
+
+# Teams that represent cell work (not board/strategic)
+CELL_TEAMS = frozenset({"backend", "frontend", "ux_ui"})
+
+
+def validate_assignee_can_work_on_team(
+    assignee: str, task_team: str | None
+) -> dict[str, Any] | None:
+    """Validate assignee can work on the task's team.
+
+    Board members (product_owner, head_marketing, auditor) cannot be assigned
+    to cell-specific work (backend, frontend, ux_ui tasks).
+
+    Returns error dict or None if valid.
+    """
+    assignee_role = get_agent_role(assignee)
+    assignee_team = get_agent_team(assignee)
+
+    # Board members cannot work on cell tasks
+    if assignee_role in BOARD_ROLES and task_team in CELL_TEAMS:
+        return format_error_response(
+            "INVALID_ASSIGNEE",
+            f"Cannot assign {assignee_role} to {task_team} tasks. "
+            "Board members handle strategic work, not cell tasks.",
+            {
+                "assignee": assignee,
+                "assignee_role": assignee_role,
+                "task_team": task_team,
+                "guidance": "Assign to a cell member (e.g., be-dev-1, be-pm) instead.",
+            },
+        )
+
+    # Cell members should only work on their own team's tasks
+    if assignee_team and task_team and assignee_team != task_team:
+        # Main PM is an exception - can work across teams
+        if assignee_role == "main_pm":
+            return None
+        return format_error_response(
+            "TEAM_MISMATCH",
+            f"Cannot assign {assignee} ({assignee_team}) to {task_team} task.",
+            {
+                "assignee": assignee,
+                "assignee_team": assignee_team,
+                "task_team": task_team,
+                "guidance": f"Assign to a {task_team} team member instead.",
+            },
+        )
+
+    return None
+
 
 def validate_cell_pm_assignment(
     role: str,
@@ -32,10 +84,15 @@ def validate_cell_pm_assignment(
     assignee: str,
 ) -> dict[str, Any] | None:
     """Validate Cell PM assignment restrictions. Returns error dict or None."""
+    # First validate assignee can work on the team (applies to ALL roles)
+    task_team = task.get("team")
+    if error := validate_assignee_can_work_on_team(assignee, task_team):
+        return error
+
+    # Additional Cell PM restrictions
     if role != "cell_pm":
         return None
 
-    task_team = task.get("team")
     if task_team != agent_team:
         return format_error_response(
             "TEAM_MISMATCH",
@@ -43,14 +100,6 @@ def validate_cell_pm_assignment(
             {"task_team": task_team},
         )
 
-    assignee_team = get_agent_team(assignee)
-    if assignee_team and assignee_team != agent_team:
-        return format_error_response(
-            "ASSIGNEE_MISMATCH",
-            "Cannot assign to agent outside your team",
-            {"assignee_team": assignee_team, "your_team": agent_team},
-        )
-
     return None
 
 
@@ -191,6 +240,14 @@ async def handle_task_create(
     if error := _validate_cell_pm_team(agent_id, input_data.team):
         return error
 
+    # Validate assignee BEFORE creating task (avoid orphan tasks)
+    if input_data.assigned_to:
+        error = validate_assignee_can_work_on_team(
+            input_data.assigned_to, input_data.team
+        )
+        if error:
+            return error
+
     payload = _build_task_payload(input_data)
 
     try:

roboco/mcp/tasks/handlers/substitute.py

@@ -0,0 +1,203 @@
+"""
+Task Substitute Handler
+
+Handler for agent substitution requests.
+Allows agents to release tasks gracefully when they can't continue.
+Bypasses the "can't claim while in_progress" rule.
+"""
+
+from dataclasses import dataclass
+from typing import Any
+
+from roboco.mcp.tasks import format_task_response
+from roboco.mcp.tasks.handlers._helpers import (
+    fetch_task_or_error,
+    resolve_agent_uuid_cached,
+)
+from roboco.mcp.utils import ApiClient, format_error_response
+from roboco.models import SubstituteReason, TaskStatus
+
+# HTTP status code for "Not Found"
+HTTP_NOT_FOUND = 404
+
+# Map substitute reasons to target task statuses
+REASON_TO_STATUS: dict[SubstituteReason, TaskStatus] = {
+    SubstituteReason.TASK_COMPLETE: TaskStatus.AWAITING_QA,
+    SubstituteReason.LOW_CONTEXT: TaskStatus.PENDING,
+    SubstituteReason.OUT_OF_SCOPE_TEAM: TaskStatus.PENDING,
+    SubstituteReason.OUT_OF_SCOPE_ROLE: TaskStatus.PENDING,
+    SubstituteReason.MAX_RETRIES: TaskStatus.PENDING,
+    SubstituteReason.BLOCKED_EXTERNAL: TaskStatus.BLOCKED,
+}
+
+
+@dataclass
+class SubstituteRequest:
+    """Request data for substitution."""
+
+    task_id: str
+    agent_id: str
+    reason: SubstituteReason
+    details: str
+    suggested_role: str | None = None
+    suggested_team: str | None = None
+
+
+async def _validate_substitute_request(
+    task: dict[str, Any],
+    agent_id: str,
+    reason: str,
+    client: ApiClient,
+) -> dict[str, Any] | None:
+    """Validate substitution request. Returns error or None."""
+    # Validate reason
+    try:
+        SubstituteReason(reason)
+    except ValueError:
+        valid_reasons = [r.value for r in SubstituteReason]
+        return format_error_response(
+            "INVALID_REASON",
+            f"Invalid substitute reason: {reason}",
+            {"valid_reasons": valid_reasons},
+        )
+
+    # Check agent owns the task
+    assigned_to = task.get("assigned_to")
+    if assigned_to:
+        agent_uuid = await resolve_agent_uuid_cached(agent_id, client)
+        if agent_uuid and str(assigned_to) != agent_uuid:
+            return format_error_response(
+                "NOT_OWNER",
+                "You can only substitute out of tasks you own",
+                {"task_owner": str(assigned_to), "requester": agent_id},
+            )
+
+    return None
+
+
+async def _execute_substitute(
+    client: ApiClient, req: SubstituteRequest
+) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
+    """Execute substitution. Returns (result, None) or (None, error)."""
+    # Determine new status based on reason
+    new_status = REASON_TO_STATUS.get(req.reason, TaskStatus.PENDING)
+
+    # Call API to execute substitution
+    resp = await client.post(
+        f"/tasks/{req.task_id}/substitute",
+        json={
+            "agent_id": req.agent_id,
+            "reason": req.reason.value,
+            "details": req.details,
+            "new_status": new_status.value,
+            "suggested_role": req.suggested_role,
+            "suggested_team": req.suggested_team,
+        },
+    )
+
+    if not resp.ok:
+        # If endpoint doesn't exist yet, fall back to manual status update
+        if resp.status_code == HTTP_NOT_FOUND:
+            # Fallback: just update status and clear assignment
+            update_resp = await client.put(
+                f"/tasks/{req.task_id}",
+                json={
+                    "status": new_status.value,
+                    "assigned_to": None,  # Clear assignment
+                    "dev_notes": (
+                        f"[SUBSTITUTE] Reason: {req.reason.value}\n{req.details}"
+                    ),
+                },
+            )
+            if not update_resp.ok:
+                return None, format_error_response(
+                    "SUBSTITUTE_FAILED",
+                    "Failed to execute substitution",
+                    {"api_error": update_resp.text},
+                )
+            result: dict[str, Any] = update_resp.json()
+            return result, None
+
+        return None, format_error_response(
+            "SUBSTITUTE_FAILED",
+            "Failed to execute substitution",
+            {"api_error": resp.text},
+        )
+
+    result = resp.json()
+    return result, None
+
+
+async def handle_task_substitute(
+    client: ApiClient,
+    task_id: str,
+    agent_id: str,
+    reason: str,
+    details: str,
+    **kwargs: str | None,
+) -> dict[str, Any]:
+    """Handle task substitution request.
+
+    Allows agents to release tasks gracefully when they can't continue.
+    This bypasses the normal "can't claim while in_progress" rule.
+
+    Args:
+        client: API client
+        task_id: Task to release
+        agent_id: Agent requesting substitution
+        reason: Substitution reason (SubstituteReason enum value)
+        details: Human-readable explanation
+        **kwargs: Optional suggested_role and suggested_team hints
+
+    Returns:
+        Response dict with next steps
+    """
+    # Fetch task
+    task, error = await fetch_task_or_error(client, task_id)
+    if error:
+        return error
+    assert task is not None
+
+    # Validate request
+    if error := await _validate_substitute_request(task, agent_id, reason, client):
+        return error
+
+    # Parse reason and build request
+    substitute_reason = SubstituteReason(reason)
+    req = SubstituteRequest(
+        task_id=task_id,
+        agent_id=agent_id,
+        reason=substitute_reason,
+        details=details,
+        suggested_role=kwargs.get("suggested_role"),
+        suggested_team=kwargs.get("suggested_team"),
+    )
+
+    # Execute substitution
+    result, error = await _execute_substitute(client, req)
+    if error:
+        return error
+    assert result is not None
+
+    # Determine next action message
+    if substitute_reason == SubstituteReason.TASK_COMPLETE:
+        next_action = (
+            "Task released for QA review. "
+            "You are now free to claim new work with roboco_task_scan()."
+        )
+    elif substitute_reason == SubstituteReason.BLOCKED_EXTERNAL:
+        next_action = (
+            "Task marked as blocked. PM will be notified. "
+            "You are now free to claim new work with roboco_task_scan()."
+        )
+    else:
+        next_action = (
+            "Task released and will be reassigned. "
+            "You are now free to claim new work with roboco_task_scan()."
+        )
+
+    return format_task_response(
+        result,
+        "RELEASED",
+        f"Substitution successful ({substitute_reason.value}). {next_action}",
+    )

roboco/mcp/utils.py

@@ -66,23 +66,42 @@ def format_error_response(
     """
     Format a standardized error response for MCP tools.
 
+    Uses the common error_response format for consistency with API layer.
+
     Args:
         code: Error code (e.g., "NOT_FOUND", "API_ERROR", "PERMISSION_DENIED")
         message: Human-readable error message
         details: Optional additional error details
 
     Returns:
-        Standardized error response dict
+        Standardized error response dict with status="error"
     """
-    response: dict[str, Any] = {
-        "error": {
-            "code": code,
-            "message": message,
-        }
-    }
-    if details:
-        response["error"]["details"] = details
-    return response
+    from roboco.api.schemas.common import error_response
+
+    return error_response(code, message, details)
+
+
+def format_success_response(
+    data: Any,
+    guidance: str | None = None,
+    next_step: str | None = None,
+) -> dict[str, Any]:
+    """
+    Format a standardized success response for MCP tools.
+
+    Uses the common success_response format for consistency with API layer.
+
+    Args:
+        data: Response payload
+        guidance: Actionable next step guidance
+        next_step: Workflow hint (e.g., PLAN, EXECUTE)
+
+    Returns:
+        Standardized success response dict with status="success"
+    """
+    from roboco.api.schemas.common import success_response
+
+    return success_response(data, guidance, next_step)
 
 
 async def resolve_agent_uuid(