From 01ff44b83f983b9e33998b37aa31bdc24e08c001 Mon Sep 17 00:00:00 2001 From: Renn F Date: Fri, 8 May 2026 07:45:18 +0200 Subject: [PATCH] fix(gateway): unblock PM planning + drop magic delegate task_type MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 64 ++++++++++++ roboco/api/schemas/v2/flow.py | 6 +- .../services/gateway/choreographer/_impl.py | 33 +++++-- tests/unit/api/routes/v2/test_flow_cell_pm.py | 3 + tests/unit/api/routes/v2/test_flow_main_pm.py | 4 + tests/unit/api/test_middleware.py | 71 ++++++++++++++ tests/unit/api/test_schemas_v2_flow.py | 42 ++++++++ .../test_choreographer_claim_guards.py | 18 +++- .../test_choreographer_delegate_guards.py | 1 + .../test_choreographer_impl_branches.py | 97 +++++++++++++++++++ .../gateway/test_choreographer_pm_extras.py | 28 +++++- 11 files changed, 350 insertions(+), 17 deletions(-) create mode 100644 tests/unit/api/test_schemas_v2_flow.py diff --git a/roboco/api/middleware.py b/roboco/api/middleware.py index 017a650f..7f7fa2c6 100644 --- a/roboco/api/middleware.py +++ b/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) diff --git a/roboco/api/schemas/v2/flow.py b/roboco/api/schemas/v2/flow.py index bd8ab91a..3b6de192 100644 --- a/roboco/api/schemas/v2/flow.py +++ b/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" diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index 935ef151..51ad7efb 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/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 - # non-code tasks. role_typed_claim_guard is skipped here because - # i_will_plan only services PM roles, which fall into the PM-code - # branch. ALREADY_ACTIVE/PAUSED still apply. + # Gate Set A: ALREADY_ACTIVE / PAUSED guards only. + # + # `pm_cannot_execute_code_guard` is INTENTIONALLY skipped here: + # 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( diff --git a/tests/unit/api/routes/v2/test_flow_cell_pm.py b/tests/unit/api/routes/v2/test_flow_cell_pm.py index eb894c41..3327ee44 100644 --- a/tests/unit/api/routes/v2/test_flow_cell_pm.py +++ b/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 diff --git a/tests/unit/api/routes/v2/test_flow_main_pm.py b/tests/unit/api/routes/v2/test_flow_main_pm.py index 18dcf5bc..1351ba5e 100644 --- a/tests/unit/api/routes/v2/test_flow_main_pm.py +++ b/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 diff --git a/tests/unit/api/test_middleware.py b/tests/unit/api/test_middleware.py index b908abfa..059b67f1 100644 --- a/tests/unit/api/test_middleware.py +++ b/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") diff --git a/tests/unit/api/test_schemas_v2_flow.py b/tests/unit/api/test_schemas_v2_flow.py new file mode 100644 index 00000000..3edb655e --- /dev/null +++ b/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" diff --git a/tests/unit/gateway/test_choreographer_claim_guards.py b/tests/unit/gateway/test_choreographer_claim_guards.py index 031dae6e..4fa82e87 100644 --- a/tests/unit/gateway/test_choreographer_claim_guards.py +++ b/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" - assert "code" in body["message"].lower() or "execute" in body["message"].lower() + # 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 diff --git a/tests/unit/gateway/test_choreographer_delegate_guards.py b/tests/unit/gateway/test_choreographer_delegate_guards.py index a98afb39..79b74d34 100644 --- a/tests/unit/gateway/test_choreographer_delegate_guards.py +++ b/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", ) diff --git a/tests/unit/gateway/test_choreographer_impl_branches.py b/tests/unit/gateway/test_choreographer_impl_branches.py index 7f64bf04..0bdb5c52 100644 --- a/tests/unit/gateway/test_choreographer_impl_branches.py +++ b/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() diff --git a/tests/unit/gateway/test_choreographer_pm_extras.py b/tests/unit/gateway/test_choreographer_pm_extras.py index c328b75b..f467d7e0 100644 --- a/tests/unit/gateway/test_choreographer_pm_extras.py +++ b/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"