feat(gateway): developers author the full rich plan, at parity with PMs

A dev's i_will_work_on stored a flat {text} plan and routed steps to
progress, so the dev leaf's Plan tab rendered empty (no approach,
sub_tasks, technical_considerations, risks) — zero audit/tracing on the
task that does the actual work.

i_will_work_on now captures the same rich plan a PM authors via
i_will_plan: plan(>=150) doubles as approach, steps become sub_tasks,
plus technical_considerations + risks (open_questions optional). A new
_dev_plan_gate enforces them on FRESH claims only (re-entry/recovery
short-circuit before it). set_plan gains a no-downgrade guard so a
flaked-then-recovered dev can't clobber its rich plan back to flat (the
actual mechanism behind the empty leaf).

tests/unit/gateway/test_choreographer_claim_guards.py

@@ -33,6 +33,21 @@ _STEPS = [
         ),
     }
 ]
+# Full parity: a fresh dev claim authors the same rich plan a PM does.
+# These satisfy _dev_plan_gate (plan/approach >= 150 chars,
+# technical_considerations, risks).
+_GOOD_PLAN = (
+    "Append the timestamp HTML comment to the very bottom of README.md without "
+    "touching any other line, then commit it on the task branch and open a PR. "
+    "Verify the diff is a single-line addition before submitting for QA."
+)
+_GOOD_TC = ["Use a trailing newline so the comment sits on its own line."]
+_GOOD_RISKS = [
+    {
+        "risk": "An accidental reformat of README.md balloons the diff.",
+        "mitigation": "Append only; assert the diff touches one line pre-commit.",
+    }
+]
 
 
 def _make_deps(**overrides: Any) -> ChoreographerDeps:
@@ -189,7 +204,14 @@ async def test_i_will_work_on_allows_when_earlier_sibling_terminal() -> None:
     deps = _make_deps(task=task_svc)
     c = Choreographer(deps)
 
-    env = await c.i_will_work_on(agent_id, target_id, steps=_STEPS)
+    env = await c.i_will_work_on(
+        agent_id,
+        target_id,
+        plan=_GOOD_PLAN,
+        steps=_STEPS,
+        technical_considerations=_GOOD_TC,
+        risks=_GOOD_RISKS,
+    )
     assert env.error is None
     task_svc.claim.assert_awaited_once_with(target_id, agent_id)
 
@@ -219,7 +241,14 @@ async def test_root_task_no_sequence_check() -> None:
     deps = _make_deps(task=task_svc)
     c = Choreographer(deps)
 
-    env = await c.i_will_work_on(agent_id, target_id, steps=_STEPS)
+    env = await c.i_will_work_on(
+        agent_id,
+        target_id,
+        plan=_GOOD_PLAN,
+        steps=_STEPS,
+        technical_considerations=_GOOD_TC,
+        risks=_GOOD_RISKS,
+    )
     assert env.error is None
     # Sequence check should not have queried siblings on a root task
     task_svc.get_subtasks.assert_not_awaited()

tests/unit/gateway/test_choreographer_dev.py

@@ -19,6 +19,21 @@ _STEPS = [
         ),
     }
 ]
+# Full parity: a fresh dev claim authors the same rich plan a PM does.
+# These satisfy _dev_plan_gate (plan/approach >= 150 chars,
+# technical_considerations, risks).
+_GOOD_PLAN = (
+    "Append the timestamp HTML comment to the very bottom of README.md without "
+    "touching any other line, then commit it on the task branch and open a PR. "
+    "Verify the diff is a single-line addition before submitting for QA."
+)
+_GOOD_TC = ["Use a trailing newline so the comment sits on its own line."]
+_GOOD_RISKS = [
+    {
+        "risk": "An accidental reformat of README.md balloons the diff.",
+        "mitigation": "Append only; assert the diff touches one line pre-commit.",
+    }
+]
 
 
 def _make_deps(**overrides: AsyncMock) -> ChoreographerDeps:
@@ -149,7 +164,14 @@ async def test_i_will_work_on_pending_with_plan() -> None:
     deps = _make_deps(task=task_svc)
     c = Choreographer(deps)
 
-    env = await c.i_will_work_on(agent_id, task_id, plan="do x then y", steps=_STEPS)
+    env = await c.i_will_work_on(
+        agent_id,
+        task_id,
+        plan=_GOOD_PLAN,
+        steps=_STEPS,
+        technical_considerations=_GOOD_TC,
+        risks=_GOOD_RISKS,
+    )
     assert env.error is None
     assert env.status == "in_progress"
     task_svc.claim.assert_awaited_once_with(task_id, agent_id)
@@ -238,7 +260,14 @@ async def test_i_will_work_on_needs_revision_re_starts() -> None:
     deps = _make_deps(task=task_svc)
     c = Choreographer(deps)
 
-    env = await c.i_will_work_on(agent_id, task_id, steps=_STEPS)
+    env = await c.i_will_work_on(
+        agent_id,
+        task_id,
+        plan=_GOOD_PLAN,
+        steps=_STEPS,
+        technical_considerations=_GOOD_TC,
+        risks=_GOOD_RISKS,
+    )
     assert env.status == "in_progress"
     task_svc.start.assert_awaited_once_with(task_id, agent_id)
 
@@ -343,7 +372,14 @@ async def test_i_will_work_on_blocks_when_journal_note_at_claim_missing() -> Non
     deps = _make_deps(task=task_svc, journal=journal_svc)
     c = Choreographer(deps)
 
-    env = await c.i_will_work_on(agent_id, task_id, plan="do x then y", steps=_STEPS)
+    env = await c.i_will_work_on(
+        agent_id,
+        task_id,
+        plan=_GOOD_PLAN,
+        steps=_STEPS,
+        technical_considerations=_GOOD_TC,
+        risks=_GOOD_RISKS,
+    )
     body = env.as_dict()
     assert body["error"] == "tracing_gap"
     assert "journal:note_at_claim" in body["missing"]

tests/unit/gateway/test_choreographer_impl_branches.py

@@ -30,6 +30,21 @@ _STEPS = [
         ),
     }
 ]
+# Full parity: a fresh dev claim authors the same rich plan a PM does.
+# These satisfy _dev_plan_gate (plan/approach >= 150 chars,
+# technical_considerations, risks).
+_GOOD_PLAN = (
+    "Append the timestamp HTML comment to the very bottom of README.md without "
+    "touching any other line, then commit it on the task branch and open a PR. "
+    "Verify the diff is a single-line addition before submitting for QA."
+)
+_GOOD_TC = ["Use a trailing newline so the comment sits on its own line."]
+_GOOD_RISKS = [
+    {
+        "risk": "An accidental reformat of README.md balloons the diff.",
+        "mitigation": "Append only; assert the diff touches one line pre-commit.",
+    }
+]
 
 
 def _wire_dev_task_svc(
@@ -152,7 +167,14 @@ async def test_i_will_work_on_pending_claim_raises_returns_invalid_state() -> No
     task_svc.claim.side_effect = RuntimeError("workspace down")
     deps = _make_deps(task=task_svc)
     c = Choreographer(deps)
-    env = await c.i_will_work_on(agent_id, task_id, plan="plan", steps=_STEPS)
+    env = await c.i_will_work_on(
+        agent_id,
+        task_id,
+        plan=_GOOD_PLAN,
+        steps=_STEPS,
+        technical_considerations=_GOOD_TC,
+        risks=_GOOD_RISKS,
+    )
     body = env.as_dict()
     assert body["error"] == "invalid_state"
     assert "verb runner failed" in body["message"]
@@ -168,7 +190,14 @@ async def test_i_will_work_on_pending_claim_returns_none_invalid_state() -> None
     task_svc.claim.return_value = None
     deps = _make_deps(task=task_svc)
     c = Choreographer(deps)
-    env = await c.i_will_work_on(agent_id, task_id, plan="plan", steps=_STEPS)
+    env = await c.i_will_work_on(
+        agent_id,
+        task_id,
+        plan=_GOOD_PLAN,
+        steps=_STEPS,
+        technical_considerations=_GOOD_TC,
+        risks=_GOOD_RISKS,
+    )
     body = env.as_dict()
     assert body["error"] == "invalid_state"
 
@@ -214,7 +243,14 @@ async def test_i_will_work_on_start_returns_none_invalid_state() -> None:
     task_svc.start.return_value = None  # start fails
     deps = _make_deps(task=task_svc)
     c = Choreographer(deps)
-    env = await c.i_will_work_on(agent_id, task_id, plan="ok plan", steps=_STEPS)
+    env = await c.i_will_work_on(
+        agent_id,
+        task_id,
+        plan=_GOOD_PLAN,
+        steps=_STEPS,
+        technical_considerations=_GOOD_TC,
+        risks=_GOOD_RISKS,
+    )
     body = env.as_dict()
     assert body["error"] == "invalid_state"
     assert "start failed" in body["message"]
@@ -237,7 +273,14 @@ async def test_needs_revision_branch_claim_fails_invalid_state() -> None:
     task_svc.claim.return_value = None
     deps = _make_deps(task=task_svc)
     c = Choreographer(deps)
-    env = await c.i_will_work_on(agent_id, task_id, plan="ok", steps=_STEPS)
+    env = await c.i_will_work_on(
+        agent_id,
+        task_id,
+        plan=_GOOD_PLAN,
+        steps=_STEPS,
+        technical_considerations=_GOOD_TC,
+        risks=_GOOD_RISKS,
+    )
     body = env.as_dict()
     assert body["error"] == "invalid_state"
 
@@ -253,7 +296,14 @@ async def test_needs_revision_branch_start_fails() -> None:
     task_svc.start.return_value = None
     deps = _make_deps(task=task_svc)
     c = Choreographer(deps)
-    env = await c.i_will_work_on(agent_id, task_id, plan="ok", steps=_STEPS)
+    env = await c.i_will_work_on(
+        agent_id,
+        task_id,
+        plan=_GOOD_PLAN,
+        steps=_STEPS,
+        technical_considerations=_GOOD_TC,
+        risks=_GOOD_RISKS,
+    )
     body = env.as_dict()
     assert body["error"] == "invalid_state"
 
@@ -1147,7 +1197,14 @@ async def test_i_will_work_on_envelope_carries_introspection_on_success() -> Non
     task_svc.start.return_value = claimed_task
     deps = _make_deps(task=task_svc)
     c = Choreographer(deps)
-    env = await c.i_will_work_on(agent_id, task_id, plan="ok plan", steps=_STEPS)
+    env = await c.i_will_work_on(
+        agent_id,
+        task_id,
+        plan=_GOOD_PLAN,
+        steps=_STEPS,
+        technical_considerations=_GOOD_TC,
+        risks=_GOOD_RISKS,
+    )
     body = env.as_dict()
     assert body["error"] is None
     assert body["current_state"] == "in_progress"

tests/unit/gateway/test_claim_arg_order.py

@@ -36,6 +36,21 @@ _STEPS = [
         ),
     }
 ]
+# Full parity: a fresh dev claim authors the same rich plan a PM does.
+# These satisfy _dev_plan_gate (plan/approach >= 150 chars,
+# technical_considerations, risks).
+_GOOD_PLAN = (
+    "Append the timestamp HTML comment to the very bottom of README.md without "
+    "touching any other line, then commit it on the task branch and open a PR. "
+    "Verify the diff is a single-line addition before submitting for QA."
+)
+_GOOD_TC = ["Use a trailing newline so the comment sits on its own line."]
+_GOOD_RISKS = [
+    {
+        "risk": "An accidental reformat of README.md balloons the diff.",
+        "mitigation": "Append only; assert the diff touches one line pre-commit.",
+    }
+]
 
 
 def _make_deps(**overrides: Any) -> ChoreographerDeps:
@@ -132,7 +147,14 @@ async def test_i_will_work_on_pending_calls_claim_with_task_id_first() -> None:
     deps = _make_deps(task=task_svc)
     c = Choreographer(deps)
 
-    env = await c.i_will_work_on(agent_id, task_id, plan="x", steps=_STEPS)
+    env = await c.i_will_work_on(
+        agent_id,
+        task_id,
+        plan=_GOOD_PLAN,
+        steps=_STEPS,
+        technical_considerations=_GOOD_TC,
+        risks=_GOOD_RISKS,
+    )
 
     # Service signature is (task_id, agent_id, ...) — pin that order.
     task_svc.claim.assert_awaited_once_with(task_id, agent_id)
@@ -184,7 +206,14 @@ async def test_i_will_work_on_needs_revision_calls_start_with_task_id_first() ->
     deps = _make_deps(task=task_svc)
     c = Choreographer(deps)
 
-    env = await c.i_will_work_on(agent_id, task_id, steps=_STEPS)
+    env = await c.i_will_work_on(
+        agent_id,
+        task_id,
+        plan=_GOOD_PLAN,
+        steps=_STEPS,
+        technical_considerations=_GOOD_TC,
+        risks=_GOOD_RISKS,
+    )
 
     task_svc.start.assert_awaited_once_with(task_id, agent_id)
     task_svc.claim.assert_awaited_once_with(task_id, agent_id)

tests/unit/gateway/test_dev_steps_gate.py

@@ -22,6 +22,31 @@ _GOOD_STEP_DESC = (
     "be-dev-1 prepends the smoke-test HTML comment above the README H1, "
     "leaving the rest of the file untouched, then stages the change."
 )
+# Full parity: a fresh dev claim authors the same rich plan a PM does. These
+# satisfy _dev_plan_gate (plan/approach >= 150 chars, substantive steps,
+# technical_considerations, risks).
+_GOOD_PLAN = (
+    "Append the timestamp HTML comment to the very bottom of README.md without "
+    "touching any other line, then commit it on the task branch and open a PR. "
+    "Verify the diff is a single-line addition before submitting for QA."
+)
+_GOOD_TC = ["Use a trailing newline so the comment sits on its own line."]
+_GOOD_RISKS = [
+    {
+        "risk": "An accidental reformat of README.md balloons the diff.",
+        "mitigation": "Append only; assert the diff touches one line pre-commit.",
+    }
+]
+
+
+def _full_plan_kwargs(steps: list[dict[str, str]]) -> dict[str, Any]:
+    """The full rich-plan kwargs a fresh dev claim must supply post-parity."""
+    return {
+        "plan": _GOOD_PLAN,
+        "steps": steps,
+        "technical_considerations": _GOOD_TC,
+        "risks": _GOOD_RISKS,
+    }
 
 
 def _make_deps(**overrides: Any) -> ChoreographerDeps:
@@ -137,22 +162,42 @@ async def test_dev_with_substantive_steps_passes_gate_and_persists_checklist() -
         {"title": "Edit README", "description": _GOOD_STEP_DESC},
         {"title": "Commit + open PR", "description": _GOOD_STEP_DESC},
     ]
-    env = await c.i_will_work_on(
-        dev_id,
-        task_id,
-        plan="implement the README change end to end",
-        steps=steps_in,
-    )
+    env = await c.i_will_work_on(dev_id, task_id, **_full_plan_kwargs(steps_in))
     body = env.as_dict()
     assert body.get("error") != "incomplete_input", body
-    # Steps were layered into the panel-shaped plan dict and persisted.
+    # The full rich plan was layered into the panel-shaped dict and persisted,
+    # so the dev leaf's Plan tab renders like a PM's (approach + sub_tasks +
+    # technical_considerations + risks).
     svc.set_plan.assert_awaited_once()
     persisted = svc.set_plan.await_args.args[1]
     assert isinstance(persisted, dict), persisted
     sub_tasks = persisted.get("sub_tasks") or []
     assert len(sub_tasks) == len(steps_in), persisted
     assert all(st.get("title") for st in sub_tasks), sub_tasks
-    assert persisted.get("text") == "implement the README change end to end"
+    assert persisted.get("approach") == _GOOD_PLAN
+    assert persisted.get("technical_considerations") == _GOOD_TC
+    assert len(persisted.get("risks") or []) == 1
+
+
+@pytest.mark.asyncio
+async def test_dev_fresh_claim_missing_considerations_and_risks_rejected() -> None:
+    """Full parity: substantive steps + long plan are not enough — a fresh dev
+    claim must also carry technical_considerations and risks."""
+    dev_id = uuid4()
+    task_id = uuid4()
+    c = Choreographer(_make_deps(task=_dev_task_svc(task_id)))
+
+    env = await c.i_will_work_on(
+        dev_id,
+        task_id,
+        plan=_GOOD_PLAN,
+        steps=[{"title": "Edit README", "description": _GOOD_STEP_DESC}],
+    )
+    body = env.as_dict()
+    assert body["error"] == "incomplete_input", body
+    missing = body.get("missing") or []
+    assert "technical_considerations" in missing, body
+    assert "risks" in missing, body
 
 
 @pytest.mark.asyncio

tests/unit/gateway/test_work_session_auto_create.py

@@ -30,6 +30,21 @@ _STEPS = [
         ),
     }
 ]
+# Full parity: a fresh dev claim authors the same rich plan a PM does.
+# These satisfy _dev_plan_gate (plan/approach >= 150 chars,
+# technical_considerations, risks).
+_GOOD_PLAN = (
+    "Append the timestamp HTML comment to the very bottom of README.md without "
+    "touching any other line, then commit it on the task branch and open a PR. "
+    "Verify the diff is a single-line addition before submitting for QA."
+)
+_GOOD_TC = ["Use a trailing newline so the comment sits on its own line."]
+_GOOD_RISKS = [
+    {
+        "risk": "An accidental reformat of README.md balloons the diff.",
+        "mitigation": "Append only; assert the diff touches one line pre-commit.",
+    }
+]
 
 
 def _make_task_svc(agent_id, task_id, *, status: str):
@@ -132,7 +147,14 @@ async def test_i_will_work_on_calls_ensure_work_session() -> None:
     deps = _make_deps(task_svc)
     c = Choreographer(deps)
 
-    env = await c.i_will_work_on(agent_id, task_id, plan="do x then y", steps=_STEPS)
+    env = await c.i_will_work_on(
+        agent_id,
+        task_id,
+        plan=_GOOD_PLAN,
+        steps=_STEPS,
+        technical_considerations=_GOOD_TC,
+        risks=_GOOD_RISKS,
+    )
 
     assert env.error is None, f"Expected ok, got error={env.error} msg={env.message}"
     assert env.status == "in_progress"
@@ -267,7 +289,14 @@ async def test_ensure_work_session_not_called_when_start_fails() -> None:
     deps = _make_deps(task_svc)
     c = Choreographer(deps)
 
-    env = await c.i_will_work_on(agent_id, task_id, plan="do x then y", steps=_STEPS)
+    env = await c.i_will_work_on(
+        agent_id,
+        task_id,
+        plan=_GOOD_PLAN,
+        steps=_STEPS,
+        technical_considerations=_GOOD_TC,
+        risks=_GOOD_RISKS,
+    )
 
     assert env.error == "invalid_state"
     task_svc.ensure_work_session.assert_not_awaited()

tests/unit/mcp_servers/test_flow_server.py

@@ -144,6 +144,9 @@ def test_i_will_work_on_passes_plan(flow_module: types.ModuleType) -> None:
         "task_id": "task-uuid",
         "plan": "my plan",
         "steps": [],
+        "technical_considerations": [],
+        "risks": [],
+        "open_questions": [],
     }
     assert "/api/v2/flow/developer/i_will_work_on" in args[0]
 
@@ -155,7 +158,14 @@ def test_i_will_work_on_plan_defaults_to_none(flow_module: types.ModuleType) ->
         flow_module.i_will_work_on("task-uuid")
 
     _, kwargs = fake_client.post.call_args
-    assert kwargs["json"] == {"task_id": "task-uuid", "plan": None, "steps": []}
+    assert kwargs["json"] == {
+        "task_id": "task-uuid",
+        "plan": None,
+        "steps": [],
+        "technical_considerations": [],
+        "risks": [],
+        "open_questions": [],
+    }
 
 
 def test_i_will_work_on_passes_steps(flow_module: types.ModuleType) -> None:
@@ -176,6 +186,9 @@ def test_i_will_work_on_passes_steps(flow_module: types.ModuleType) -> None:
         "task_id": "task-uuid",
         "plan": "p",
         "steps": steps,
+        "technical_considerations": [],
+        "risks": [],
+        "open_questions": [],
     }