feat(steward): validate a team preview without letting it apply

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

@@ -474,6 +474,57 @@ Validation: `tests/capabilities/test_steward_executor_machine_defaults.py`,
 `tests/capabilities/test_capability_configuration_ui.py`, and
 `examples/loopx-steward-channel-binding-smoke.py`.
 
+## Steward Team Intake (planned, 2026-09-16)
+
+The steward answers questions today, and since `2026-09-16` its shipped
+guidance carries one bounded procedure for a different request: one owner
+sentence that asks for a *team* rather than a task. That procedure is guidance,
+not machine enforcement, so this section records where the enforced contract
+belongs and what it must validate before any implementation lands.
+
+The intake boundary is the canonical governed-proposal owner
+(`loopx/control_plane/work_items/governed_transition_proposal.py`), not a new
+CLI command and not a new capability. That owner already dispatches proposals
+by kind, publishes a typed receipt with a proposal digest, and the Chat Turn
+already projects `response.proposals` into `proposal.ready` events. A team
+request is therefore one proposal of a new kind, not a parallel intake path
+beside the existing one. A command with no second caller, and a builder module
+with no caller at all, both stay out: this repository keeps an uncalled
+abstraction in design state until its call site exists.
+
+The proposal payload is validated before anything may be applied, and it names:
+
+- each lane and the Agent that runs it, resolved from Agents Core already
+  registers for the Goal;
+- that lane's first bounded Todo, with its declared priority, task class and
+  action kind;
+- the quota or cadence envelope that bounds the lanes;
+- the acceptance signal that ends each lane;
+- the stop condition that ends the team.
+
+A requested lane that cannot be staffed is a typed gap naming the missing
+registration or grant; it is never filled in by inventing an Agent, a Todo
+capability, or a lane the machine cannot run. The plan is a preview: it creates
+no Todo, registers no Agent, sets no quota, and spends none, and an owner's
+confirmation of that exact preview is the only thing that admits an apply.
+Apply routes to the canonical owners each effect already has -- Agent
+registration, Todo creation, quota or goal policy -- reuses the identities the
+preview named, and returns one readback of what exists. It may not widen the
+confirmed scope, and a team plan is never settled as if the work were done.
+
+Delivery is two slices, in this order:
+
+1. **Preview slice (next).** The typed payload contract and its validator, with
+   focused tests, and no materializer registered, so a preview cannot apply
+   even by mistake.
+2. **Apply slice.** A materializer for that kind, with its settlement phase and
+   readback, routed through the owners above.
+
+What this planned contract does not authorize: the steward still only proposes
+and delegates; selecting a steward executor or storing a credential grants none
+of these effects; and nothing here widens OS, provider, audience or work-state
+authority.
+
 ## Steward Channel Readiness by Milestone (2026-09-15)
 
 The steward channel consumes both this document's host selection and the manager

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

@@ -373,6 +373,42 @@ journal 与配额语义；B 作为上游接口出现时的低成本替代；只
 `tests/capabilities/test_capability_configuration_ui.py`，以及
 `examples/loopx-steward-channel-binding-smoke.py`。
 
+## 管家团队入端口径（规划中，2026-09-16）
+
+管家今天回答问题；自 2026-09-16 起，它的出厂指引里已带一段有界流程，用于另一类请求：
+业主一句话要的是**团队**而不是单个任务。那段流程是指引，不是机器强制，因此本节记录被强制
+的契约属于哪里、以及实现落地前必须校验什么。
+
+入端口径是既有的受治理提案所有者
+（`loopx/control_plane/work_items/governed_transition_proposal.py`），不是新的 CLI 命令，
+也不是新的能力。该所有者本就按 kind 分派提案、产出带 proposal digest 的类型化回执，而
+Chat Turn 也早已把 `response.proposals` 投影成 `proposal.ready` 事件。因此"团队请求"是
+**一种新 kind 的提案**，而不是在既有路径旁再开一条入端口。没有第二个调用方的命令、以及
+完全没有调用方的 builder 模块都不新增：本仓库要求未获调用的抽象先留在设计态。
+
+在允许任何落地之前先校验提案载荷，它必须点名：
+
+- 每条 lane 及其运行的 Agent，且只能来自 Core 已为该 Goal 注册的 Agent；
+- 该 lane 的首个有界 Todo，含其声明优先级、task class 与 action kind；
+- 约束这些 lane 的 quota 或节奏包络；
+- 结束每条 lane 的验收信号；
+- 结束整个团队的终止条件。
+
+配不齐的 lane 是类型化的 gap（缺哪个注册或授予），不允许靠编造 Agent、Todo 能力或本机跑不动
+的 lane 来填。计划是**预览**：不建 Todo、不注册 Agent、不设 quota、不扣额度；只有业主对这
+份确切预览的确认，才允许进入落地。落地只经各 effect 既有的 canonical owner——Agent 注册、
+Todo 创建、quota 或 goal policy——复用预览点名的身份，并返回一份"现在存在什么"的回读；不得
+扩大已确认范围，也不得把团队计划当作工作已完成的结算。
+
+按此顺序分两片交付：
+
+1. **预览片（下一步）**：类型化载荷契约与其校验器，配聚焦测试，且不注册 materializer，
+   使预览即使被误用也无法落地。
+2. **落地片**：该 kind 的 materializer，含其结算相位与回读，并按上文经既有 owner 路由。
+
+这条规划契约不授权什么：管家仍然只提议与委托；选择管家执行器或存凭据都不带来这些 effect；
+这里也不会扩大 OS、provider、受众或工作状态权限。
+
 ## 按里程碑看管家通道的就绪度（2026-09-15）
 
 管家通道同时消费本文的宿主选型与

loopx/control_plane/work_items/governed_transition_proposal.py

@@ -328,3 +328,141 @@ def settle_governed_transition_proposals(
         by_proposal_id[proposal_id] = receipt
         checkpoint(receipts)
     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")
+STEWARD_TEAM_PLAN_GAP_REASONS = (
+    "agent_not_registered",
+    "capability_not_granted",
+    "audience_not_authorized",
+)
+
+
+def _plan_text(value: object, label: str) -> str:
+    text = " ".join(str(value or "").split())
+    if not text:
+        raise ValueError(f"{label} must be a non-empty string")
+    if len(text) > 600:
+        raise ValueError(f"{label} exceeds the public-safe preview length")
+    validate_public_safe_value({"value": text}, path=label)
+    return text
+
+
+def validate_steward_team_plan_preview(
+    payload: object,
+    *,
+    registered_agent_ids: Sequence[str],
+    supported_action_kinds: Sequence[str],
+) -> dict[str, Any]:
+    """Validate one steward team preview, and refuse to invent its staffing.
+
+    The preview is the whole effect of this kind: it creates nothing, so it has
+    no materializer and no settlement phase. A lane whose Agent this Goal does
+    not register becomes a typed gap that keeps the work it did *not* staff
+    under ``declined_first_todo``, so the owner sees what was asked for and what
+    is missing instead of a lane that was quietly filled in or dropped.
+    """
+
+    plan = _mapping(payload, "steward_team_plan_preview")
+    if plan.get("schema_version") != STEWARD_TEAM_PLAN_PREVIEW_SCHEMA_VERSION:
+        raise ValueError("steward team plan preview schema_version is invalid")
+    if plan.get("kind") != STEWARD_TEAM_PLAN_PREVIEW_KIND:
+        raise ValueError("steward team plan preview kind is invalid")
+    registered = {str(value) for value in registered_agent_ids}
+    action_kinds = {str(value) for value in supported_action_kinds}
+    lanes_value = plan.get("lanes")
+    if not isinstance(lanes_value, Sequence) or isinstance(lanes_value, (str, bytes)):
+        raise ValueError("steward team plan preview requires a lane list")
+    if not 1 <= len(lanes_value) <= STEWARD_TEAM_PLAN_LANE_LIMIT:
+        raise ValueError(
+            f"steward team plan preview requires 1..{STEWARD_TEAM_PLAN_LANE_LIMIT} lanes"
+        )
+    lanes: list[dict[str, Any]] = []
+    gaps: list[dict[str, str]] = []
+    seen_lanes: set[str] = set()
+    for raw_lane in lanes_value:
+        lane = _mapping(raw_lane, "steward_team_plan_lane")
+        lane_id = _plan_text(lane.get("lane_id"), "lane_id")
+        if lane_id in seen_lanes:
+            raise ValueError("steward team plan preview repeats a lane_id")
+        seen_lanes.add(lane_id)
+        agent_id = _plan_text(lane.get("agent_id"), "agent_id")
+        acceptance = _plan_text(lane.get("acceptance"), "lane acceptance")
+        declared_gap = lane.get("staffing_gap")
+        if declared_gap is not None:
+            gap = _mapping(declared_gap, "staffing_gap")
+            reason_code = str(gap.get("reason_code") or "")
+            if reason_code not in STEWARD_TEAM_PLAN_GAP_REASONS:
+                raise ValueError("staffing_gap reason_code is invalid")
+            if lane.get("first_todo") is not None:
+                raise ValueError("a lane that declares a gap may not declare work")
+            normalized = {
+                "lane_id": lane_id,
+                "agent_id": agent_id,
+                "acceptance": acceptance,
+                "staffing": "gap",
+                "gap_reason_code": reason_code,
+                "gap_note": _plan_text(gap.get("note"), "staffing_gap note"),
+            }
+            lanes.append(normalized)
+            gaps.append({"lane_id": lane_id, "reason_code": reason_code})
+            continue
+        first_todo = _mapping(lane.get("first_todo"), "first_todo")
+        text = _plan_text(first_todo.get("text"), "first_todo text")
+        priority = str(first_todo.get("priority") or "")
+        if priority not in STEWARD_TEAM_PLAN_PRIORITIES:
+            raise ValueError("first_todo priority is invalid")
+        task_class = str(first_todo.get("task_class") or "")
+        if task_class != "advancement_task":
+            raise ValueError("a lane's first bounded Todo must be an advancement_task")
+        action_kind = str(first_todo.get("action_kind") or "")
+        if action_kind not in action_kinds:
+            raise ValueError("first_todo action_kind is not supported by this host")
+        normalized_todo = {
+            "text": text,
+            "priority": priority,
+            "task_class": task_class,
+            "action_kind": action_kind,
+        }
+        if agent_id not in registered:
+            lane_result = {
+                "lane_id": lane_id,
+                "agent_id": agent_id,
+                "acceptance": acceptance,
+                "staffing": "gap",
+                "gap_reason_code": "agent_not_registered",
+                "declined_first_todo": normalized_todo,
+            }
+            lanes.append(lane_result)
+            gaps.append({"lane_id": lane_id, "reason_code": "agent_not_registered"})
+            continue
+        lanes.append(
+            {
+                "lane_id": lane_id,
+                "agent_id": agent_id,
+                "acceptance": acceptance,
+                "staffing": "ready",
+                "first_todo": normalized_todo,
+            }
+        )
+    envelope = _mapping(plan.get("quota_envelope"), "quota_envelope")
+    if not envelope:
+        raise ValueError("steward team plan preview requires a quota envelope")
+    validate_public_safe_value(envelope, path="quota_envelope")
+    preview = {
+        "schema_version": STEWARD_TEAM_PLAN_PREVIEW_SCHEMA_VERSION,
+        "kind": STEWARD_TEAM_PLAN_PREVIEW_KIND,
+        "objective": _plan_text(plan.get("objective"), "objective"),
+        "lanes": lanes,
+        "gaps": gaps,
+        "quota_envelope": dict(envelope),
+        "stop_condition": _plan_text(plan.get("stop_condition"), "stop_condition"),
+        # A preview is never an effect: the contract states it, so a reader does
+        # not have to know which materializers happen to be registered.
+        "applies": False,
+    }
+    validate_public_safe_value(preview, path="steward_team_plan_preview")
+    return preview

tests/test_steward_team_plan_preview.py

@@ -0,0 +1,121 @@
+"""The steward team preview validates without ever becoming an effect."""
+
+from __future__ import annotations
+
+import pytest
+
+from loopx.control_plane.work_items.governed_transition_proposal import (
+    _SETTLEMENT_PHASE_BY_PROPOSAL_KIND,
+    STEWARD_TEAM_PLAN_PREVIEW_KIND,
+    STEWARD_TEAM_PLAN_PREVIEW_SCHEMA_VERSION,
+    validate_steward_team_plan_preview,
+)
+
+
+def _plan(**overrides: object) -> dict[str, object]:
+    plan: dict[str, object] = {
+        "schema_version": STEWARD_TEAM_PLAN_PREVIEW_SCHEMA_VERSION,
+        "kind": STEWARD_TEAM_PLAN_PREVIEW_KIND,
+        "objective": "Ship 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-alpha",
+                "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",
+                },
+            }
+        ],
+    }
+    plan.update(overrides)
+    return plan
+
+
+def _validate(plan: object) -> dict:
+    return validate_steward_team_plan_preview(
+        plan,
+        registered_agent_ids=["agent-alpha"],
+        supported_action_kinds=["implement"],
+    )
+
+
+def test_a_staffed_lane_becomes_a_preview_that_cannot_apply() -> None:
+    preview = _validate(_plan())
+
+    assert preview["applies"] is False
+    assert preview["gaps"] == []
+    assert preview["lanes"][0]["staffing"] == "ready"
+    assert preview["lanes"][0]["first_todo"]["priority"] == "P1"
+    assert preview["quota_envelope"] == {"slots_per_day": 4}
+
+
+def test_an_unregistered_agent_becomes_a_gap_instead_of_being_invented() -> None:
+    plan = _plan()
+    plan["lanes"][0]["agent_id"] = "agent-not-registered"  # type: ignore[index]
+
+    preview = _validate(plan)
+
+    lane = preview["lanes"][0]
+    assert lane["staffing"] == "gap"
+    assert lane["gap_reason_code"] == "agent_not_registered"
+    # The work the owner asked for is kept with the gap, never silently dropped.
+    assert lane["declined_first_todo"]["text"] == "Advance the intake contract"
+    assert "first_todo" not in lane
+    assert preview["gaps"] == [
+        {"lane_id": "lane-alpha", "reason_code": "agent_not_registered"}
+    ]
+    assert preview["applies"] is False
+
+
+def test_a_declared_gap_keeps_its_reason_and_carries_no_work() -> None:
+    plan = _plan()
+    lane = plan["lanes"][0]  # type: ignore[index]
+    lane.pop("first_todo")
+    lane["staffing_gap"] = {
+        "reason_code": "capability_not_granted",
+        "note": "The reviewer capability is not granted on this machine",
+    }
+
+    preview = _validate(plan)
+
+    assert preview["gaps"] == [
+        {"lane_id": "lane-alpha", "reason_code": "capability_not_granted"}
+    ]
+    with pytest.raises(ValueError, match="may not declare work"):
+        _validate(
+            _plan(
+                lanes=[
+                    {
+                        **lane,
+                        "first_todo": _plan()["lanes"][0]["first_todo"],  # type: ignore[index]
+                    }
+                ]
+            )
+        )
+
+
+@pytest.mark.parametrize(
+    "mutation,match",
+    [
+        ({"kind": "something_else"}, "kind is invalid"),
+        ({"schema_version": "v0"}, "schema_version is invalid"),
+        ({"lanes": []}, "requires 1..8 lanes"),
+        ({"stop_condition": ""}, "stop_condition must be a non-empty string"),
+        ({"quota_envelope": {}}, "requires a quota envelope"),
+    ],
+)
+def test_a_malformed_preview_fails_closed(mutation: dict, match: str) -> None:
+    with pytest.raises(ValueError, match=match):
+        _validate(_plan(**mutation))
+
+
+def test_the_preview_kind_has_no_materializer() -> None:
+    """Nothing may apply the preview while only the preview slice exists."""
+
+    assert STEWARD_TEAM_PLAN_PREVIEW_KIND not in _SETTLEMENT_PHASE_BY_PROPOSAL_KIND