feat(steward): materialize a confirmed team plan through the Todo owner

loopx/control_plane/work_items/governed_transition_proposal.py

@@ -20,6 +20,7 @@
 from ..todos.contract import (
     TODO_STATUS_DONE,
     TODO_STATUS_OPEN,
+    TODO_TASK_CLASS_ADVANCEMENT,
     TODO_TASK_CLASS_MONITOR,
     normalize_todo_capability_binding_ref,
 )
@@ -50,9 +51,12 @@ class GovernedTransitionSettlementPhase(StrEnum):
     POST_SETTLEMENT = "post_settlement"
 
 
+STEWARD_TEAM_PLAN_PREVIEW_KIND = "steward_team_plan_preview"
+
 _SETTLEMENT_PHASE_BY_PROPOSAL_KIND = {
     "continuous_monitor_upsert": GovernedTransitionSettlementPhase.PRE_SETTLEMENT,
     "continuous_monitor_complete": GovernedTransitionSettlementPhase.POST_SETTLEMENT,
+    STEWARD_TEAM_PLAN_PREVIEW_KIND: GovernedTransitionSettlementPhase.PRE_SETTLEMENT,
 }
 
 
@@ -94,6 +98,7 @@ def validate_governed_transition_receipts(
         if receipt.get("kind") not in {
             "continuous_monitor_upsert",
             "continuous_monitor_complete",
+            STEWARD_TEAM_PLAN_PREVIEW_KIND,
         }:
             raise ValueError("governed transition proposal receipt kind is invalid")
         if receipt.get("status") != "committed":
@@ -216,6 +221,80 @@ def _upsert_monitor(
     }
 
 
+def _apply_team_plan(
+    *,
+    registry_path: Path,
+    goal_id: str,
+    agent_id: str,
+    proposal: Mapping[str, Any],
+) -> dict[str, Any]:
+    """Create the confirmed lanes' first bounded Todos through the Todo owner.
+
+    The plan is re-validated here against this Goal's registered Agents and the
+    shipped advancement action kinds, so a proposal cannot become work by
+    bypassing admission. Only lanes the preview already marked ready are
+    materialized; a lane the preview reported as a gap stays a gap and creates
+    nothing, and the canonical Todo owner decides whether a row is added or
+    reused, which makes a replayed settlement idempotent.
+    """
+
+    from ...agent_registry import registered_agent_ids_for_goal
+    from ...history import load_registry
+    from ...registry import registry_goals
+    from ..todos.contract import TODO_ACTION_KIND_ADVANCEMENT_VALUES
+
+    registry = load_registry(registry_path)
+    goal = next(
+        (
+            item
+            for item in registry_goals(registry)
+            if str(item.get("id") or "") == goal_id
+        ),
+        None,
+    )
+    if goal is None:
+        raise ValueError("steward team plan proposal names an unknown Goal")
+    preview = validate_steward_team_plan_preview(
+        proposal,
+        registered_agent_ids=registered_agent_ids_for_goal(goal),
+        supported_action_kinds=sorted(TODO_ACTION_KIND_ADVANCEMENT_VALUES),
+    )
+    created: list[str] = []
+    reused: list[str] = []
+    for lane in preview["lanes"]:
+        if lane.get("staffing") != "ready":
+            continue
+        first_todo = lane["first_todo"]
+        result = add_goal_todo(
+            registry_path=Path(registry_path).expanduser(),
+            goal_id=goal_id,
+            role="agent",
+            text=str(first_todo["text"]),
+            status=TODO_STATUS_OPEN,
+            task_class=TODO_TASK_CLASS_ADVANCEMENT,
+            action_kind=str(first_todo["action_kind"]),
+            claimed_by=str(lane["agent_id"]),
+            agent_id=str(lane["agent_id"]),
+        )
+        if result.get("added"):
+            created.append(str(result["todo_id"]))
+        else:
+            reused.append(str(result["todo_id"]))
+    # The receipt names every lane Todo this settlement ensured, whether the
+    # canonical owner added it or found it already present, so a replayed
+    # settlement still reports the same identities instead of an empty one.
+    lane_todo_ids = [*created, *reused]
+    return {
+        "action": "created" if created else "reused",
+        "todo_id": lane_todo_ids[0] if lane_todo_ids else "",
+        "target_key": None,
+        "created_todo_ids": created,
+        "lane_todo_ids": lane_todo_ids,
+        "reused_lane_count": len(reused),
+        "gap_count": len(preview["gaps"]),
+    }
+
+
 def _complete_monitor(
     *,
     registry_path: Path,
@@ -302,6 +381,13 @@ def settle_governed_transition_proposals(
                 agent_id=agent_id,
                 proposal=proposal,
             )
+        elif kind == STEWARD_TEAM_PLAN_PREVIEW_KIND:
+            result = _apply_team_plan(
+                registry_path=Path(registry_path),
+                goal_id=goal_id,
+                agent_id=agent_id,
+                proposal=proposal,
+            )
         elif kind == "continuous_monitor_complete":
             result = _complete_monitor(
                 registry_path=Path(registry_path).expanduser(),
@@ -317,7 +403,11 @@ def settle_governed_transition_proposals(
             "proposal_id": proposal_id,
             "proposal_digest": proposal_digest,
             "kind": kind,
-            "monitor_key": str(proposal["monitor_key"]),
+            "monitor_key": (
+                str(proposal["monitor_key"])
+                if proposal.get("monitor_key") is not None
+                else None
+            ),
             "action": str(result["action"]),
             "todo_id": str(result["todo_id"]),
             "status": "committed",
@@ -330,7 +420,6 @@ def settle_governed_transition_proposals(
     return receipts
 
 
-STEWARD_TEAM_PLAN_PREVIEW_KIND = "steward_team_plan_preview"
 STEWARD_TEAM_PLAN_PREVIEW_SCHEMA_VERSION = "steward_team_plan_preview_v0"
 STEWARD_TEAM_PLAN_LANE_LIMIT = 8
 STEWARD_TEAM_PLAN_PRIORITIES = ("P0", "P1", "P2", "P3")

tests/test_steward_team_plan_apply.py

@@ -0,0 +1,182 @@
+"""A confirmed team plan creates its lanes' first Todos and nothing else."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from loopx.control_plane.work_items.governed_transition_proposal import (
+    GovernedTransitionSettlementPhase,
+    settle_governed_transition_proposals,
+)
+
+GOAL_ID = "team-plan-apply-fixture"
+AGENT_ID = "agent-alpha"
+
+
+def _fixture(tmp_path: Path) -> tuple[Path, Path]:
+    project = tmp_path / "project"
+    runtime = tmp_path / "runtime"
+    state_file = f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
+    state_path = project / state_file
+    state_path.parent.mkdir(parents=True, exist_ok=True)
+    state_path.write_text(
+        "---\n"
+        "status: active-read-only\n"
+        "owner_mode: goal\n"
+        'objective: "Stand up one digital team."\n'
+        "updated_at: 2026-01-01T00:00:00+00:00\n"
+        "---\n\n"
+        "# Team Plan Apply Fixture\n\n"
+        "## Next Action\n\n"
+        "- Confirm the team plan.\n\n"
+        "## Agent Todo\n\n",
+        encoding="utf-8",
+    )
+    registry_path = project / ".loopx" / "registry.json"
+    registry_path.parent.mkdir(parents=True, exist_ok=True)
+    registry_path.write_text(
+        json.dumps(
+            {
+                "schema_version": "0.1",
+                "updated_at": "2026-01-01T00:00:00+00:00",
+                "common_runtime_root": str(runtime),
+                "goals": [
+                    {
+                        "id": GOAL_ID,
+                        "domain": "team-plan-apply-fixture",
+                        "status": "active-read-only",
+                        "repo": str(project),
+                        "state_file": state_file,
+                        "adapter": {
+                            "kind": "read_only_project_map_v0",
+                            "status": "connected-read-only",
+                        },
+                        "coordination": {
+                            "registered_agents": [AGENT_ID],
+                            "agent_model": "peer_v1",
+                        },
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+    return project, registry_path
+
+
+def _proposal(*, agent_id: str = AGENT_ID, extra_lane: dict | None = None) -> dict:
+    lanes = [
+        {
+            "lane_id": "lane-alpha",
+            "agent_id": agent_id,
+            "acceptance": "The lane's first Todo is delivered with evidence",
+            "first_todo": {
+                "text": "Advance the intake contract",
+                "priority": "P1",
+                "task_class": "advancement_task",
+                "action_kind": "implement",
+            },
+        }
+    ]
+    if extra_lane is not None:
+        lanes.append(extra_lane)
+    return {
+        "schema_version": "steward_team_plan_preview_v0",
+        "kind": "steward_team_plan_preview",
+        "proposal_id": "proposal-team-plan",
+        "objective": "Stand up the intake lane",
+        "quota_envelope": {"slots_per_day": 4},
+        "stop_condition": "Stop when the owner withdraws the request",
+        "lanes": lanes,
+    }
+
+
+def _settle(registry_path: Path, proposal: dict) -> list[dict]:
+    return settle_governed_transition_proposals(
+        registry_path=registry_path,
+        goal_id=GOAL_ID,
+        agent_id=AGENT_ID,
+        effect_id="effect-team-plan",
+        proposals=[proposal],
+        existing_receipts=[],
+        checkpoint=lambda _receipts: None,
+        phase=GovernedTransitionSettlementPhase.PRE_SETTLEMENT,
+    )
+
+
+def _todos(project: Path) -> str:
+    return (project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md").read_text(
+        encoding="utf-8"
+    )
+
+
+def test_a_confirmed_plan_creates_each_ready_lane_first_todo(tmp_path: Path) -> None:
+    project, registry_path = _fixture(tmp_path)
+
+    receipts = _settle(registry_path, _proposal())
+
+    assert len(receipts) == 1
+    receipt = receipts[0]
+    assert receipt["kind"] == "steward_team_plan_preview"
+    assert receipt["status"] == "committed"
+    assert receipt["action"] == "created"
+    assert receipt["todo_id"].startswith("todo_")
+    state = _todos(project)
+    assert "Advance the intake contract" in state
+    assert f"claimed_by={AGENT_ID}" in state
+    # One lane, one Todo: no monitor rows, no extra work.
+    assert state.count("loopx:todo ") == 1
+
+
+def test_a_gap_lane_creates_nothing_and_a_replay_adds_no_second_row(
+    tmp_path: Path,
+) -> None:
+    project, registry_path = _fixture(tmp_path)
+    gap_lane = {
+        "lane_id": "lane-beta",
+        "agent_id": "agent-not-registered",
+        "acceptance": "Never reached",
+        "first_todo": {
+            "text": "Work that cannot be staffed",
+            "priority": "P1",
+            "task_class": "advancement_task",
+            "action_kind": "implement",
+        },
+    }
+
+    first = _settle(registry_path, _proposal(extra_lane=gap_lane))
+
+    assert first[0]["action"] == "created"
+    state = _todos(project)
+    assert "Work that cannot be staffed" not in state
+    assert state.count("loopx:todo ") == 1
+
+    # The canonical Todo owner decides reuse, so a replayed settlement does not
+    # duplicate the lane it already created.
+    replay = _settle(registry_path, _proposal(extra_lane=gap_lane))
+    assert replay[0]["action"] == "reused"
+    # The receipt still names the same lane Todo, and the gap lane stays absent.
+    assert replay[0]["todo_id"] == first[0]["todo_id"]
+    assert replay[0]["proposal_digest"] == first[0]["proposal_digest"]
+    assert _todos(project).count("loopx:todo ") == 1
+
+
+def test_an_unknown_goal_is_refused_before_any_todo(tmp_path: Path) -> None:
+    project, registry_path = _fixture(tmp_path)
+
+    with pytest.raises(ValueError, match="unknown Goal"):
+        settle_governed_transition_proposals(
+            registry_path=registry_path,
+            goal_id="goal-that-does-not-exist",
+            agent_id=AGENT_ID,
+            effect_id="effect-team-plan",
+            proposals=[_proposal()],
+            existing_receipts=[],
+            checkpoint=lambda _receipts: None,
+            phase=GovernedTransitionSettlementPhase.PRE_SETTLEMENT,
+        )
+
+    assert "loopx:todo " not in _todos(project)

tests/test_steward_team_plan_preview.py

@@ -6,6 +6,7 @@
 
 from loopx.control_plane.work_items.governed_transition_proposal import (
     _SETTLEMENT_PHASE_BY_PROPOSAL_KIND,
+    GovernedTransitionSettlementPhase,
     STEWARD_TEAM_PLAN_PREVIEW_KIND,
     STEWARD_TEAM_PLAN_PREVIEW_SCHEMA_VERSION,
     validate_steward_team_plan_preview,
@@ -115,10 +116,12 @@ def test_a_malformed_preview_fails_closed(mutation: dict, match: str) -> None:
         _validate(_plan(**mutation))
 
 
-def test_the_preview_kind_has_no_materializer() -> None:
-    """Nothing may apply the preview while only the preview slice exists."""
+def test_the_preview_kind_is_settled_only_at_pre_settlement() -> None:
+    """The kind applies through one phase of the canonical owner, and nowhere else."""
 
-    assert STEWARD_TEAM_PLAN_PREVIEW_KIND not in _SETTLEMENT_PHASE_BY_PROPOSAL_KIND
+    assert _SETTLEMENT_PHASE_BY_PROPOSAL_KIND[STEWARD_TEAM_PLAN_PREVIEW_KIND] is (
+        GovernedTransitionSettlementPhase.PRE_SETTLEMENT
+    )
 
 
 def test_the_chat_normalizer_admits_a_preview_only_with_host_facts() -> None: