feat(gateway): A1 plan-required-at-claim gate

i_will_plan now requires approach (min_length=20) at the schema and
non-empty sub_tasks at the gateway when the caller is a PM role. Restores
pre-gateway parity for _validate_claimed_start — agents could not
transition claimed -> in_progress without filling the rich plan.

Smoke run 3 (2026-05-11) showed PMs calling i_will_plan with just
plan='paragraph' and the gateway accepting it; Plan tab stayed empty
because no agent filled approach/sub_tasks/risks/open_questions.

Spec ref: docs/superpowers/specs/2026-05-12-post-smoke-3-fixes-design.md
section A1.
This commit is contained in:
Renn F
2026-05-12 02:30:00 +02:00
parent 62d1084a0c
commit a1009c05e8
10 changed files with 434 additions and 48 deletions
+5 -5
View File
@@ -122,11 +122,11 @@ class EscalateToCeoRequest(BaseModel):
class IWillPlanRequest(BaseModel):
task_id: UUID
plan: str = Field(..., min_length=1)
# Optional rich-plan fields. These persist into Task.plan as a structured
# dict matching roboco.models.task.TaskPlan, so the panel's Plan tab
# shows Approach / Sub-Tasks / Technical Considerations / Risks /
# Open Questions instead of an empty pane. Pre-gateway parity.
approach: str = ""
# Pre-gateway parity (Wave A1, 2026-05-12). Approach is REQUIRED — agents
# could not transition claimed → in_progress without filling this in the
# pre-gateway flow. The Plan tab depends on it; smoke run 3 confirmed
# the empty default lets agents through with thin plans.
approach: str = Field(..., min_length=20)
sub_tasks: list[dict[str, str]] = Field(
default_factory=list,
description="List of {title, description} — server assigns id + order",
+96 -23
View File
@@ -334,6 +334,84 @@ class Choreographer:
context_briefing=ctx.briefing,
).with_introspection(task=task, role=ctx.role_str)
async def _handle_pm_reentry(
self,
ctx: _ClaimPlanStartContext,
t: Any,
pm_agent_id: UUID,
task_id: UUID,
role_str: str,
briefing: dict[str, Any],
) -> Envelope | None:
"""Handle i_will_plan re-entry cases so the caller stays within PLR0911.
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.
Returns None when neither re-entry condition applies, signalling the
caller should continue to the normal claim-plan-start path.
"""
status = str(t.status)
if status == "in_progress" and t.assigned_to == pm_agent_id:
await self._touch(task_id)
return Envelope.ok(
status=status,
task_id=str(task_id),
next=spec_module._INTENT_VERBS["i_will_plan"].next_hint(t),
context_briefing=briefing,
).with_introspection(task=t, role=role_str)
if status == "claimed" and t.assigned_to == pm_agent_id:
envelope = await self._resume_from_claimed(ctx)
return await self._post_claim_journal_gate(
"i_will_plan", pm_agent_id, task_id, envelope
)
return None
async def _pm_sub_tasks_gate(
self,
*,
role_str: str,
rich_plan: dict[str, Any] | None,
task: Any,
agent_id: UUID,
task_id: UUID,
briefing: dict[str, Any],
) -> Envelope | None:
"""Wave A1 gate: PMs must supply at least one sub_task in i_will_plan.
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.
"""
if role_str not in ("cell_pm", "main_pm"):
return None
if rich_plan and rich_plan.get("sub_tasks"):
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."
)
},
remediate=(
"re-issue i_will_plan(task_id, plan, approach, "
"sub_tasks=[{'title': '...', 'description': '...'}, ...]) "
"with a non-empty sub_tasks list."
),
context_briefing=briefing,
).with_introspection(task=task, role=role_str),
agent_id=agent_id,
task_id=task_id,
verb="i_will_plan",
)
async def _emit_rejection(
self,
env: Envelope,
@@ -1897,6 +1975,18 @@ 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:
@@ -1942,29 +2032,12 @@ class Choreographer:
plan=effective_plan,
verb_name="i_will_plan",
)
# Idempotent re-entry: PM already owns the task in_progress.
# Touch heartbeat and short-circuit before the spec gate (which
# would otherwise reject because in_progress is not a source state
# for the composed `claim` action).
if str(t.status) == "in_progress" and t.assigned_to == pm_agent_id:
await self._touch(task_id)
return Envelope.ok(
status=str(t.status),
task_id=str(task_id),
next=spec_module._INTENT_VERBS["i_will_plan"].next_hint(t),
context_briefing=briefing,
).with_introspection(task=t, role=role_str)
# Recovery re-entry: task stuck in `claimed` (e.g. orchestrator restart
# or a partial-claim race) and the PM already owns it. The spec
# `claim` action's source-statuses do NOT include CLAIMED, so the spec
# gate would reject. Surface this as a runner call that runs only
# set_plan + start. Without this block, a PM reclaiming from a
# crashed mid-sequence would loop forever.
if str(t.status) == "claimed" and t.assigned_to == pm_agent_id:
envelope = await self._resume_from_claimed(ctx)
return await self._post_claim_journal_gate(
"i_will_plan", pm_agent_id, task_id, envelope
)
# Re-entry paths (idempotent + recovery) are handled in a shared helper
# so i_will_plan stays within the PLR0911 return-statement budget.
if reentry := await self._handle_pm_reentry(
ctx, t, pm_agent_id, task_id, role_str, briefing
):
return reentry
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)
+11 -1
View File
@@ -233,7 +233,17 @@ async def test_i_will_plan_dispatches_to_choreographer() -> None:
resp = client.post(
"/api/v2/flow/cell_pm/i_will_plan",
json={"task_id": _TASK_ID, "plan": "break into 3 subtasks for backend"},
json={
"task_id": _TASK_ID,
"plan": "break into 3 subtasks for backend",
"approach": (
"Decompose into backend API slice, QA verification pass, "
"and documentation update."
),
"sub_tasks": [
{"title": "Backend API slice", "description": "Implement endpoint"}
],
},
headers=_HEADERS,
)
+12 -1
View File
@@ -204,7 +204,18 @@ async def test_i_will_plan_dispatches_to_choreographer() -> None:
resp = client.post(
"/api/v2/flow/main_pm/i_will_plan",
json={"task_id": _TASK_ID, "plan": "split into backend, frontend, ux cells"},
json={
"task_id": _TASK_ID,
"plan": "split into backend, frontend, ux cells",
"approach": (
"Three-cell decomposition: backend handles API, frontend "
"handles UI integration, ux-ui handles design."
),
"sub_tasks": [
{"title": "Backend cell", "description": "API implementation"},
{"title": "Frontend cell", "description": "UI integration"},
],
},
headers=_HEADERS,
)
@@ -0,0 +1,117 @@
"""Wave A1: i_will_plan rejects PM claims that lack the rich plan shape.
Pre-gateway parity for `_validate_claimed_start` agents could not
transition claimed in_progress without filling approach + sub_tasks.
"""
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
_AGENT_ID = "00000000-0000-0000-0004-000000000001"
_HEADERS = {"X-Agent-ID": _AGENT_ID, "X-Agent-Role": "main_pm"}
def _make_envelope(error: str, missing: list[str] | None = None) -> MagicMock:
env = MagicMock()
payload: dict = {"error": error}
if missing is not None:
payload["missing"] = missing
env.as_dict.return_value = payload
env.correlation_id = None
return env
def _build_app(mock_choreographer: MagicMock | None = None) -> FastAPI:
"""Minimal FastAPI app with the flow_main_pm router and optional mocked dep."""
app = FastAPI()
app.include_router(router)
if mock_choreographer is not None:
app.dependency_overrides[get_choreographer] = lambda: mock_choreographer
return app
def test_i_will_plan_rejects_missing_approach() -> None:
"""A PM calling i_will_plan with bare `plan` (no approach) is rejected."""
client = TestClient(_build_app())
resp = client.post(
"/api/v2/flow/main_pm/i_will_plan",
headers=_HEADERS,
json={
"task_id": str(uuid4()),
"plan": "I will route this to backend cell",
# no approach, no sub_tasks
},
)
assert resp.status_code == 422, resp.text
detail = resp.json()
assert any(
"approach" in str(err.get("loc", []))
for err in detail.get("detail", [])
), detail
def test_i_will_plan_rejects_empty_subtasks_for_pm() -> None:
"""Approach satisfied but sub_tasks empty -> gateway rejects via incomplete_input.
Devs are NOT required to have sub_tasks; PMs are.
"""
mock_chore = MagicMock()
incomplete_env = _make_envelope(error="incomplete_input", missing=["sub_tasks"])
mock_chore.i_will_plan = AsyncMock(return_value=incomplete_env)
client = TestClient(_build_app(mock_chore))
resp = client.post(
"/api/v2/flow/main_pm/i_will_plan",
headers=_HEADERS,
json={
"task_id": str(uuid4()),
"plan": "Route to backend cell only",
"approach": (
"Single-cell decomposition: backend cell handles the "
"smoke test end-to-end; frontend and ux unaffected."
),
"sub_tasks": [], # empty — gateway rejects for PM
},
)
# Schema validation passes (sub_tasks is a valid empty list).
# 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:
assert body.get("error") == "incomplete_input", body
assert "sub_tasks" in (body.get("missing") or []), body
else:
# 422 or other 4xx; confirm sub_tasks is mentioned
text = resp.text.lower()
assert "sub_tasks" in text, text
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",
approach=(
"Single-cell decomposition for the smoke test: be-pm handles "
"git workflow validation end to end."
),
sub_tasks=[
{"title": "Backend slice", "description": "Branch + edit + PR"}
],
risks=[],
open_questions=[],
)
assert len(req.approach) >= 20
assert len(req.sub_tasks) == 1
@@ -383,7 +383,20 @@ async def test_cell_pm_can_plan_code_typed_parent_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="Decompose into 2 dev subtasks.")
env = await c.i_will_plan(
pm_id,
task_id,
plan="Decompose into 2 dev subtasks.",
rich_plan={
"approach": (
"Split code-typed parent into two developer-claimable subtasks: "
"one for API implementation, one for test coverage."
),
"sub_tasks": [
{"title": "API subtask", "description": "Implement endpoint"},
],
},
)
body = env.as_dict()
# The PM-cannot-execute-code rejection must NOT fire on i_will_plan.
assert body.get("error") != "not_authorized", (
@@ -422,7 +435,20 @@ async def test_pm_can_plan_non_code_parent() -> None:
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.i_will_plan(pm_id, task_id, plan="break it down")
env = await c.i_will_plan(
pm_id,
task_id,
plan="break it down",
rich_plan={
"approach": (
"Single-cell decomposition: backend handles the full scope; "
"no frontend or ux work required for this planning task."
),
"sub_tasks": [
{"title": "Backend planning slice", "description": "Scope and assign"}
],
},
)
assert env.error is None
@@ -312,7 +312,15 @@ async def test_i_will_plan_pm_with_already_active_task_rejects() -> None:
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)
env = await c.i_will_plan(
pm_id,
task_id,
plan="x" * 30,
rich_plan={
"approach": "Decompose planning task into backend and frontend subtasks.",
"sub_tasks": [{"title": "Slice A", "description": "backend API work"}],
},
)
body = env.as_dict()
assert body["error"] == "invalid_state"
assert "in_progress task" in body["message"]
@@ -362,7 +370,20 @@ async def test_i_will_plan_cell_pm_on_code_typed_parent_succeeds() -> None:
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.")
env = await c.i_will_plan(
pm_id,
task_id,
plan="Decompose into 2 dev subtasks.",
rich_plan={
"approach": (
"Split code-typed parent into developer-claimable subtasks: "
"one for API, one for test coverage validation."
),
"sub_tasks": [
{"title": "API subtask", "description": "Implement the endpoint"},
],
},
)
body = env.as_dict()
# The PM-cannot-execute-code rejection must NOT fire on i_will_plan.
assert body.get("error") != "not_authorized", (
@@ -393,7 +414,15 @@ async def test_i_will_plan_pending_claim_fails() -> None:
task_svc.claim.return_value = None # claim fails
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.i_will_plan(pm_id, task_id, plan="my plan that is long enough")
env = await c.i_will_plan(
pm_id,
task_id,
plan="my plan that is long enough",
rich_plan={
"approach": "Decompose planning task into backend and frontend subtasks.",
"sub_tasks": [{"title": "Slice A", "description": "backend API work"}],
},
)
body = env.as_dict()
assert body["error"] == "invalid_state"
@@ -1002,7 +1031,15 @@ async def test_i_will_plan_pending_claim_returns_none_emit_rejection() -> None:
task_svc.claim.return_value = None
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.i_will_plan(pm_id, task_id, plan="my plan that is long enough")
env = await c.i_will_plan(
pm_id,
task_id,
plan="my plan that is long enough",
rich_plan={
"approach": "Decompose planning task into backend and frontend subtasks.",
"sub_tasks": [{"title": "Slice A", "description": "backend API work"}],
},
)
body = env.as_dict()
assert body["error"] == "invalid_state"
assert "verb runner failed" in body["message"]
@@ -103,7 +103,21 @@ async def test_i_will_plan_claims_starts_and_sets_plan() -> None:
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.i_will_plan(pm_id, task_id, plan="break the work into 3 subtasks")
env = await c.i_will_plan(
pm_id,
task_id,
plan="break the work into 3 subtasks",
rich_plan={
"approach": (
"Three-cell decomposition: backend, frontend, and ux each "
"own a vertical slice of the work."
),
"sub_tasks": [
{"title": "Backend slice", "description": "API + DB"},
{"title": "Frontend slice", "description": "UI integration"},
],
},
)
assert env.error is None
assert env.status == "in_progress"
task_svc.claim.assert_awaited_once_with(task_id, pm_id)
@@ -159,7 +173,21 @@ async def test_i_will_plan_blocks_when_journal_decision_at_claim_missing() -> No
deps = _make_deps(task=task_svc, journal=journal_svc)
c = Choreographer(deps)
env = await c.i_will_plan(pm_id, task_id, plan="break the work into 3 subtasks")
env = await c.i_will_plan(
pm_id,
task_id,
plan="break the work into 3 subtasks",
rich_plan={
"approach": (
"Three-cell decomposition: backend, frontend, and ux each "
"own a vertical slice of the work."
),
"sub_tasks": [
{"title": "Backend slice", "description": "API + DB"},
{"title": "Frontend slice", "description": "UI integration"},
],
},
)
body = env.as_dict()
assert body["error"] == "tracing_gap"
assert "journal:decision_at_claim" in body["missing"]
@@ -190,12 +218,20 @@ async def test_i_will_plan_rejects_non_pending_state() -> None:
pm_id = uuid4()
task_id = uuid4()
task_svc = AsyncMock()
task_svc.get.return_value = MagicMock(id=task_id, status="in_progress")
task_svc.get.return_value = MagicMock(id=task_id, status="in_progress", assigned_to=uuid4())
task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend")
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="x",
rich_plan={
"approach": "Single-cell decomposition: backend handles all scope.",
"sub_tasks": [{"title": "Slice A", "description": "backend API work"}],
},
)
body = env.as_dict()
assert body["error"] == "invalid_state"
@@ -247,7 +283,21 @@ async def test_i_will_plan_calls_claim_when_pre_assigned_and_pending() -> None:
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.i_will_plan(pm_id, task_id, plan="distribute to be-pm and fe-pm")
env = await c.i_will_plan(
pm_id,
task_id,
plan="distribute to be-pm and fe-pm",
rich_plan={
"approach": (
"Two-cell dispatch: be-pm owns backend vertical, "
"fe-pm owns frontend vertical."
),
"sub_tasks": [
{"title": "Backend cell", "description": "Assign to be-pm"},
{"title": "Frontend cell", "description": "Assign to fe-pm"},
],
},
)
assert env.error is None
assert env.status == "in_progress"
@@ -297,7 +347,15 @@ async def test_i_will_plan_surfaces_start_failure_instead_of_faking_ok() -> 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="x",
rich_plan={
"approach": "Single-cell decomposition: backend handles all scope.",
"sub_tasks": [{"title": "Slice A", "description": "backend API work"}],
},
)
body = env.as_dict()
assert body["error"] == "invalid_state"
assert "start failed" in body["message"]
@@ -329,7 +387,15 @@ async def test_i_will_plan_idempotent_when_already_in_progress_for_caller() -> N
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.i_will_plan(pm_id, task_id, plan="re-entry plan")
env = await c.i_will_plan(
pm_id,
task_id,
plan="re-entry plan",
rich_plan={
"approach": "Idempotent re-entry: task already in progress, refresh heartbeat.",
"sub_tasks": [{"title": "Re-entry subtask", "description": "Resume work"}],
},
)
assert env.error is None
assert env.status == "in_progress"
@@ -380,7 +446,15 @@ async def test_i_will_plan_recovery_when_already_claimed_for_caller() -> None:
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.i_will_plan(pm_id, task_id, plan="re-entry plan")
env = await c.i_will_plan(
pm_id,
task_id,
plan="re-entry plan",
rich_plan={
"approach": "Recovery re-entry: task claimed but not started; run set_plan + start.",
"sub_tasks": [{"title": "Recovery subtask", "description": "Resume from claimed"}],
},
)
assert env.error is None
assert env.status == "in_progress"
@@ -412,7 +486,15 @@ async def test_i_will_plan_still_rejects_in_progress_for_other_agent() -> 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="x",
rich_plan={
"approach": "Single-cell decomposition: backend handles all scope.",
"sub_tasks": [{"title": "Slice A", "description": "backend API work"}],
},
)
body = env.as_dict()
assert body["error"] == "invalid_state"
@@ -430,7 +512,15 @@ async def test_i_will_plan_returns_tracing_gap_without_plan() -> None:
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.i_will_plan(pm_id, task_id, plan="")
env = await c.i_will_plan(
pm_id,
task_id,
plan="",
rich_plan={
"approach": "Single-cell decomposition: backend handles all scope.",
"sub_tasks": [{"title": "Slice A", "description": "backend API work"}],
},
)
body = env.as_dict()
assert body["error"] == "tracing_gap"
assert "plan" in body["missing"]
+15 -1
View File
@@ -260,7 +260,21 @@ async def test_i_will_plan_calls_claim_and_start_with_task_id_first() -> None:
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
env = await c.i_will_plan(pm_id, task_id, plan="break the work into 3 subtasks")
env = await c.i_will_plan(
pm_id,
task_id,
plan="break the work into 3 subtasks",
rich_plan={
"approach": (
"Three-cell decomposition: backend, frontend, and ux each "
"own a vertical slice of the work."
),
"sub_tasks": [
{"title": "Backend slice", "description": "API + DB"},
{"title": "Frontend slice", "description": "UI integration"},
],
},
)
task_svc.claim.assert_awaited_once_with(task_id, pm_id)
task_svc.start.assert_awaited_once_with(task_id, pm_id)
+9 -1
View File
@@ -240,6 +240,14 @@ async def test_i_will_plan_calls_heartbeat() -> None:
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
await c.i_will_plan(pm_id, tid, plan="plan-text")
await c.i_will_plan(
pm_id,
tid,
plan="plan-text",
rich_plan={
"approach": "Single-cell decomposition: backend handles all scope.",
"sub_tasks": [{"title": "Slice A", "description": "backend API work"}],
},
)
task_svc.heartbeat.assert_awaited_with(tid)