fix: preserve recovery contracts across canonical retirement

.github/workflows/python-tests.yml

@@ -848,6 +848,7 @@ jobs:
           tests/test_file_lock.py
           tests/test_file_lock_cross_process.py
           tests/control_plane/test_effect_runtime_integration.py
+          tests/control_plane/test_shadow_observation_retirement.py
           tests/test_self_update_runtime_activation.py
           tests/test_windows_atomic_writes.py
           tests/test_windows_install.py

docs/reference/canonical-todo-completion-update.md

@@ -598,8 +598,12 @@ canonical 写入；provider 丢失时明确拒绝，不重建空库。
 
 Canonical team-plan preview, confirmed batch commit and original-operation
 replay use the TypeScript batch transaction and canonical projection outbox.
-An unpromoted request returns `canonical_authority_required`; it cannot load a
-capture producer or Markdown editor.
+An unpromoted fresh request returns `canonical_authority_required`; it cannot
+load a capture producer or Markdown editor. An exact pre-cutoff receipt may be
+read back before import through the same typed identity validator, without
+creating tasks, capture evidence or execution authority. The locked source
+binding is revalidated first; a selected but unavailable provider still fails
+rather than replaying from Markdown as fallback.
 
 The real File/SQLite Chat oracle physically removes the capture adapter from
 the package before preview and confirmation. It checks provider outage without
@@ -613,8 +617,10 @@ Projection IO and historical receipt readers remain necessary. Assignment still
 grants neither receiver adoption nor execution authority.
 
 canonical team-plan 的预览、确认后整批提交与原操作重放复用 TS batch 事务及
-canonical 投影 outbox。未迁移请求返回 `canonical_authority_required`，不能加载
-capture producer 或 Markdown editor。
+canonical 投影 outbox。未迁移的新请求返回 `canonical_authority_required`，不能加载
+capture producer 或 Markdown editor。原 pre-cutoff 回执在导入前也可通过同一 TS
+身份校验读回，不创建 Todo、capture 或执行授权。先在原锁内核验来源绑定；已选定
+provider 不可用时仍明确拒绝，不能将 Markdown 历史读取当作 fallback。
 真实 File/SQLite Chat 检查在预览及确认前物理移除该模块，验证 provider 失效无
 Markdown fallback、恢复、后续写入后的原操作重放、审核依据过期及提交后投影失败
 恢复；导入前的历史回执重放保留后来写入与当前 revision。正常 wheel 复用同一

examples/control_plane/auto-research-role-successor-completion-smoke.py

@@ -13,6 +13,7 @@
     sys.path.insert(0, str(REPO_ROOT))
 
 from loopx.control_plane.testing.canary_harness import (  # noqa: E402
+    import_fixture_authority,
     run_json_cli,
     run_json_cli_result,
     write_fixture_registry,
@@ -71,6 +72,7 @@ def write_fixture(root: Path, *, auto_research: bool) -> tuple[Path, Path, Path]
         registered_agents=[CURATOR, PROPOSER, EXECUTOR],
         peer_independent_worktree_required=False,
     )
+    import_fixture_authority(registry_path, GOAL_ID)
     return project, runtime, registry_path
 
 
@@ -122,7 +124,7 @@ def assert_actor_attribution_fails_closed() -> None:
             actor_agent_id=None,
         )
         assert missing_code == 1, missing
-        assert "requires --agent-id" in missing["error"], missing
+        assert missing["error_code"] == "actor_required", missing
         assert state_path.read_text(encoding="utf-8") == before
 
         wrong_code, wrong = complete_proposer(
@@ -131,7 +133,7 @@ def assert_actor_attribution_fails_closed() -> None:
             actor_agent_id=CURATOR,
         )
         assert wrong_code == 1, wrong
-        assert f"claimed_by={PROPOSER!r}" in wrong["error"], wrong
+        assert wrong["error_code"] == "claim_owner_mismatch", wrong
         assert state_path.read_text(encoding="utf-8") == before
 
 
@@ -142,7 +144,6 @@ def assert_auto_research_role_successor_allowed() -> None:
         assert returncode == 0, completed
         assert completed["ok"] is True, completed
         assert completed["status"] == "done", completed
-        assert completed["linked_successor_id"] == EXECUTOR_TODO, completed
         assert completed["successor_todo_ids"] == [EXECUTOR_TODO], completed
         assert completed["next_todos"] == [], completed
 
@@ -168,7 +169,6 @@ def assert_generic_peer_route_links_existing_successor() -> None:
         returncode, completed = complete_proposer(registry_path, runtime)
         assert returncode == 0, completed
         assert completed["ok"] is True, completed
-        assert completed["linked_successor_id"] == EXECUTOR_TODO, completed
         assert completed["successor_todo_ids"] == [EXECUTOR_TODO], completed
 
 

examples/control_plane/todo-claim-visibility-lanes-smoke.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import sys
+import tempfile
 from pathlib import Path
 
 
@@ -17,7 +18,9 @@
 from loopx.control_plane.todos.quota_selection import project_quota_planning  # noqa: E402
 from loopx.control_plane.todos.contract import parse_todo_metadata_line  # noqa: E402
 from loopx.quota import build_quota_should_run  # noqa: E402
-from loopx.todos import apply_todo_update_to_lines  # noqa: E402
+from loopx.control_plane.testing.canary_harness import (  # noqa: E402
+    import_fixture_authority, run_json_cli, run_json_cli_result, write_fixture_registry,
+)
 from work_lane_contract_fixtures import GOAL_ID, status_payload  # noqa: E402
 
 
@@ -423,40 +426,45 @@ def assert_legacy_review_handoff_write_requires_explicit_repair() -> None:
         "continuation_policy=review_handoff -->",
     ]
 
-    for updates in (
-        {"claimed_by": CURRENT_AGENT, "claim_only": True},
-        {"note": "Keep this legacy review blocked."},
-    ):
-        lines = list(original_lines)
-        try:
-            apply_todo_update_to_lines(
-                lines,
-                todo_id="todo_legacy_write",
-                role="agent",
-                updated_at="2026-07-11T06:00:00+08:00",
-                **updates,
-            )
-        except ValueError as exc:
-            assert "repair it" in str(exc), exc
-        else:
-            raise AssertionError("legacy review writes must fail closed")
-        assert lines == original_lines
-
-    repaired_lines = list(original_lines)
-    repaired = apply_todo_update_to_lines(
-        repaired_lines,
-        todo_id="todo_legacy_write",
-        role="agent",
-        continuation_policy="independent_handoff",
-        excluded_agents=[CURRENT_AGENT],
-        updated_at="2026-07-11T06:00:00+08:00",
-    )
-    assert repaired["continuation_policy"] == "independent_handoff", repaired
-    assert repaired["excluded_agents"] == [CURRENT_AGENT], repaired
-    repaired_text = "\n".join(repaired_lines)
-    assert "continuation_policy=review_handoff" not in repaired_text
-    assert "continuation_policy=independent_handoff" in repaired_text
-    assert f"excluded_agents={CURRENT_AGENT}" in repaired_text
+    with tempfile.TemporaryDirectory(prefix="loopx-claim-repair-") as tmp:
+        root = Path(tmp).resolve()
+        state = root / "ACTIVE_GOAL_STATE.md"
+        state.write_text("\n".join(original_lines) + "\n", encoding="utf-8")
+        registry = root / ".loopx/registry.json"
+        write_fixture_registry(
+            project=root, runtime_root=root / "runtime", registry_path=registry,
+            goal_id=GOAL_ID, domain="claim-repair-fixture",
+            adapter_kind="generic_project_goal_v0", state_file=state.name,
+            registered_agents=[CURRENT_AGENT, OTHER_AGENT],
+        )
+        import_fixture_authority(registry, GOAL_ID)
+        before = state.read_bytes()
+        prefix = ("todo", "update", "--goal-id", GOAL_ID,
+                  "--todo-id", "todo_legacy_write", "--agent-id", OTHER_AGENT)
+        revision = run_json_cli("todo", "list", "--goal-id", GOAL_ID,
+                                registry_path=registry)["authority_read"]["provider_revision"]
+        claim = ("todo", "claim", "--goal-id", GOAL_ID,
+                 "--todo-id", "todo_legacy_write", "--agent-id", CURRENT_AGENT,
+                 "--claimed-by", CURRENT_AGENT)
+        for request in (claim, (*prefix, "--note", "Keep this legacy review blocked.")):
+            code, rejected = run_json_cli_result(*request, registry_path=registry)
+            assert code == 1, rejected
+            if request is claim:
+                assert rejected["error_code"] == "removed_continuation_policy", rejected
+            else:
+                assert "repair it" in rejected["error"], rejected
+            assert state.read_bytes() == before
+            assert run_json_cli("todo", "list", "--goal-id", GOAL_ID,
+                                registry_path=registry)["authority_read"]["provider_revision"] == revision
+        repaired = run_json_cli(*prefix, "--continuation-policy", "independent_handoff",
+                                "--excluded-agent", CURRENT_AGENT, registry_path=registry)
+        assert repaired["source_authority"] == "file_v0", repaired
+        current = run_json_cli("todo", "list", "--goal-id", GOAL_ID,
+                               "--todo-id", "todo_legacy_write", registry_path=registry)["todo"]
+        assert current["continuation_policy"] == "independent_handoff", current
+        assert current["excluded_agents"] == [CURRENT_AGENT], current
+        assert not current.get("removed_continuation_policy"), current
+        assert "continuation_policy=review_handoff" not in state.read_text()
 
 
 def assert_claim_visibility_lanes_split_current_other_and_task_class() -> None: