From cfb7424c80f7595e0c5bfa25f9255fe29dbced1a Mon Sep 17 00:00:00 2001 From: Renn F Date: Tue, 12 May 2026 02:45:48 +0200 Subject: [PATCH] =?UTF-8?q?fix(gateway):=20A1=20review-fixes=20=E2=80=94?= =?UTF-8?q?=20re-entry=20ordering,=20gate=20unit-coverage,=20approach=20ch?= =?UTF-8?q?eck?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three fixes from the code-quality review of a1009c0: 1. Critical: _pm_sub_tasks_gate ran before _handle_pm_reentry, breaking idempotent re-entry for PMs whose containers crashed mid-run. Moved the gate to after the re-entry short-circuit so initial-claim is the only path that hits the gate. 2. Critical: gate had no direct unit test (the HTTP-layer test mocked the choreographer). Added tests/unit/gateway/test_i_will_plan_sub_tasks_gate.py with six tests: empty sub_tasks → incomplete_input, missing rich_plan → incomplete_input, filled sub_tasks → gate passes, developer with empty sub_tasks → gate passes (devs don't decompose), sub_tasks filled but approach empty → incomplete_input, in_progress re-entry short- circuits before gate even with no sub_tasks. 3. Important: approach was only enforced at the HTTP Pydantic boundary. Direct service-layer callers (MCP, test fixtures, orchestrator- internal) could persist a plan with no approach. Gate now also checks approach >= 20 chars (_PM_APPROACH_MIN_LEN constant) and includes it in the rejection's missing list when absent. Plus: stale docstring on i_will_plan corrected; _handle_pm_reentry docstring rewritten to lead with the domain reason (re-entry contracts) not the PLR0911 linter justification; unused pytest import removed from test_i_will_plan_rich_required.py; pre-existing PLR2004/PLC0415 issues in that file fixed. --- .../services/gateway/choreographer/_impl.py | 116 +++--- .../v2/test_i_will_plan_rich_required.py | 15 +- .../test_i_will_plan_sub_tasks_gate.py | 382 ++++++++++++++++++ 3 files changed, 459 insertions(+), 54 deletions(-) create mode 100644 tests/unit/gateway/test_i_will_plan_sub_tasks_gate.py diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index e79b24a8..4f4e3f8c 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/roboco/services/gateway/choreographer/_impl.py @@ -45,6 +45,10 @@ from roboco.services.gateway.remediation import ( logger = structlog.get_logger() +# Minimum character length enforced on rich_plan["approach"] by the PM sub-tasks +# gate. Must match the Pydantic min_length on IWillPlanRequest.approach. +_PM_APPROACH_MIN_LEN = 20 + def _normalize_sub_task(st: dict[str, Any], order: int) -> dict[str, Any]: """Shape a sub_task entry to panel/src/types/index.ts::SubTask.""" @@ -343,16 +347,22 @@ class Choreographer: role_str: str, briefing: dict[str, Any], ) -> Envelope | None: - """Handle i_will_plan re-entry cases so the caller stays within PLR0911. + """Handle two distinct re-entry contracts for i_will_plan. - Returns an Envelope for two short-circuit paths: - - Idempotent re-entry: PM already owns the task in in_progress — touch - the heartbeat and return OK without re-running the spec gate. - - Recovery re-entry: task stuck in claimed after a crash — skip re-claim - (CLAIMED is not a valid source state for claim) and run set_plan+start. + Idempotent heartbeat: the PM already owns the task in in_progress — + touch the heartbeat and return OK without re-running the spec gate. + This is the crash-recovery path: a PM container that respawns after a + mid-run crash re-calls i_will_plan with thin args ("resume") and must + receive OK so it can proceed from where it left off. - Returns None when neither re-entry condition applies, signalling the - caller should continue to the normal claim-plan-start path. + Crash-recovery claim: task is stuck in claimed after a crash — skip + re-claim (claimed is not a valid source for the claim transition) and + run set_plan+start to complete the interrupted sequence. + + Returns None when neither condition applies, signalling the caller to + continue to the normal claim-plan-start path. PLR0911 budget is the + secondary reason this lives in a helper; the domain contract above is + the primary one. """ status = str(t.status) if status == "in_progress" and t.assigned_to == pm_agent_id: @@ -380,30 +390,43 @@ class Choreographer: task_id: UUID, briefing: dict[str, Any], ) -> Envelope | None: - """Wave A1 gate: PMs must supply at least one sub_task in i_will_plan. + """Wave A1 gate: PMs must supply approach (>= 20 chars) and sub_tasks. - Returns a rejection Envelope when the caller is a PM role and - rich_plan.sub_tasks is absent or empty; returns None to signal - the gate passed and the caller should continue. + Enforces both fields at the choreographer layer so direct service-layer + callers (MCP server, test fixtures, orchestrator-internal Python) cannot + persist a plan that bypassed the HTTP Pydantic boundary. + + Returns a rejection Envelope when the caller is a PM role and either + field is absent/insufficient; returns None to signal the gate passed. """ if role_str not in ("cell_pm", "main_pm"): return None - if rich_plan and rich_plan.get("sub_tasks"): + missing: list[str] = [] + field_hints: dict[str, str] = {} + approach_raw = (rich_plan or {}).get("approach", "") + if len(str(approach_raw).strip()) < _PM_APPROACH_MIN_LEN: + missing.append("approach") + field_hints["approach"] = ( + "approach must be a non-empty string of at least 20 characters " + "describing how the PM will decompose and route this task." + ) + if not (rich_plan and rich_plan.get("sub_tasks")): + missing.append("sub_tasks") + field_hints["sub_tasks"] = ( + "PMs must list at least one sub_task — a " + "non-empty list of {title, description}. " + "Each becomes a delegate target after i_will_plan." + ) + if not missing: return None return await self._emit_rejection( Envelope.incomplete_input( - missing=["sub_tasks"], - field_hints={ - "sub_tasks": ( - "PMs must list at least one sub_task — a " - "non-empty list of {title, description}. " - "Each becomes a delegate target after i_will_plan." - ) - }, + missing=missing, + field_hints=field_hints, remediate=( "re-issue i_will_plan(task_id, plan, approach, " "sub_tasks=[{'title': '...', 'description': '...'}, ...]) " - "with a non-empty sub_tasks list." + "with approach >= 20 chars and a non-empty sub_tasks list." ), context_briefing=briefing, ).with_introspection(task=task, role=role_str), @@ -1952,17 +1975,18 @@ class Choreographer: Atomic: spec.can_invoke_intent runs before any state mutation; the composed (claim, set_plan, start) sequence is wrapped in a savepoint by the runner so a mid-sequence failure rolls back - the DB. Idempotent re-entry: a respawned PM re-calling on a - task they already own in claimed/in_progress just refreshes - the heartbeat. + the DB. - The rich-plan kwargs (``approach``, ``technical_considerations``, - ``risks``, ``open_questions``) populate the panel's Plan tab - (pre-gateway parity). When any are non-empty/non-default they - are persisted as a structured ``TaskPlan``-shaped dict via - ``TaskService.set_plan``; otherwise ``plan`` is stored as a - narrative string. Empty defaults keep behavior backward-compatible - for callers that don't pass rich fields. + PM callers must supply ``approach`` (>= 20 chars) and a non-empty + ``sub_tasks`` list inside ``rich_plan`` — these are enforced in + ``_pm_sub_tasks_gate``. Developer callers may omit ``sub_tasks`` + (their plan is execution-shaped) but still need ``approach`` via + the HTTP schema layer. + + Control flow: re-entry check → if not re-entry → sub_tasks gate + → spec gate → claim+plan+start. The re-entry check must come first + so a respawned PM calling with thin args ("resume", no sub_tasks) + is short-circuited before the gate can reject them. """ t = await self.task.get(task_id) if t is None: @@ -1975,18 +1999,6 @@ class Choreographer: agent = await self.task.agent_for(pm_agent_id) role_str = str(agent.role) if agent is not None else "cell_pm" briefing = await self._briefing_for(pm_agent_id, task_id) - # Wave A1 (2026-05-12) — pre-gateway parity for _validate_claimed_start. - # PMs decompose; their plan MUST include at least one sub_task. Devs - # execute; sub_tasks list can be empty (their plan is execution-shaped). - if rejection := await self._pm_sub_tasks_gate( - role_str=role_str, - rich_plan=rich_plan, - task=t, - agent_id=pm_agent_id, - task_id=task_id, - briefing=briefing, - ): - return rejection try: role = spec_module.Role(role_str) except ValueError: @@ -2032,12 +2044,24 @@ class Choreographer: plan=effective_plan, verb_name="i_will_plan", ) - # Re-entry paths (idempotent + recovery) are handled in a shared helper - # so i_will_plan stays within the PLR0911 return-statement budget. + # Re-entry check runs first — a respawned PM with thin args ("resume", + # no sub_tasks) must short-circuit here before the sub_tasks gate. if reentry := await self._handle_pm_reentry( ctx, t, pm_agent_id, task_id, role_str, briefing ): return reentry + # Gate runs only for initial-claim paths (not re-entry). + # PMs decompose; their plan MUST include approach + at least one sub_task. + # Devs execute; sub_tasks list can be empty (their plan is execution-shaped). + if rejection := await self._pm_sub_tasks_gate( + role_str=role_str, + rich_plan=rich_plan, + task=t, + agent_id=pm_agent_id, + task_id=task_id, + briefing=briefing, + ): + return rejection if rejection := await self._claim_plan_start_gate(ctx, role, spec_ctx): return rejection envelope = await self._claim_plan_start_run(ctx, agent, spec_ctx) diff --git a/tests/unit/api/routes/v2/test_i_will_plan_rich_required.py b/tests/unit/api/routes/v2/test_i_will_plan_rich_required.py index c56122a6..3959eb4a 100644 --- a/tests/unit/api/routes/v2/test_i_will_plan_rich_required.py +++ b/tests/unit/api/routes/v2/test_i_will_plan_rich_required.py @@ -9,16 +9,17 @@ from __future__ import annotations from unittest.mock import AsyncMock, MagicMock from uuid import uuid4 -import pytest from fastapi import FastAPI from fastapi.testclient import TestClient - from roboco.api.deps import get_choreographer from roboco.api.routes.v2.flow_main_pm import router - +from roboco.api.schemas.v2.flow import IWillPlanRequest _AGENT_ID = "00000000-0000-0000-0004-000000000001" _HEADERS = {"X-Agent-ID": _AGENT_ID, "X-Agent-Role": "main_pm"} +_HTTP_UNPROCESSABLE = 422 +_HTTP_OK = 200 +_MIN_APPROACH_LEN = 20 def _make_envelope(error: str, missing: list[str] | None = None) -> MagicMock: @@ -52,7 +53,7 @@ def test_i_will_plan_rejects_missing_approach() -> None: # no approach, no sub_tasks }, ) - assert resp.status_code == 422, resp.text + assert resp.status_code == _HTTP_UNPROCESSABLE, resp.text detail = resp.json() assert any( "approach" in str(err.get("loc", [])) @@ -87,7 +88,7 @@ def test_i_will_plan_rejects_empty_subtasks_for_pm() -> None: # The gateway-side gate returns an envelope with error='incomplete_input' # and missing=['sub_tasks']. HTTP status is 200 (envelope carries error). body = resp.json() - if resp.status_code == 200: + if resp.status_code == _HTTP_OK: assert body.get("error") == "incomplete_input", body assert "sub_tasks" in (body.get("missing") or []), body else: @@ -98,8 +99,6 @@ def test_i_will_plan_rejects_empty_subtasks_for_pm() -> None: def test_i_will_plan_schema_accepts_rich_plan() -> None: """Pydantic schema accepts a fully-formed request with rich plan fields.""" - from roboco.api.schemas.v2.flow import IWillPlanRequest - req = IWillPlanRequest( task_id=uuid4(), plan="Route to backend", @@ -113,5 +112,5 @@ def test_i_will_plan_schema_accepts_rich_plan() -> None: risks=[], open_questions=[], ) - assert len(req.approach) >= 20 + assert len(req.approach) >= _MIN_APPROACH_LEN assert len(req.sub_tasks) == 1 diff --git a/tests/unit/gateway/test_i_will_plan_sub_tasks_gate.py b/tests/unit/gateway/test_i_will_plan_sub_tasks_gate.py new file mode 100644 index 00000000..977adc83 --- /dev/null +++ b/tests/unit/gateway/test_i_will_plan_sub_tasks_gate.py @@ -0,0 +1,382 @@ +"""Direct choreographer-layer unit tests for _pm_sub_tasks_gate. + +Critical #2: the HTTP-layer test (test_i_will_plan_rich_required.py) mocks +the choreographer and only exercises Pydantic validation. These tests call +Choreographer.i_will_plan() directly so the gate logic in _pm_sub_tasks_gate +is exercised at the correct layer. + +Also covers Important #1: approach is now enforced inside the gate (not only +at the HTTP Pydantic boundary) so direct service-layer callers (MCP, test +fixtures, orchestrator-internal) cannot bypass it. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 + +import pytest +from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps + +_MIN_APPROACH_LEN = 20 + + +# --------------------------------------------------------------------------- +# Shared fixture helpers — same pattern as test_choreographer_pm_extras.py +# --------------------------------------------------------------------------- + + +def _make_deps(**overrides: Any) -> ChoreographerDeps: + base: dict[str, Any] = { + "task": AsyncMock(), + "work_session": AsyncMock(), + "git": AsyncMock(), + "a2a": AsyncMock(), + "journal": AsyncMock(), + "audit": AsyncMock(), + "evidence_repo": AsyncMock(), + } + base.update(overrides) + task = base["task"] + task.session = MagicMock() + task.session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) + repo = base["evidence_repo"] + for method in ( + "list_unread_a2a", + "list_unread_mentions", + "list_pending_notifications", + "task_metadata_gaps", + "recent_team_activity", + "blockers_in_lane", + "journal_highlights_for_task", + ): + getattr(repo, method).return_value = [] + return ChoreographerDeps(**base) + + +def _pm_task_svc(task_id: object, *, role: str = "cell_pm") -> AsyncMock: + """Build a TaskService mock for a PM caller with a pending planning task.""" + task_svc = AsyncMock() + task_svc.get.return_value = MagicMock( + id=task_id, + status="pending", + plan=None, + assigned_to=None, + task_type="planning", + parent_task_id=None, + sequence=0, + team="backend", + commits=[], + pr_number=None, + branch_name=None, + quick_context=None, + ) + task_svc.agent_for.return_value = MagicMock( + id=uuid4(), role=role, team="backend", slug=None + ) + task_svc.list_in_progress_for_agent.return_value = [] + task_svc.list_paused_for_agent.return_value = [] + task_svc.get_subtasks.return_value = [] + task_svc.session = MagicMock() + task_svc.session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) + return task_svc + + +# --------------------------------------------------------------------------- +# Test 1: PM with empty sub_tasks → incomplete_input (sub_tasks in missing) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_pm_with_empty_sub_tasks_gets_incomplete_input() -> None: + """cell_pm with rich_plan.sub_tasks=[] must be rejected by _pm_sub_tasks_gate. + + The gate fires for PM roles when sub_tasks is absent or empty; the returned + envelope must carry error='incomplete_input' with 'sub_tasks' in missing. + """ + pm_id = uuid4() + task_id = uuid4() + task_svc = _pm_task_svc(task_id, role="cell_pm") + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.i_will_plan( + pm_id, + task_id, + plan="decompose work", + rich_plan={ + "approach": "Single-cell decomposition covering the full vertical slice.", + "sub_tasks": [], # empty — gate must fire + }, + ) + body = env.as_dict() + assert body["error"] == "incomplete_input", body + assert "sub_tasks" in (body.get("missing") or []), body + + +# --------------------------------------------------------------------------- +# Test 2: PM with rich_plan=None → incomplete_input +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_pm_with_missing_rich_plan_gets_incomplete_input() -> None: + """main_pm passing rich_plan=None must be rejected by _pm_sub_tasks_gate. + + The gate treats missing rich_plan the same as missing sub_tasks: the + envelope must carry error='incomplete_input' with 'sub_tasks' in missing. + """ + pm_id = uuid4() + task_id = uuid4() + task_svc = _pm_task_svc(task_id, role="main_pm") + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.i_will_plan( + pm_id, + task_id, + plan="decompose work", + rich_plan=None, # not supplied at all + ) + body = env.as_dict() + assert body["error"] == "incomplete_input", body + assert "sub_tasks" in (body.get("missing") or []), body + + +# --------------------------------------------------------------------------- +# Test 3: PM with filled sub_tasks → gate passes +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_pm_with_filled_sub_tasks_passes_gate() -> None: + """cell_pm with at least one sub_task must NOT be rejected by the gate. + + The call may still fail on downstream gates (spec lifecycle, journal + decision), but the sub_tasks gate itself must not fire. We assert + error != 'incomplete_input' and that the gate did not block. + """ + pm_id = uuid4() + task_id = uuid4() + task_svc = _pm_task_svc(task_id, role="cell_pm") + claimed = MagicMock( + id=task_id, + status="claimed", + plan=None, + assigned_to=pm_id, + task_type="planning", + ) + started = MagicMock( + id=task_id, + status="in_progress", + plan={"text": "x"}, + assigned_to=pm_id, + task_type="planning", + ) + task_svc.claim.return_value = claimed + task_svc.set_plan.return_value = claimed + task_svc.start.return_value = started + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.i_will_plan( + pm_id, + task_id, + plan="decompose work", + rich_plan={ + "approach": "Three-cell decomposition: backend, frontend, ux.", + "sub_tasks": [{"title": "Backend slice", "description": "API + DB"}], + }, + ) + body = env.as_dict() + # The gate must not have fired; the call may fail on other checks but not + # with incomplete_input from the sub_tasks gate. + assert body.get("error") != "incomplete_input", body + + +# --------------------------------------------------------------------------- +# Test 4: Developer with empty sub_tasks → gate does NOT fire +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_developer_with_empty_sub_tasks_passes_gate() -> None: + """Developers are not required to supply sub_tasks; the gate must skip them. + + A developer calling i_will_plan with an empty sub_tasks list should NOT + get incomplete_input from _pm_sub_tasks_gate. The gate is PM-only; + developers' plans are execution-shaped. + """ + dev_id = uuid4() + task_id = uuid4() + task_svc = AsyncMock() + code_task = MagicMock( + id=task_id, + status="pending", + plan=None, + assigned_to=None, + task_type="code", + parent_task_id=None, + sequence=0, + team="backend", + commits=[], + pr_number=None, + branch_name=None, + quick_context=None, + ) + task_svc.get.return_value = code_task + task_svc.agent_for.return_value = MagicMock( + id=dev_id, role="developer", team="backend", slug=None + ) + task_svc.list_in_progress_for_agent.return_value = [] + task_svc.list_paused_for_agent.return_value = [] + task_svc.get_subtasks.return_value = [] + task_svc.session = MagicMock() + task_svc.session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) + claimed = MagicMock( + id=task_id, + status="claimed", + plan=None, + assigned_to=dev_id, + task_type="code", + ) + started = MagicMock( + id=task_id, + status="in_progress", + plan={"text": "x"}, + assigned_to=dev_id, + task_type="code", + ) + task_svc.claim.return_value = claimed + task_svc.set_plan.return_value = claimed + task_svc.start.return_value = started + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.i_will_plan( + dev_id, + task_id, + plan="implement the feature", + rich_plan={ + "approach": "step-by-step TDD approach for the feature.", + "sub_tasks": [], + }, + ) + body = env.as_dict() + # Gate must NOT fire for developers. + assert body.get("error") != "incomplete_input", body + + +# --------------------------------------------------------------------------- +# Test 5 (Important #1): PM with sub_tasks filled but approach missing → rejection +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_pm_with_sub_tasks_but_missing_approach_gets_incomplete_input() -> None: + """cell_pm with non-empty sub_tasks but no approach must be rejected. + + The gate enforces approach (>= _MIN_APPROACH_LEN chars) as well as + sub_tasks. Direct service-layer callers bypass Pydantic so the gate is + the last line of defense. + """ + pm_id = uuid4() + task_id = uuid4() + task_svc = _pm_task_svc(task_id, role="cell_pm") + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.i_will_plan( + pm_id, + task_id, + plan="decompose work", + rich_plan={ + "approach": "", # empty — gate must fire + "sub_tasks": [{"title": "Backend slice", "description": "API + DB"}], + }, + ) + body = env.as_dict() + assert body["error"] == "incomplete_input", body + assert "approach" in (body.get("missing") or []), body + + +# --------------------------------------------------------------------------- +# Test 6 (Critical #1 ordering): re-entry by in_progress PM short-circuits +# BEFORE the gate, even with thin args (no sub_tasks) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_pm_reentry_in_progress_short_circuits_before_gate() -> None: + """A PM whose container crashed re-calls i_will_plan with thin args (no sub_tasks). + + If the task is already in_progress and assigned to this PM, the idempotent + re-entry path must fire and return OK before _pm_sub_tasks_gate is reached. + Without the ordering fix the gate would fire first and reject with + incomplete_input, breaking crash recovery. + """ + pm_id = uuid4() + task_id = uuid4() + task_svc = AsyncMock() + in_progress_task = MagicMock( + id=task_id, + status="in_progress", + plan={"text": "already set"}, + assigned_to=pm_id, # same PM — triggers re-entry + task_type="planning", + parent_task_id=None, + sequence=0, + team="backend", + commits=[], + pr_number=None, + branch_name="feature/backend/abc", + quick_context=None, + ) + task_svc.get.return_value = in_progress_task + task_svc.agent_for.return_value = MagicMock( + id=pm_id, role="cell_pm", team="backend", slug=None + ) + task_svc.list_in_progress_for_agent.return_value = [] + task_svc.list_paused_for_agent.return_value = [] + task_svc.get_subtasks.return_value = [] + task_svc.session = MagicMock() + task_svc.session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + # Thin args: just plan="resume", no sub_tasks — as a respawned PM would send. + env = await c.i_will_plan( + pm_id, + task_id, + plan="resume", + rich_plan=None, # no sub_tasks because PM is resuming, not planning + ) + body = env.as_dict() + # Re-entry must short-circuit to OK; incomplete_input means gate fired first. + assert body.get("error") is None, ( + f"Expected OK (re-entry short-circuit) but got " + f"error={body.get('error')!r}. " + "The gate is firing before _handle_pm_reentry — ordering bug not fixed." + ) + assert body.get("status") == "in_progress", body