feat(steward): record the intent basis a materialized team plan advances

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

@@ -555,13 +555,16 @@ Shipped enforcement, in delivery order:
    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.
+per-Goal coverage behind it: a multi-lane preview has no frontend confirmation
+surface yet. Traceability is recorded rather than implied: the settlement reads
+the Goal's canonical source basis before it writes, and the receipt carries it as
+a bounded `intent_basis`, so each lane Todo can be tied to the revision it was
+meant to advance even though the Todo row itself does not carry the field. The
+readback is no longer a gap either: the apply publishes every lane Todo it ensured
+under a bounded `lane_todo_ids` field, both fields are additive exceptions to the
+closed, persisted receipt field set so a receipt written before them 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

@@ -430,11 +430,12 @@ Todo 创建、quota 或 goal policy——复用预览点名的身份，不得扩
    建出首个有界 Todo 并返回 lane 回读。预览与落地之间若发生注册变化，提案会变为 stale，而
    不是把 staffing 已经漂移的计划落地。
 
-仍然缺的是**发出这次确认的表面**与它背后的可追溯性：多 lane 预览还没有前端确认面；建出的
-lane Todo 也还没有携带它本应推进的规范意图修订。回读本身已经不再是缺口：落地会把这次确保的每
-一条 lane Todo 以有界字段 `lane_todo_ids` 发布出去；该字段是那个封闭且持久化的回执字段集的
-**唯一**加性例外，因此早前写下的回执仍然通过校验，而团队计划回执不带 monitor key——计划不是
-monitor。
+仍然缺的是**发出这次确认的表面**：多 lane 预览还没有前端确认面。可追溯性已经被记录而不是被
+暗示：结算在写入**之前**读取该 Goal 的规范 source basis，回执以有界字段 `intent_basis`
+携带它，因此每条 lane Todo 都能被追溯回它本应推进的那个修订——尽管 Todo 行本身还不携带该
+字段。回读也不再是缺口：落地会把这次确保的每一条 lane Todo 以有界字段 `lane_todo_ids` 发布
+出去；这两个字段都是那个封闭且持久化的回执字段集的加性例外，因此早前写下的回执仍然通过校验，
+而团队计划回执不带 monitor key——计划不是 monitor。
 
 ### 与 multi-agent / shared authority 契约的关系
 

loopx/chat_actions.py

@@ -967,6 +967,7 @@ def _apply_team_plan(
         )
         settlement = settlements[0]
         lane_todo_ids = [str(item) for item in (settlement.get("lane_todo_ids") or [])]
+        intent_basis = str(settlement.get("intent_basis") or "")
         receipt = {
             "receipt_id": _digest(
                 {
@@ -987,6 +988,10 @@ def _apply_team_plan(
                 "lane_todo_ids": lane_todo_ids,
             },
         }
+        if intent_basis:
+            # The canonical revision these lanes were created against, so the
+            # owner's readback can name what the work advances.
+            receipt["intent_basis"] = intent_basis
         stored = self.store.apply(
             proposal_id, current_state_fingerprint=current_fingerprint, receipt=receipt
         )

loopx/control_plane/work_items/governed_transition_proposal.py

@@ -44,9 +44,10 @@
 # closed and a new field is admitted only as an explicitly bounded addition
 # that an older receipt may still omit. `lane_todo_ids` is the readback of a
 # team plan: every lane Todo the settlement ensured, not just the first one.
-_OPTIONAL_RECEIPT_FIELDS = {"lane_todo_ids"}
+_OPTIONAL_RECEIPT_FIELDS = {"lane_todo_ids", "intent_basis"}
 _LANE_TODO_ID_LIMIT = 8
 _LANE_TODO_ID = re.compile(r"^todo_[A-Za-z0-9]{1,40}$")
+_INTENT_BASIS = re.compile(r"^sha256:[0-9a-f]{64}$")
 
 
 TransitionCheckpoint = Callable[[list[dict[str, Any]]], None]
@@ -152,6 +153,14 @@ def validate_governed_transition_receipts(
             raise ValueError(
                 "governed transition proposal receipt lane_todo_ids is invalid"
             )
+        intent_basis = receipt.get("intent_basis")
+        if intent_basis is not None and (
+            not isinstance(intent_basis, str)
+            or not _INTENT_BASIS.fullmatch(intent_basis)
+        ):
+            raise ValueError(
+                "governed transition proposal receipt intent_basis is invalid"
+            )
         validate_public_safe_value(receipt, path=f"transition_receipts[{index}]")
         receipts.append(receipt)
     return receipts
@@ -254,6 +263,48 @@ def _upsert_monitor(
     }
 
 
+def _intent_basis_for(
+    *,
+    goal_id: str,
+    goal: Mapping[str, Any],
+    registry_path: Path,
+    preview: Mapping[str, Any],
+) -> str | None:
+    """Read the canonical source basis one work-graph edit is applied against.
+
+    The source basis is a Goal-level fact, so any of the Goal's Agents reads the
+    same one; a ready lane is preferred because that is where the work will live.
+    A Goal whose basis cannot be read omits the field rather than inventing one.
+    """
+
+    lanes = preview.get("lanes") or []
+    basis_agent = next(
+        (
+            str(lane.get("agent_id"))
+            for lane in lanes
+            if lane.get("staffing") == "ready"
+        ),
+        str(lanes[0].get("agent_id")) if lanes else "",
+    )
+    if not basis_agent:
+        return None
+    try:
+        from ...control_plane.goals.shared_goal_alignment import (
+            project_shared_goal_alignment,
+        )
+
+        alignment = project_shared_goal_alignment(
+            goal_id=goal_id,
+            agent_id=basis_agent,
+            project=Path(str(goal.get("repo") or ".")).expanduser(),
+            registry_path=Path(registry_path),
+        )
+    except (OSError, ValueError, TypeError, KeyError, RuntimeError):
+        return None
+    basis = (alignment.get("source_basis") or {}).get("source_basis_digest")
+    return str(basis) if basis else None
+
+
 def _apply_team_plan(
     *,
     registry_path: Path,
@@ -299,6 +350,11 @@ def _apply_team_plan(
         registered_agent_ids=registered_agent_ids_for_goal(goal),
         supported_action_kinds=sorted(TODO_ACTION_KIND_ADVANCEMENT_VALUES),
     )
+    # Traceability is read before the edit: the receipt names the canonical
+    # basis this work-graph edit was applied against, so the lanes could not
+    # make the basis describe their own creation. A Goal whose basis cannot be
+    # read omits the field instead of inventing one.
+    intent_basis = _intent_basis_for(goal_id=goal_id, goal=goal, registry_path=registry_path, preview=preview)
     created: list[str] = []
     reused: list[str] = []
     for lane in preview["lanes"]:
@@ -330,6 +386,7 @@ def _apply_team_plan(
         "target_key": None,
         "created_todo_ids": created,
         "lane_todo_ids": lane_todo_ids,
+        "intent_basis": intent_basis,
         "reused_lane_count": len(reused),
         "gap_count": len(preview["gaps"]),
     }
@@ -458,6 +515,11 @@ def settle_governed_transition_proposals(
             # The apply ensured every ready lane's first Todo; a receipt that
             # named only the first one could not be read as "what exists now".
             receipt["lane_todo_ids"] = [str(item) for item in lane_todo_ids]
+        if result.get("intent_basis"):
+            # The work-graph edit this receipt records is traceable to the
+            # canonical basis it was applied against, so a lane Todo can be tied
+            # back to the intent revision it was meant to advance.
+            receipt["intent_basis"] = str(result["intent_basis"])
         validate_public_safe_value(receipt, path="transition_receipt")
         receipts.append(receipt)
         by_proposal_id[proposal_id] = receipt

tests/test_chat_team_plan_action.py

@@ -122,6 +122,9 @@ def test_a_confirmed_plan_creates_each_ready_lane_first_todo(tmp_path: Path) ->
     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]
+    # The readback names the canonical revision these lanes were created
+    # against, so the owner sees what the new work is meant to advance.
+    assert receipt["intent_basis"].startswith("sha256:")
     state = _todos(project)
     assert "Advance the intake contract" in state
     assert f"claimed_by={AGENT_ID}" in state

tests/test_steward_team_plan_apply.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import re
 from pathlib import Path
 
 import pytest
@@ -288,6 +289,46 @@ def test_the_lane_readback_is_optional_bounded_and_additive() -> None:
         validate_governed_transition_receipts([_receipt(monitor_key=None)])
 
 
+def test_the_receipt_records_the_intent_basis_it_was_applied_against(
+    tmp_path: Path,
+) -> None:
+    """A work-graph edit is traceable to the canonical basis it advanced."""
+
+    from loopx.control_plane.goals.shared_goal_alignment import (
+        project_shared_goal_alignment,
+    )
+
+    project, registry_path = _fixture(tmp_path)
+
+    def basis() -> str:
+        return project_shared_goal_alignment(
+            goal_id=GOAL_ID,
+            agent_id=AGENT_ID,
+            project=project,
+            registry_path=registry_path,
+        )["source_basis"]["source_basis_digest"]
+
+    before = basis()
+    receipts = _settle(registry_path, _proposal())
+    recorded = receipts[0]["intent_basis"]
+    assert re.fullmatch(r"sha256:[0-9a-f]{64}", recorded)
+    # The receipt names the revision the edit was applied against, not the one
+    # the edit itself produced, and it is the canonical basis rather than a
+    # digest this module invented.
+    assert recorded == before
+    assert basis() != before
+    assert len(validate_governed_transition_receipts(receipts)) == 1
+
+    for malformed in ("sha256:short", "2836abc7", "sha256:" + "A" * 64):
+        with pytest.raises(ValueError, match="intent_basis is invalid"):
+            validate_governed_transition_receipts([_receipt(intent_basis=malformed)])
+    assert len(
+        validate_governed_transition_receipts(
+            [_receipt(intent_basis="sha256:" + "a" * 64)]
+        )
+    ) == 1
+
+
 def test_a_team_plan_receipt_must_not_invent_a_monitor_key() -> None:
     """A plan is not a monitor, so its receipt carries no monitor identity."""