feat(steward): let one owner confirmation apply a confirmed team plan

docs/architecture/rfcs/harness-selection-dsh-pi-v0.md

@@ -543,17 +543,23 @@ Shipped enforcement, in delivery order:
    Goal, an external manager channel resolves only the Goals it is bound to, and
    a Goal the registry does not know - or one outside that channel's scope -
    drops the preview instead of validating it against another Goal's Agents.
-
-What is still missing is the effect, not the preview: a confirmed plan has no
-Chat-side apply path yet, because the apply entry point today is a governed
-capability execution journal, so an owner's confirmation has nowhere to land,
-and a multi-lane preview has no frontend confirmation surface. A materialized
-lane Todo also does not yet carry the canonical intent revision it is meant to
-advance. The readback is no longer one of those gaps: the apply publishes every
-lane Todo it ensured under a bounded `lane_todo_ids` field, that field is the one
-additive exception to the closed, persisted receipt field set so a receipt
-written before it still validates, and a team-plan receipt carries no monitor key
-because a plan is not a monitor.
+5. **Confirmed apply from Chat.** The typed Chat action surface owns a
+   `team.plan` action. Its preview validates the plan against that Goal's
+   registered Agents and the host's advancement action kinds, and its apply
+   re-validates the same payload through the governed transition owner at
+   `PRE_SETTLEMENT`, so one owner confirmation creates each ready lane's first
+   bounded Todo and returns the lane readback. A registration change between
+   preview and apply makes the proposal stale rather than applying a plan whose
+   staffing has drifted.
+
+What is still missing is the surface that sends that confirmation and the
+traceability behind it: a multi-lane preview has no frontend confirmation
+surface yet, and a materialized lane Todo does not carry the canonical intent
+revision it is meant to advance. The readback is no longer one of those gaps: the
+apply publishes every lane Todo it ensured under a bounded `lane_todo_ids` field,
+that field is the one additive exception to the closed, persisted receipt field
+set so a receipt written before it still validates, and a team-plan receipt
+carries no monitor key because a plan is not a monitor.
 
 ### Relationship to the multi-agent and shared-authority contracts
 

docs/architecture/rfcs/harness-selection-dsh-pi-v0.zh-CN.md

@@ -423,11 +423,15 @@ Todo 创建、quota 或 goal policy——复用预览点名的身份，不得扩
    预览只会用**它点名那个 Goal** 的 Agent 来校验：业主自己的通道可解析任意已注册 Goal，
    外部管家通道只解析它被绑定的 Goal，而 registry 不认识的 Goal——或超出该通道范围的
    Goal——会让预览被丢弃，而不是拿另一个 Goal 的 Agent 去校验它。
-
-仍然缺的是**效果**而不是预览：被确认的计划还没有 Chat 侧的落地路径——今天的落地入口是受治理
-能力执行 journal——所以业主的确认暂时无处落地；多 lane 预览也还没有前端确认面；另外，建出
-的 lane Todo 还没有携带它本应推进的规范意图修订。回读本身已经不再是缺口：落地会把这次确保的
-每一条 lane Todo 以有界字段 `lane_todo_ids` 发布出去；该字段是那个封闭且持久化的回执字段集的
+5. **从 Chat 确认落地。** 类型化 Chat action 面拥有一个 `team.plan` 动作：它的预览用该 Goal
+   已注册 Agent 与本机 advancement action kind 校验计划，它的落地则把同一份载荷交给受治理
+   提案所有者在 `PRE_SETTLEMENT` 相位重新校验，因此**一次业主确认**就会为每条 ready lane
+   建出首个有界 Todo 并返回 lane 回读。预览与落地之间若发生注册变化，提案会变为 stale，而
+   不是把 staffing 已经漂移的计划落地。
+
+仍然缺的是**发出这次确认的表面**与它背后的可追溯性：多 lane 预览还没有前端确认面；建出的
+lane Todo 也还没有携带它本应推进的规范意图修订。回读本身已经不再是缺口：落地会把这次确保的每
+一条 lane Todo 以有界字段 `lane_todo_ids` 发布出去；该字段是那个封闭且持久化的回执字段集的
 **唯一**加性例外，因此早前写下的回执仍然通过校验，而团队计划回执不带 monitor key——计划不是
 monitor。
 

loopx/chat_action_normalization.py

@@ -507,6 +507,41 @@ def _normalize(
             if operation == "edit" and len(result) == 3:
                 raise ValueError("heartbeat edit requires a configuration change")
             return result
+        if action_kind == "team.plan":
+            from .control_plane.todos.contract import (
+                TODO_ACTION_KIND_ADVANCEMENT_VALUES,
+            )
+            from .control_plane.work_items.governed_transition_proposal import (
+                validate_steward_team_plan_preview,
+            )
+
+            values = self._allowed_parameters(
+                parameters,
+                allowed={"goal_id", "plan", "requested_by"},
+            )
+            goal_id = _opaque(values.get("goal_id"), field="goal_id")
+            goal = self._goal(goal_id)
+            plan = values.get("plan")
+            if not isinstance(plan, Mapping):
+                raise ValueError("team.plan requires the validated plan object")
+            if str(plan.get("goal_id") or "") != goal_id:
+                raise ValueError("team.plan Goal must match the plan's own Goal")
+            # The plan is validated here against this Goal's registered Agents
+            # and the host's shipped action kinds, and the apply re-validates the
+            # same payload with the host's own facts before it creates anything,
+            # so the stored parameters are never the thing that authorizes work.
+            validate_steward_team_plan_preview(
+                plan,
+                registered_agent_ids=registered_agent_ids_for_goal(goal),
+                supported_action_kinds=sorted(TODO_ACTION_KIND_ADVANCEMENT_VALUES),
+            )
+            return {
+                "goal_id": goal_id,
+                "plan": dict(plan),
+                "requested_by": _opaque(
+                    values.get("requested_by") or "owner", field="requested_by"
+                ),
+            }
         if action_kind == "monitor.create":
             values = self._allowed_parameters(
                 parameters,

loopx/chat_action_store.py

@@ -32,6 +32,7 @@
     "gate.resolve",
     "run.correct",
     "operation.execute",
+    "team.plan",
 }
 PROPOSAL_STATES = {
     "preview_ready",

loopx/chat_actions.py

@@ -42,6 +42,7 @@
     "monitor.update",
     "gate.resolve",
     "operation.execute",
+    "team.plan",
 }
 _OPAQUE_ID = re.compile(r"^[A-Za-z0-9._:-]{1,200}$")
 # Runtime Endpoint ids and durable Goal agent ids are chosen independently, so
@@ -929,6 +930,68 @@ def _apply_monitor_create(
         )
         return {"proposal": stored, "turn": None}
 
+    def _apply_team_plan(
+        self, proposal_id: str, proposal: dict[str, Any], parameters: dict[str, Any]
+    ) -> dict[str, Any]:
+        """Create each ready lane's first bounded Todo through the Todo owner."""
+
+        from .control_plane.work_items.governed_transition_proposal import (
+            GovernedTransitionSettlementPhase,
+            settle_governed_transition_proposals,
+        )
+
+        current_fingerprint = self._registry_fingerprint()
+        if current_fingerprint != proposal.get("expected_state_fingerprint"):
+            stale = self.store.apply(
+                proposal_id,
+                current_state_fingerprint=current_fingerprint,
+                receipt={},
+            )
+            return {"proposal": stale, "turn": None}
+        goal_id = str(parameters["goal_id"])
+        plan = parameters.get("plan")
+        if not isinstance(plan, Mapping):
+            raise ValueError("team plan proposal is malformed")
+        # The governed transition owner re-validates the plan with the host's own
+        # facts and owns the settlement phase, so this action never becomes a
+        # second writer of lanes.
+        settlements = settle_governed_transition_proposals(
+            registry_path=self.registry_path,
+            goal_id=goal_id,
+            agent_id=str(parameters.get("requested_by") or "owner"),
+            effect_id=proposal_id,
+            proposals=[{**dict(plan), "proposal_id": proposal_id}],
+            existing_receipts=[],
+            checkpoint=lambda _receipts: None,
+            phase=GovernedTransitionSettlementPhase.PRE_SETTLEMENT,
+        )
+        settlement = settlements[0]
+        lane_todo_ids = [str(item) for item in (settlement.get("lane_todo_ids") or [])]
+        receipt = {
+            "receipt_id": _digest(
+                {
+                    "proposal_id": proposal_id,
+                    "goal_id": goal_id,
+                    "lane_todo_ids": lane_todo_ids,
+                }
+            )[:32],
+            "outcome": (
+                "team_plan_applied"
+                if settlement.get("action") == "created"
+                else "team_plan_lanes_already_present"
+            ),
+            "projection_verified": True,
+            "resource_ids": {
+                "goal_id": goal_id,
+                "todo_id": str(settlement.get("todo_id") or ""),
+                "lane_todo_ids": lane_todo_ids,
+            },
+        }
+        stored = self.store.apply(
+            proposal_id, current_state_fingerprint=current_fingerprint, receipt=receipt
+        )
+        return {"proposal": stored, "turn": None}
+
     def preview(self, request: Mapping[str, Any]) -> dict[str, Any]:
         unknown = set(request) - {
             "action_kind",
@@ -980,6 +1043,15 @@ def preview(self, request: Mapping[str, Any]) -> dict[str, Any]:
             )
             evidence = ["The recoverable Goal and Agent Chat Session is available."]
             permission = "scoped_correction"
+        elif action_kind == "team.plan":
+            # A plan staffs registered Agents, so the registration facts it was
+            # validated against are the state that can make this preview stale.
+            fingerprint = self._registry_fingerprint()
+            evidence = [
+                "The plan was validated against this Goal's registered Agents and the host's advancement action kinds.",
+                "Applying it creates the first bounded Todo of each ready lane, through the canonical Todo owner.",
+            ]
+            permission = "durable_write"
         elif action_kind in {"todo.update", "monitor.update"}:
             if action_kind == "todo.update":
                 canonical_preview = self._run_todo_update(normalized, dry_run=True)
@@ -1147,6 +1219,8 @@ def apply(self, proposal_id: str) -> dict[str, Any]:
             raise self._heartbeat_gate(parameters)
         if action_kind == "monitor.create":
             return self._apply_monitor_create(proposal_id, proposal, parameters)
+        if action_kind == "team.plan":
+            return self._apply_team_plan(proposal_id, proposal, parameters)
         if action_kind == "todo.update":
             return self._apply_todo_update(proposal_id, proposal, parameters)
         if action_kind == "monitor.update":

tests/test_chat_team_plan_action.py

@@ -0,0 +1,163 @@
+"""A confirmed team plan applies through the Chat action service."""
+
+from __future__ import annotations
+
+import json
+import itertools
+from pathlib import Path
+
+import pytest
+
+from loopx.chat_action_store import ChatActionStore
+from loopx.chat_actions import ChatActionService
+
+GOAL_ID = "team-plan-action-fixture"
+AGENT_ID = "agent-alpha"
+_PREVIEWS = itertools.count(1)
+
+
+def _fixture(tmp_path: Path, *, agents: tuple[str, ...] = (AGENT_ID,)):
+    project = tmp_path / "project"
+    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 Action 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",
+                "goals": [
+                    {
+                        "id": GOAL_ID,
+                        "domain": GOAL_ID,
+                        "status": "active-read-only",
+                        "repo": str(project),
+                        "state_file": state_file,
+                        "coordination": {
+                            "registered_agents": list(agents),
+                            "agent_model": "peer_v1",
+                        },
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+    service = ChatActionService(
+        store=ChatActionStore(tmp_path / "runtime" / "chat" / "actions"),
+        registry_path=registry_path,
+    )
+    return project, registry_path, service
+
+
+def _plan(*, agent_id: str = AGENT_ID, goal_id: str = GOAL_ID) -> dict:
+    return {
+        "schema_version": "steward_team_plan_preview_v0",
+        "kind": "steward_team_plan_preview",
+        "goal_id": goal_id,
+        "objective": "Stand up the intake lane",
+        "quota_envelope": {"slots_per_day": 4},
+        "stop_condition": "Stop when the owner withdraws the request",
+        "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",
+                },
+            }
+        ],
+    }
+
+
+def _preview(service: ChatActionService, plan: dict | None = None) -> dict:
+    return service.preview(
+        {
+            "action_kind": "team.plan",
+            "summary": "Confirm the team plan",
+            "normalized_parameters": {"goal_id": GOAL_ID, "plan": plan or _plan()},
+            "context": {},
+            "idempotency_key": f"team-plan-preview-{next(_PREVIEWS)}",
+        }
+    )
+
+
+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, service = _fixture(tmp_path)
+
+    preview = _preview(service)
+    assert preview["action_kind"] == "team.plan"
+    assert preview["permission_classification"] == "durable_write"
+
+    applied = service.apply(preview["proposal_id"])
+    proposal = applied["proposal"]
+    assert proposal["status"] == "applied"
+    receipt = proposal["receipt"]
+    assert receipt["outcome"] == "team_plan_applied"
+    lane_todo_ids = receipt["resource_ids"]["lane_todo_ids"]
+    assert len(lane_todo_ids) == 1 and lane_todo_ids[0].startswith("todo_")
+    assert receipt["resource_ids"]["todo_id"] == lane_todo_ids[0]
+    state = _todos(project)
+    assert "Advance the intake contract" in state
+    assert f"claimed_by={AGENT_ID}" in state
+    assert state.count("loopx:todo ") == 1
+
+
+def test_a_lane_with_an_unregistered_agent_becomes_a_gap_and_creates_nothing(
+    tmp_path: Path,
+) -> None:
+    project, _registry_path, service = _fixture(tmp_path)
+
+    preview = _preview(service, _plan(agent_id="agent-not-registered"))
+    applied = service.apply(preview["proposal_id"])
+    # The preview is admitted with its gap, and confirming it creates nothing:
+    # the owner sees what was asked for and what is missing.
+    assert applied["proposal"]["receipt"]["resource_ids"]["lane_todo_ids"] == []
+    assert "loopx:todo " not in _todos(project)
+
+
+def test_a_plan_for_another_goal_is_refused_at_preview(tmp_path: Path) -> None:
+    _project, _registry_path, service = _fixture(tmp_path)
+
+    with pytest.raises(ValueError, match="must match the plan's own Goal"):
+        _preview(service, _plan(goal_id="some-other-goal"))
+
+
+def test_a_changed_registry_makes_the_confirmed_plan_stale(tmp_path: Path) -> None:
+    project, registry_path, service = _fixture(tmp_path, agents=(AGENT_ID,))
+
+    preview = _preview(service)
+    registry = json.loads(registry_path.read_text(encoding="utf-8"))
+    registry["goals"][0]["coordination"]["registered_agents"] = [AGENT_ID, "agent-beta"]
+    registry_path.write_text(json.dumps(registry), encoding="utf-8")
+
+    applied = service.apply(preview["proposal_id"])
+    # The Agents a plan was validated against are the state that can invalidate
+    # it, so a registration change asks the owner to confirm the current plan.
+    assert applied["proposal"]["status"] == "stale"
+    assert "loopx:todo " not in _todos(project)