fix(storage): recover original team plans during a writer hold

docs/architecture/rfcs/typescript-control-plane-migration-v0.md

@@ -113,6 +113,16 @@ The existing delivery Todos should carry these concrete exits:
 This updates the existing R5/T4 frontier and release proposal; it creates no
 parallel migration owner, runtime gate or release activation.
 
+Team-plan recovery prerequisite: an unpromoted plan that published its original
+receipt can now recover through the Chat confirmation owner while shadow
+bootstrap is interrupted. Registry/Todo/source locks still protect the read,
+and the existing typed identity validator rejects changed intent, actor or
+reviewed basis. Recovery verifies source durability without loading the capture
+producer or admitting new writes; fresh plans still obey all source, management
+and promotion guards. This repairs the existing 1.x recovery path and does not
+activate the 2.0 cutoff or settle the remaining lease/outbox/Host obligations.
+
+
 The retained import-compatible field writer now composes resume normalization
 and contradictory-clear rejection in its existing typed field plan. Keep blank
 Python input omission and locked IO; retire the duplicate normalization crossing,

loopx/control_plane/coordination/legacy_writer_fence.py

@@ -339,6 +339,22 @@ def require_legacy_coordination_write_allowed(
     )
 
 
+def require_legacy_todo_write_allowed(
+    *, registry_path: Path, runtime_root: Path, goal_id: str, state_file: Path,
+) -> None:
+    """Admit a fresh source mutation while its Todo/state locks are held.
+
+    Historical receipt reads use the same locks but do not admit a new write.
+    Keep source binding, management and promotion checks together for writers.
+    """
+    require_registry_source_write_allowed(
+        registry_path=registry_path, runtime_root=runtime_root, goal_id=goal_id,
+        state_file=state_file,
+    )
+    require_shadow_primary_write_allowed(runtime_root, goal_id)
+    require_legacy_coordination_write_allowed(runtime_root=runtime_root, goal_id=goal_id)
+
+
 @contextmanager
 def legacy_todo_write_transaction(
     registry_path: Path,
@@ -376,13 +392,8 @@ def legacy_todo_write_transaction(
         operation=operation,
     ):
         if not dry_run:
-            require_registry_source_write_allowed(
-                registry_path=registry_path, runtime_root=resolved_runtime_root, goal_id=goal_id,
-                state_file=state_file,
-            )
-            require_shadow_primary_write_allowed(resolved_runtime_root, goal_id)
-            require_legacy_coordination_write_allowed(
-                runtime_root=resolved_runtime_root,
-                goal_id=goal_id,
+            require_legacy_todo_write_allowed(
+                registry_path=registry_path, runtime_root=resolved_runtime_root,
+                goal_id=goal_id, state_file=state_file,
             )
         yield

loopx/control_plane/work_items/team_plan_adapter.py

@@ -231,39 +231,52 @@ def previous_receipt(original: str) -> dict[str, Any] | None:
         result["projection_delivery"] = projected.get("projection_delivery")
         return result
 
-    # Only unpromoted Goals need the source writer and capture producer. Keep
-    # canonical replay and projection independent of this compatibility path.
-    from ..coordination.legacy_writer_fence import legacy_todo_write_transaction
-    from ..coordination.runtime_shadow_writer_adapter import (
-        begin_todo_runtime_shadow_capture, settle_todo_runtime_shadow_capture,
-    )
+    # Lock the original source before consulting its receipt. A management hold
+    # stops fresh writes, not readback of an already-published operation. Keep
+    # receipt identity validation in the shared typed planner.
+    from ..coordination.legacy_writer_fence import require_legacy_todo_write_allowed
     from ..runtime.document_io import verify_state_text_durable
-    from ..todos.active_state_editing import (
-        insert_into_existing_section, insert_new_section, section_bounds,
-        replace_updated_at,
-    )
-    from ..todos.contract import format_todo_metadata_line, parse_todo_metadata_line
 
     with exclusive_cross_runtime_file_lock(registry_path, operation="team_plan_commit"), \
-            legacy_todo_write_transaction(registry_path, goal_id, state, agent_id,
-            "todo_add", False, runtime_root=runtime):
+            exclusive_cross_runtime_file_lock(
+                legacy_coordination_todo_lock_path(runtime_root=runtime, goal_id=goal_id),
+                agent_id=agent_id, operation="legacy_coordination_todo_write"), \
+            exclusive_cross_runtime_file_lock(state, agent_id=agent_id, operation="todo_add"):
         read_source_request()
         original = state.read_text(encoding="utf-8")
         previous = previous_receipt(original)
+        if previous is not None:
+            # Promotion may have completed while these locks were acquired.
+            # An unavailable selected provider must still fail closed; this is
+            # original receipt recovery, never a Markdown authority fallback.
+            read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=goal_id)
+            replay = effect_runtime_result("work_items.team_plan.plan", {
+                **request, "previous_receipt": previous})
+            verify_state_text_durable(state, original)
+            return dict(replay["result"])
+        require_legacy_todo_write_allowed(
+            registry_path=registry_path, runtime_root=runtime, goal_id=goal_id, state_file=state)
+        # Recovery above needs neither the source producer nor Markdown editing.
+        from ..coordination.runtime_shadow_writer_adapter import (
+            begin_todo_runtime_shadow_capture, settle_todo_runtime_shadow_capture,
+        )
+        from ..todos.active_state_editing import (
+            insert_into_existing_section, insert_new_section, section_bounds,
+            replace_updated_at,
+        )
+        from ..todos.contract import format_todo_metadata_line, parse_todo_metadata_line
+
         existing = [metadata for line in original.splitlines()
                     if (metadata := parse_todo_metadata_line(line)) is not None]
         current_fingerprint = read_fingerprint() if read_fingerprint else None
         try:
             planned = effect_runtime_result("work_items.team_plan.plan", {
-                **request, "previous_receipt": previous, "todos": existing,
+                **request, "todos": existing,
                 "current_state_fingerprint": current_fingerprint,
             })
         except EffectRuntimeRejected as error:
             raise TeamPlanCommitError(error.diagnostic_code, str(error), current_fingerprint) from None
         result = dict(planned["result"])
-        if planned["replayed"]:
-            verify_state_text_durable(state, original)
-            return result
         capture = begin_todo_runtime_shadow_capture(registry_path=registry_path,
             runtime_root=runtime, goal_id=goal_id, state_path=state,
             write_class="todo_add", original_text=original)

loopx/semantics/project_registry_io_manifest_v1.json

@@ -1119,7 +1119,7 @@
     },
     {
       "site": "loopx/control_plane/coordination/legacy_writer_fence.py::<module>.legacy_todo_write_transaction::codec_read:load_registry#1",
-      "line": 362,
+      "line": 378,
       "column": 9,
       "kind": "codec_read",
       "api": "load_registry",

tests/control_plane/test_team_plan_authority.py

@@ -478,3 +478,120 @@ def test_capture_free_chat_preserves_pre_promotion_receipt(tmp_path, monkeypatch
     assert replay["lane_todo_ids"] == applied["receipt"]["resource_ids"]["lane_todo_ids"]
     assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a") == current
     assert state.read_bytes() == display
+
+    # Simulate promotion finishing after the first mode observation but before
+    # acquiring source locks; its selected backend subsequently goes offline.
+    from loopx.control_plane.coordination.local_authority import LocalCoordinationAuthorityUnavailable
+    from loopx.control_plane.work_items import team_plan_adapter
+    read_authority = team_plan_adapter.read_canonical_todos_if_promoted
+    observed = []
+
+    def promotion_race(**kwargs):
+        observed.append(True)
+        return None if len(observed) == 1 else read_authority(**kwargs)
+
+    backend = runtime / "authority" / current["source_authority"].replace("_", "-")
+    offline = backend.with_name(backend.name + "-offline")
+    backend.rename(offline)
+    try:
+        with monkeypatch.context() as race:
+            race.setattr(team_plan_adapter, "read_canonical_todos_if_promoted", promotion_race)
+            saved = service.store.load(preview["proposal_id"])
+            with pytest.raises(LocalCoordinationAuthorityUnavailable):
+                team_plan_adapter.settle_team_plan(
+                    registry_path=registry, goal_id="goal-a", agent_id=None,
+                    proposal={**saved["normalized_parameters"]["plan"], "proposal_id": preview["proposal_id"]},
+                    expected_state_fingerprint=saved["expected_state_fingerprint"], basis_agent_id="alpha")
+            assert len(observed) == 2
+        assert state.read_bytes() == display and not backend.exists()
+    finally:
+        offline.rename(backend)
+    assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a") == current
+
+
+@pytest.mark.skipif(os.name != "posix", reason="real bootstrap crash barrier uses select on a pipe")
+def test_legacy_chat_replay_survives_interrupted_shadow_bootstrap(tmp_path, monkeypatch):
+    """A stopped writer must not strand an already-published plan receipt."""
+    import select
+    from loopx.control_plane.work_items import team_plan_adapter
+    from loopx.control_plane.coordination.runtime_shadow import build_runtime_shadow_source_snapshot
+    from loopx.control_plane.coordination.shadow_management import read_shadow_management_state
+    from loopx.control_plane.coordination.coordination_state_contract_generated import (
+        COORDINATION_RUNTIME_SHADOW_BOOTSTRAP_REQUEST_SCHEMA,
+    )
+
+    runtime = tmp_path / "runtime"
+    state = tmp_path / "state.md"
+    state.write_text("# Goal\n\n## Agent Todo\n\n")
+    goal = {"id": "goal-a", "repo": str(tmp_path), "state_file": "state.md",
+        "coordination": {"registered_agents": ["alpha"], "agent_model": "peer_v1"}}
+    registry = tmp_path / "registry.json"
+    registry.write_text(json.dumps({"common_runtime_root": str(runtime), "goals": [goal]}))
+    service = ChatActionService(store=ChatActionStore(runtime / "chat/actions"), registry_path=registry)
+    preview = service.preview({"action_kind": "team.plan", "summary": "Assign work", "context": {},
+        "normalized_parameters": {"goal_id": "goal-a", "plan": _agent_plan()},
+        "idempotency_key": "lost-original-response"})
+    write = team_plan_adapter.write_captured_todo_state
+
+    def lose_response(*args, **kwargs):
+        write(*args, **kwargs)
+        raise OSError("response lost after publication")
+
+    with monkeypatch.context() as failed_response:
+        failed_response.setattr(team_plan_adapter, "write_captured_todo_state", lose_response)
+        assert service.apply(preview["proposal_id"])["proposal"]["status"] == "failed"
+    assert state.read_text().count("loopx:team-plan:") == 1
+    state.write_text(state.read_text() + "\nLater owner data must survive.\n")
+    original = state.read_bytes()
+    projection, snapshot = build_runtime_shadow_source_snapshot(
+        goal=goal, runtime_root=runtime, state_path=state, registry_path=registry)
+    request = {"schema_version": COORDINATION_RUNTIME_SHADOW_BOOTSTRAP_REQUEST_SCHEMA,
+        "runtime_root": str(runtime), "goal_id": "goal-a", "operation_id": "bootstrap:plan-recovery",
+        "source_version": "source:before-crash", "projection": projection, "source_snapshot": snapshot}
+    repo = Path(__file__).resolve().parents[2]
+    child = subprocess.Popen(["node", "--no-warnings", "--experimental-strip-types",
+        str(repo / "tests/control_plane_ts/fixtures/shadow_management_crash_worker.ts"),
+        "bootstrap-public", json.dumps(request), "bootstrap_prepared"],
+        cwd=repo, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+    try:
+        ready, _, _ = select.select([child.stdout], [], [], 15)
+        assert ready, "bootstrap did not reach its real filesystem barrier"
+        assert child.stdout.readline().strip() == "ready:bootstrap_prepared"
+    finally:
+        child.kill()
+        child.communicate(timeout=10)
+    pending = read_shadow_management_state(runtime, "goal-a")
+    assert pending["status"] == "bootstrapping"
+    # Restart from an isolated package with the old capture producer absent.
+    invoke = _capture_free_chat((runtime, state, service, preview), tmp_path)
+    replay = invoke("historical")
+    assert replay["action"] == "reused" and replay["created_todo_ids"] == []
+    recovered = service.apply(preview["proposal_id"])["proposal"]
+    assert recovered["status"] == "applied", recovered
+    assert replay["lane_todo_ids"] == recovered["receipt"]["resource_ids"]["lane_todo_ids"]
+    assert state.read_bytes() == original
+    assert read_shadow_management_state(runtime, "goal-a") == pending
+
+    # Recovery is tied to the exact intent, actor and original reviewed basis.
+    saved = service.store.load(preview["proposal_id"])
+    original_plan = {**saved["normalized_parameters"]["plan"], "proposal_id": preview["proposal_id"]}
+    for changed in (
+        {"proposal": {**original_plan, "objective": "Changed intent"}},
+        {"agent_id": "alpha"},
+        {"expected_state_fingerprint": "different-basis"},
+    ):
+        with pytest.raises(team_plan_adapter.EffectRuntimeRejected, match="identity mismatch"):
+            team_plan_adapter.settle_team_plan(**{
+                "registry_path": registry, "goal_id": "goal-a", "agent_id": None,
+                "proposal": original_plan, "expected_state_fingerprint": saved["expected_state_fingerprint"],
+                "basis_agent_id": "alpha", **changed,
+            })
+        assert state.read_bytes() == original
+    # A different plan still needs normal admission; recovery cannot restart work.
+    fresh = service.preview({"action_kind": "team.plan", "summary": "New work", "context": {},
+        "normalized_parameters": {"goal_id": "goal-a", "plan": _agent_plan()},
+        "idempotency_key": "new-request-during-hold"})
+    refused = service.apply(fresh["proposal_id"])["proposal"]
+    assert refused["status"] == "failed", refused
+    assert state.read_bytes() == original
+    assert read_shadow_management_state(runtime, "goal-a") == pending