fix(gateway): unblock PM planning + drop magic delegate task_type

Two coupled fixes from the 2026-05-08 smoke-test trace:

1. pm_cannot_execute_code is now scoped to i_will_work_on (the
   EXECUTION verb) only. Pre-fix it also fired on i_will_plan, which
   deadlocked any code-typed parent: cell_pm couldn't plan, so couldn't
   transition parent to in_progress, so couldn't delegate. PMs PLAN
   code-typed parents and DELEGATE the work — that's exactly the verb
   we were blocking.

2. delegate.task_type is now REQUIRED at both the HTTP boundary
   (DelegateRequest) and the choreographer dataclass (DelegateInputs).
   The pre-fix default of 'code' silently changed semantics whenever a
   caller forgot the field — main-pm's call in the smoke trace omitted
   it, schema defaulted to 'code', and the cell PM downstream was
   wedged. Also drops the choreographer's task.task_type fallback
   (the DB column is NOT NULL anyway).

Plus middleware coverage tests for the parallel ServiceError →
4xx handler hierarchy added in the prior session, restoring 100%
coverage across the touched files.

Tests: 3101 passing, 100% coverage, ruff clean.

roboco/api/middleware.py

@@ -25,6 +25,22 @@ from roboco.exceptions import (
     RobocoError,
     ValidationError,
 )
+from roboco.services.base import (
+    ConflictError as ServiceConflictError,
+)
+from roboco.services.base import (
+    NotFoundError as ServiceNotFoundError,
+)
+from roboco.services.base import (
+    ServiceError,
+    ServiceUnavailableError,
+)
+from roboco.services.base import (
+    UnauthorizedError as ServiceUnauthorizedError,
+)
+from roboco.services.base import (
+    ValidationError as ServiceValidationError,
+)
 
 logger = structlog.get_logger()
 
@@ -164,6 +180,53 @@ async def roboco_exception_handler(request: Request, exc: Exception) -> JSONResp
     )
 
 
+# `roboco.services.base.ServiceError` is a parallel exception hierarchy that
+# does NOT inherit from `RobocoError` (it extends `Exception` directly), so
+# `roboco_exception_handler` never sees it and the requests fall through to
+# `generic_exception_handler` as 500s. Map its subclasses to the same status
+# codes used in the RobocoError handler so route-layer try/except blocks can
+# surface clean 4xx codes whether the service raises from `roboco.exceptions`
+# or `roboco.services.base`.
+_SERVICE_ERROR_STATUS: dict[type[ServiceError], int] = {
+    ServiceNotFoundError: 404,
+    ServiceValidationError: 422,
+    ServiceConflictError: 409,
+    ServiceUnauthorizedError: 403,
+    ServiceUnavailableError: 503,
+}
+
+
+async def service_exception_handler(request: Request, exc: Exception) -> JSONResponse:
+    """Handle `roboco.services.base.ServiceError` and subclasses."""
+    svc_exc = cast("ServiceError", exc)
+    status_code = 500
+    for exc_type, mapped_status in _SERVICE_ERROR_STATUS.items():
+        if isinstance(svc_exc, exc_type):
+            status_code = mapped_status
+            break
+
+    correlation_id = getattr(request.state, "correlation_id", None)
+    details = dict(svc_exc.details)
+    if correlation_id:
+        details["correlation_id"] = correlation_id
+
+    logger.warning(
+        "Handled exception",
+        error_type=type(svc_exc).__name__,
+        error_message=svc_exc.message,
+        status_code=status_code,
+    )
+
+    return JSONResponse(
+        status_code=status_code,
+        content={
+            "error": type(svc_exc).__name__,
+            "message": svc_exc.message,
+            "details": details,
+        },
+    )
+
+
 async def generic_exception_handler(request: Request, exc: Exception) -> JSONResponse:
     """Handle unexpected exceptions."""
     correlation_id = getattr(request.state, "correlation_id", None)
@@ -281,6 +344,7 @@ def setup_middleware(app: FastAPI) -> None:
     app.add_exception_handler(RequestValidationError, request_validation_handler)
     app.add_exception_handler(HTTPException, http_exception_handler)
     app.add_exception_handler(RobocoError, roboco_exception_handler)
+    app.add_exception_handler(ServiceError, service_exception_handler)
     app.add_exception_handler(Exception, generic_exception_handler)
 
     # Middleware (added in reverse order due to LIFO)

roboco/api/schemas/v2/flow.py

@@ -99,7 +99,11 @@ class DelegateRequest(BaseModel):
     description: str = Field(..., min_length=1)
     assigned_to: str = Field(..., min_length=1)
     team: str = Field(..., min_length=1)
-    task_type: str = "code"
+    # task_type is REQUIRED. The 2026-05-08 trace showed agents omitting
+    # it and the old default of 'code' deadlocking the lifecycle. Force
+    # callers to declare intent: code | documentation | research |
+    # planning | design | administrative.
+    task_type: str = Field(..., min_length=1)
     acceptance_criteria: list[str] | None = None
     estimated_complexity: str = "medium"
 

roboco/services/gateway/choreographer/_impl.py

@@ -66,13 +66,19 @@ class ChoreographerDeps:
 
 @dataclass(frozen=True)
 class DelegateInputs:
-    """Bundle of fields the ``delegate`` verb receives from the route layer."""
+    """Bundle of fields the ``delegate`` verb receives from the route layer.
+
+    `task_type` has no default — the v2 schema enforces this at the HTTP
+    boundary, but defaulting here too would let direct callers (tests,
+    other internal code) silently pick 'code' and recreate the
+    2026-05-08 deadlock.
+    """
 
     title: str
     description: str
     assigned_to: str
     team: str
-    task_type: str = "code"
+    task_type: str
     acceptance_criteria: list[str] | None = None
     estimated_complexity: str = "medium"
 
@@ -279,7 +285,7 @@ class Choreographer:
         """
         agent = await self.task.agent_for(agent_id)
         role = agent.role if agent is not None else "developer"
-        task_type = str(getattr(task, "task_type", "code") or "code")
+        task_type = str(task.task_type)
 
         if guard := self._run_role_guards(
             role,
@@ -1084,12 +1090,23 @@ class Choreographer:
                 ),
                 context_briefing=await self._briefing_for(pm_agent_id, task_id),
             )
-        # Gate Set A: PM_CANNOT_EXECUTE_CODE — cell_pm/main_pm can only plan
+        # Gate Set A: ALREADY_ACTIVE / PAUSED guards only.
-        # non-code tasks. role_typed_claim_guard is skipped here because
+        #
-        # i_will_plan only services PM roles, which fall into the PM-code
+        # `pm_cannot_execute_code_guard` is INTENTIONALLY skipped here:
-        # branch. ALREADY_ACTIVE/PAUSED still apply.
+        # PMs PLAN code-typed parent tasks all the time (they decompose
+        # the work into developer-claimable subtasks via delegate).
+        # The "PMs cannot execute code" rule belongs to the EXECUTION
+        # verb (`i_will_work_on`), not the PLANNING verb (`i_will_plan`).
+        # Pre-fix this guard fired on i_will_plan and deadlocked any
+        # code-typed parent task — see the 2026-05-08 smoke-test trace.
+        #
+        # `role_typed_claim_guard` is also skipped because i_will_plan
+        # services only PM roles which aren't in its allow-table.
         guard = await self._run_claim_guards(
-            agent_id=pm_agent_id, task=t, skip_role_typed=True
+            agent_id=pm_agent_id,
+            task=t,
+            skip_role_typed=True,
+            skip_pm_code=True,
         )
         if guard:
             return self._with_briefing(

tests/unit/api/routes/v2/test_flow_cell_pm.py

@@ -258,12 +258,15 @@ async def test_delegate_dispatches_inputs_bundle() -> None:
             "description": "Add the foo endpoint with tests.",
             "assigned_to": "be-dev-1",
             "team": "backend",
+            "task_type": "code",
         },
         headers=_HEADERS,
     )
 
     assert resp.status_code == _HTTP_200
     mock_chore.delegate.assert_awaited_once()
+    inputs = mock_chore.delegate.await_args.args[2]
+    assert inputs.task_type == "code"
 
 
 @pytest.mark.asyncio

tests/unit/api/routes/v2/test_flow_main_pm.py

@@ -229,11 +229,15 @@ async def test_delegate_to_cell_pm_dispatches_inputs_bundle() -> None:
             "description": "Plan + drive backend work for feature X.",
             "assigned_to": "be-pm",
             "team": "backend",
+            "task_type": "planning",
         },
         headers=_HEADERS,
     )
 
     assert resp.status_code == _HTTP_200
+    mock_chore.delegate.assert_awaited_once()
+    inputs = mock_chore.delegate.await_args.args[2]
+    assert inputs.task_type == "planning"
 
 
 @pytest.mark.asyncio

tests/unit/api/test_middleware.py

@@ -19,6 +19,18 @@ from roboco.exceptions import (
     RobocoError,
     ValidationError,
 )
+from roboco.services.base import (
+    ConflictError as ServiceConflictError,
+)
+from roboco.services.base import (
+    NotFoundError as ServiceNotFoundError,
+)
+from roboco.services.base import (
+    UnauthorizedError as ServiceUnauthorizedError,
+)
+from roboco.services.base import (
+    ValidationError as ServiceValidationError,
+)
 
 # ---------------------------------------------------------------------------
 # get_status_code
@@ -76,6 +88,23 @@ def _make_app() -> FastAPI:
     async def _he():
         raise HTTPException(status_code=403, detail="nope")
 
+    # service-layer errors (parallel hierarchy from roboco.services.base)
+    @app.get("/svc-notfound")
+    async def _svc_nf():
+        raise ServiceNotFoundError("Channel", "main-pm")
+
+    @app.get("/svc-validation")
+    async def _svc_v():
+        raise ServiceValidationError("invalid input", field="title")
+
+    @app.get("/svc-conflict")
+    async def _svc_c():
+        raise ServiceConflictError("duplicate", resource_type="task")
+
+    @app.get("/svc-unauth")
+    async def _svc_u():
+        raise ServiceUnauthorizedError("merge_pr", reason="not your PR")
+
     setup_middleware(app)
     return app
 
@@ -114,6 +143,48 @@ def test_http_exception_handler_returns_standardized_format() -> None:
     assert "error" in body
 
 
+# `roboco.services.base.ServiceError` is a parallel exception hierarchy
+# (it does NOT inherit from RobocoError), so a separate handler maps it
+# to clean 4xx codes instead of letting the generic 500 handler eat it.
+
+
+def test_service_notfound_translates_to_404() -> None:
+    client = TestClient(_make_app(), raise_server_exceptions=False)
+    response = client.get("/svc-notfound")
+    assert response.status_code == HTTPStatus.NOT_FOUND
+    body = response.json()
+    assert body["error"] == "NotFoundError"
+    assert "main-pm" in body["message"]
+
+
+def test_service_validation_translates_to_422() -> None:
+    client = TestClient(_make_app(), raise_server_exceptions=False)
+    response = client.get("/svc-validation")
+    assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
+    body = response.json()
+    assert body["error"] == "ValidationError"
+
+
+def test_service_conflict_translates_to_409() -> None:
+    client = TestClient(_make_app(), raise_server_exceptions=False)
+    response = client.get("/svc-conflict")
+    assert response.status_code == HTTPStatus.CONFLICT
+
+
+def test_service_unauthorized_translates_to_403() -> None:
+    client = TestClient(_make_app(), raise_server_exceptions=False)
+    response = client.get("/svc-unauth")
+    assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+def test_service_handler_carries_correlation_id() -> None:
+    client = TestClient(_make_app(), raise_server_exceptions=False)
+    cid = "test-svc-correlation-987"
+    response = client.get("/svc-notfound", headers={"X-Correlation-ID": cid})
+    body = response.json()
+    assert body["details"]["correlation_id"] == cid
+
+
 def test_generic_exception_returns_500() -> None:
     client = TestClient(_make_app(), raise_server_exceptions=False)
     response = client.get("/raise")

tests/unit/api/test_schemas_v2_flow.py

@@ -0,0 +1,42 @@
+"""Schema-level tests for v2 flow request bodies."""
+
+from __future__ import annotations
+
+from uuid import uuid4
+
+import pytest
+from pydantic import ValidationError
+from roboco.api.schemas.v2.flow import DelegateRequest
+
+
+def test_delegate_request_requires_task_type() -> None:
+    """task_type must be supplied explicitly — no magic default.
+
+    Background: the 2026-05-08 smoke-test trace showed main-pm calling
+    delegate without task_type, the schema defaulted to 'code', the
+    cell PM downstream couldn't plan a code-typed parent (pre-fix), and
+    the run deadlocked. Make the field required so misuse fails at the
+    HTTP boundary with a clear 422.
+    """
+    with pytest.raises(ValidationError) as exc:
+        DelegateRequest(
+            parent_task_id=uuid4(),
+            title="t",
+            description="d",
+            assigned_to="be-dev-1",
+            team="backend",
+            # task_type intentionally omitted
+        )
+    assert "task_type" in str(exc.value)
+
+
+def test_delegate_request_accepts_explicit_task_type() -> None:
+    req = DelegateRequest(
+        parent_task_id=uuid4(),
+        title="t",
+        description="d",
+        assigned_to="be-dev-1",
+        team="backend",
+        task_type="code",
+    )
+    assert req.task_type == "code"

tests/unit/gateway/test_choreographer_claim_guards.py

@@ -336,7 +336,15 @@ async def test_main_pm_cannot_claim_code_task_via_i_will_work_on() -> None:
 
 
 @pytest.mark.asyncio
-async def test_cell_pm_cannot_claim_code_task_via_i_will_plan() -> None:
+async def test_cell_pm_can_plan_code_typed_parent_via_i_will_plan() -> None:
+    """Rule change (2026-05-08): the PM-cannot-execute-code guard belongs
+    on `i_will_work_on` (the EXECUTION verb), not on `i_will_plan` (the
+    PLANNING verb). PMs decompose code-typed parent tasks into
+    developer-claimable subtasks all the time; that's planning, not
+    executing. Pre-fix this rejection deadlocked every code-typed parent
+    in the smoke test (see PRE_GATEWAY_LIFECYCLE.md and the 2026-05-08
+    audit-log analysis).
+    """
     pm_id = uuid4()
     task_id = uuid4()
     target = MagicMock(
@@ -353,10 +361,12 @@ async def test_cell_pm_cannot_claim_code_task_via_i_will_plan() -> None:
     deps = _make_deps(task=task_svc)
     c = Choreographer(deps)
 
-    env = await c.i_will_plan(pm_id, task_id, plan="x")
+    env = await c.i_will_plan(pm_id, task_id, plan="Decompose into 2 dev subtasks.")
     body = env.as_dict()
-    assert body["error"] == "not_authorized"
+    # The PM-cannot-execute-code rejection must NOT fire on i_will_plan.
-    assert "code" in body["message"].lower() or "execute" in body["message"].lower()
+    assert body.get("error") != "not_authorized", (
+        f"i_will_plan was rejected for a code-typed parent; envelope: {body}"
+    )
 
 
 @pytest.mark.asyncio

tests/unit/gateway/test_choreographer_delegate_guards.py

@@ -53,6 +53,7 @@ def _delegate_inputs() -> DelegateInputs:
         description="Add /v1/foo endpoint with tests",
         assigned_to="be-dev-1",
         team="backend",
+        task_type="code",
     )
 
 

tests/unit/gateway/test_choreographer_impl_branches.py

@@ -15,6 +15,7 @@ import pytest
 import structlog
 from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
 from roboco.services.gateway.choreographer._impl import DelegateInputs
+from roboco.services.gateway.claim_guards import pm_cannot_execute_code_guard
 from roboco.services.gateway.envelope import Envelope
 
 
@@ -250,6 +251,99 @@ async def test_i_will_work_on_in_progress_assigned_to_self_idempotent() -> None:
 # ---------------------------------------------------------------------------
 
 
+def test_pm_cannot_execute_code_guard_passes_for_non_code_task() -> None:
+    """Direct guard unit test: PM + non-code task → no rejection.
+    Covers claim_guards.py:98 (the early-return for non-code task_type).
+    """
+    assert pm_cannot_execute_code_guard("cell_pm", "planning") is None
+    assert pm_cannot_execute_code_guard("main_pm", "documentation") is None
+
+
+@pytest.mark.asyncio
+async def test_i_will_plan_pm_with_already_active_task_rejects() -> None:
+    """The already_active_guard still fires on i_will_plan even though
+    pm_cannot_execute_code is skipped. Covers _impl.py:1106-1108
+    (with-briefing wrap of the guard rejection).
+    """
+    pm_id = uuid4()
+    task_id = uuid4()
+    other_task_id = uuid4()
+    target = MagicMock(
+        status="pending",
+        assigned_to=pm_id,
+        plan=None,
+        id=task_id,
+        title="t",
+        team="backend",
+        parent_task_id=None,
+        task_type="planning",
+    )
+    busy_task = MagicMock(id=other_task_id, status="in_progress")
+    task_svc = AsyncMock()
+    task_svc.get.return_value = target
+    task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend")
+    task_svc.list_in_progress_for_agent.return_value = [busy_task]
+    task_svc.list_paused_for_agent.return_value = []
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+    env = await c.i_will_plan(pm_id, task_id, plan="x" * 30)
+    body = env.as_dict()
+    assert body["error"] == "invalid_state"
+    assert "in_progress task" in body["message"]
+
+
+@pytest.mark.asyncio
+async def test_i_will_plan_cell_pm_on_code_typed_parent_succeeds() -> None:
+    """Regression for the smoke-test deadlock (2026-05-08 trace).
+
+    When a cell PM tries to plan a code-typed parent task, the verb must
+    succeed — PMs PLAN code work and DELEGATE the execution; they don't
+    execute. The pre-fix `pm_cannot_execute_code_guard` was wrongly fired
+    on `i_will_plan` (the planning verb) instead of being scoped to
+    `i_will_work_on` (the execution verb), causing a deadlock: cell PM
+    couldn't plan → couldn't transition parent to in_progress → couldn't
+    delegate (delegate requires parent in_progress).
+    """
+    pm_id = uuid4()
+    task_id = uuid4()
+    task = MagicMock(
+        status="pending",
+        assigned_to=pm_id,
+        plan=None,
+        id=task_id,
+        title="Backend slice: Git workflow smoke test",
+        team="backend",
+        parent_task_id=uuid4(),  # subtask of the main_pm root
+        task_type="code",  # ← the trigger; pre-fix this rejected with
+        #                    "Cell Pm cannot claim code tasks"
+    )
+    task_svc = AsyncMock()
+    task_svc.get.return_value = task
+    task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend")
+    task_svc.list_in_progress_for_agent.return_value = []
+    task_svc.list_paused_for_agent.return_value = []
+    task_svc.get_subtasks.return_value = []
+    # Claim + start succeed so we can verify the verb runs end-to-end.
+    started_task = MagicMock(
+        status="in_progress",
+        assigned_to=pm_id,
+        id=task_id,
+        title=task.title,
+        team="backend",
+        task_type="code",
+    )
+    task_svc.claim.return_value = task
+    task_svc.start.return_value = started_task
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+    env = await c.i_will_plan(pm_id, task_id, plan="Decompose into 2 dev subtasks.")
+    body = env.as_dict()
+    # The PM-cannot-execute-code rejection must NOT fire on i_will_plan.
+    assert body.get("error") != "not_authorized", (
+        f"i_will_plan was rejected for a code-typed parent; envelope: {body}"
+    )
+
+
 @pytest.mark.asyncio
 async def test_i_will_plan_pending_claim_fails() -> None:
     pm_id = uuid4()
@@ -299,6 +393,7 @@ async def test_delegate_parent_not_found() -> None:
             description="y",
             assigned_to="be-dev-1",
             team="backend",
+            task_type="code",
         ),
     )
     body = env.as_dict()
@@ -333,6 +428,7 @@ async def test_delegate_unknown_role_rejected() -> None:
             description="y",
             assigned_to="be-dev-1",
             team="backend",
+            task_type="code",
         ),
     )
     body = env.as_dict()
@@ -368,6 +464,7 @@ async def test_delegate_parent_no_project_rejected() -> None:
             description="y",
             assigned_to="be-dev-1",
             team="backend",
+            task_type="code",
         ),
     )
     body = env.as_dict()

tests/unit/gateway/test_choreographer_pm_extras.py

@@ -379,6 +379,7 @@ async def test_delegate_main_pm_to_cell_pm_creates_subtask() -> None:
             description="Plan backend work for feature X",
             assigned_to="be-pm",
             team="backend",
+            task_type="planning",
         ),
     )
     assert env.error is None
@@ -417,6 +418,7 @@ async def test_delegate_cell_pm_to_team_dev_creates_subtask() -> None:
             description="Add /v1/foo endpoint with tests",
             assigned_to="be-dev-1",
             team="backend",
+            task_type="code",
         ),
     )
     assert env.error is None
@@ -438,7 +440,11 @@ async def test_delegate_main_pm_to_dev_is_rejected() -> None:
         main_pm_id,
         parent_id,
         DelegateInputs(
-            title="x", description="y", assigned_to="be-dev-1", team="backend"
+            title="x",
+            description="y",
+            assigned_to="be-dev-1",
+            team="backend",
+            task_type="code",
         ),
     )
     body = env.as_dict()
@@ -460,7 +466,13 @@ async def test_delegate_cell_pm_to_other_pm_rejected() -> None:
     env = await c.delegate(
         cell_pm_id,
         parent_id,
-        DelegateInputs(title="x", description="y", assigned_to="be-pm", team="backend"),
+        DelegateInputs(
+            title="x",
+            description="y",
+            assigned_to="be-pm",
+            team="backend",
+            task_type="code",
+        ),
     )
     body = env.as_dict()
     assert body["error"] == "not_authorized"
@@ -481,7 +493,11 @@ async def test_delegate_unknown_assignee_returns_invalid_state() -> None:
         pm_id,
         parent_id,
         DelegateInputs(
-            title="x", description="y", assigned_to="nope-pm", team="backend"
+            title="x",
+            description="y",
+            assigned_to="nope-pm",
+            team="backend",
+            task_type="code",
         ),
     )
     body = env.as_dict()
@@ -503,7 +519,11 @@ async def test_delegate_invalid_team_enum_rejected() -> None:
         pm_id,
         parent_id,
         DelegateInputs(
-            title="x", description="y", assigned_to="be-dev-1", team="not-a-team"
+            title="x",
+            description="y",
+            assigned_to="be-dev-1",
+            team="not-a-team",
+            task_type="code",
         ),
     )
     assert env.as_dict()["error"] == "invalid_state"