test(shadow): separate Monitor replay from bounded catch-up

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

@@ -187,7 +187,12 @@ codec/replay has a small shared scheduler IO owner, while the existing TS batch
 still owns operation identity and decisions. File/SQLite CLI and normal-wheel
 source removal must preserve later canonical writes and original successors;
 promoted replay leaves old outbox bytes inert. Unpromoted markerless batch
-recovery retains its existing drain. This removes a canonical import dependency,
+recovery retains its existing bounded drain. Business-receipt replay does not
+guarantee shadow catch-up: lock deferral or an expired proof budget preserves
+pending evidence for an independent `authority-shadow drain`. Recovery keeps
+the original source bytes, successor IDs and once-only quota effect; it does
+not replay the writer to finish catch-up or increase the inline limits.
+This removes a canonical import dependency,
 not the supported legacy writer. The absent arm also removes all four shadow
 producer modules using the qualified CLI startup isolation above.
 [Retained duties and rollback](../../reference/canonical-todo-completion-update.md#monitor-source-writer-isolation).

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

@@ -125,7 +125,11 @@ cohort 采用及冻结 D2 证据保持独立。
 Monitor router 同样隔离活跃旧源 writer；历史回执 codec／replay 共用 scheduler
 只读 IO owner，原操作身份和决策继续归既有 TS batch。File/SQLite CLI 与正常
 wheel 删源检查须保留后来写入及原 successor；晋升后重放保持旧 outbox 静止，
-未晋升的 markerless batch 恢复保留原 drain。本批只移除 canonical 导入依赖，
+未晋升的 markerless batch 恢复保留原有界 drain。业务回执重放不保证 shadow
+立即追平：锁延期或 proof 预算耗尽时保留待处理证据，由独立的
+`authority-shadow drain` 恢复。恢复须保持原 source bytes、successor ID 和 quota
+effect 恰好一次；不能为追平而重放 writer 或扩大 inline 上限。
+本批只移除 canonical 导入依赖，
 不删除受支持的旧 writer；缺席臂复用上述 CLI 启动隔离，同时移除四个 shadow
 producer module。
 见[保留义务及回滚](../../reference/canonical-todo-completion-update.md#monitor-source-writer-isolation)。

tests/control_plane/test_local_authority_shadow_drain.py

@@ -201,6 +201,43 @@ def test_drain_defers_when_another_drainer_holds_the_lock(tmp_path: Path) -> Non
     assert not (runtime_root / "authority-shadow" / "file").exists()
 
 
+def test_lost_drain_response_never_replays_the_effect(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    registry, state, runtime_root = _fixture(tmp_path)
+    _record_todo_write(registry, state, runtime_root, "Commit whose response is lost")
+    source = state.read_bytes()
+    original = adapter.effect_runtime_result
+    attempts: list[dict] = []
+
+    def lose_response(method: str, params: dict, **kwargs: object) -> object:
+        if method != "coordination.runtime_shadow.drain":
+            return original(method, params, **kwargs)
+        attempts.append(kwargs)
+        committed = original(method, params, **kwargs)
+        assert committed["delivered"] == 1
+        raise TimeoutError("injected lost response after the real drain committed")
+
+    limits = {"max_entries": 16, "budget_seconds": 2, "lock_timeout_seconds": .25}
+    with monkeypatch.context() as patch:
+        patch.setattr(adapter, "effect_runtime_result", lose_response)
+        unknown = _drain(registry, runtime_root, **limits)
+    assert unknown.outcome == "stopped"
+    assert unknown.reason_code == "shadow_drain_outcome_unknown"
+    assert len(attempts) == 1
+    assert attempts[0]["retry_safe"] is False
+    assert attempts[0]["timeout"] == 17
+    committed = adapter.read_local_authority_shadow(
+        runtime_root=runtime_root, goal_id=GOAL_ID, scan_limit=20)
+    assert len(committed["proof"]["transactions"]) == 2
+    assert state.read_bytes() == source
+    assert _drain(registry, runtime_root, **limits).outcome == "nothing_pending"
+    assert adapter.read_local_authority_shadow(
+        runtime_root=runtime_root, goal_id=GOAL_ID, scan_limit=20,
+    )["proof"]["transactions"] == committed["proof"]["transactions"]
+    assert state.read_bytes() == source
+
+
 def test_drain_batch_is_bounded_and_reports_what_it_left(tmp_path: Path) -> None:
     registry, state, runtime_root = _fixture(tmp_path)
     for index in range(3):

tests/control_plane/test_monitor_poll_transaction_recovery.py

@@ -8,6 +8,7 @@
 import re
 import subprocess
 import sys
+from contextlib import nullcontext
 from pathlib import Path
 
 import pytest
@@ -29,6 +30,8 @@
     read_local_authority_shadow,
 )
 from loopx.control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result
+from loopx.file_lock import exclusive_cross_runtime_file_lock
+from loopx.control_plane.coordination.local_authority_shadow_outbox import drain_lock_target
 from loopx.todos import add_goal_todo, list_goal_todos
 
 from canonical_authority_fixture import (
@@ -303,8 +306,9 @@ def test_malformed_business_receipt_is_not_treated_as_absent(tmp_path: Path) ->
     assert _poll_events(runtime) == []
 
 
-def test_enabled_shadow_ordinary_retry_recovers_one_prepared_batch(
-    tmp_path: Path,
+@pytest.mark.parametrize("held_drain_lock", [False, True], ids=["uncontended", "deferred"])
+def test_enabled_shadow_business_replay_then_bounded_drain_recovers_batch(
+    tmp_path: Path, held_drain_lock: bool,
 ) -> None:
     registry, runtime, state, _monitor_id, poll = _fixture(tmp_path)
     config = json.loads(registry.read_text(encoding="utf-8"))
@@ -319,7 +323,8 @@ def test_enabled_shadow_ordinary_retry_recovers_one_prepared_batch(
 
     assert _cli(registry, runtime, *poll, fault="after") == (83, None)
     outbox = runtime / "authority-shadow" / "outbox" / GOAL / "todos"
-    assert len(list(outbox.glob("*.prepared.json"))) == 1
+    (prepared,) = outbox.glob("*.prepared.json")
+    prepared_bytes = prepared.read_bytes()
     assert list(outbox.glob("*.committed.json")) == []
     assert "coordination_runtime_shadow" not in _business_receipt(state)["writeback"]
 
@@ -328,15 +333,55 @@ def test_enabled_shadow_ordinary_retry_recovers_one_prepared_batch(
     assert {todo["text"] for todo in before["head"]["todos"]} == {
         "Observe public synthetic target"}
     state_after = state.read_bytes()
-    retry = _ok(registry, runtime, *poll)
+    original_ids = {
+        todo["todo_id"] for todo in _business_receipt(state)["writeback"]["next_todos"]}
+    drain = ("authority-shadow", "drain", "--goal-id", GOAL,
+             "--max-entries", "16", "--budget-seconds", "2",
+             "--lock-timeout-seconds", "0.25")
+    lock = (exclusive_cross_runtime_file_lock(
+        drain_lock_target(runtime, GOAL), timeout_seconds=1, operation="test_drain_deferral")
+        if held_drain_lock else nullcontext())
+    with lock:
+        retry = _ok(registry, runtime, *poll)
+        assert state.read_bytes() == state_after
+        if held_drain_lock:
+            # Business replay succeeds even when independent catch-up must defer.
+            code, deferred = _cli(registry, runtime, *drain)
+            assert code == 1 and isinstance(deferred, dict), deferred
+            assert deferred["outcome"] == "drain_deferred"
+            assert deferred["reason_code"] == "drain_lock_busy"
+            assert deferred["prepared_only_after"] == 1
+            assert prepared.read_bytes() == prepared_bytes
+            assert read_local_authority_shadow(
+                runtime_root=runtime, goal_id=GOAL, scan_limit=20,
+            )["proof"]["transactions"] == before["proof"]["transactions"]
     assert {todo["todo"] for todo in retry["todo_writeback"]["next_todos"]} == {
         AGENT_TASK, USER_TASK}
+    assert set(retry["successor_todo_ids"]) == original_ids
+    assert state.read_bytes() == state_after
+    receipt_after = _quota_receipt(runtime).read_bytes()
+    events_after = _poll_events(runtime)
+    assert len(events_after) == 1
+    # A bounded inline attempt is not a catch-up guarantee. Use the existing
+    # independent drain with its original limits, without replaying the writer.
+    recovered = _ok(registry, runtime, *drain)
+    assert recovered["pending_after"] == recovered["prepared_only_after"] == 0
     assert state.read_bytes() == state_after
     view = read_local_authority_shadow(runtime_root=runtime, goal_id=GOAL,
                                       scan_limit=20)
-    assert {todo["text"] for todo in view["head"]["todos"]} >= {
-        AGENT_TASK, USER_TASK}
+    assert {todo["text"] for todo in view["head"]["todos"]} == {
+        "Observe public synthetic target", AGENT_TASK, USER_TASK}
+    assert len(view["proof"]["transactions"]) == 2  # Baseline and original batch.
     assert list(outbox.glob("*.prepared.json")) == []
+    late_retry = _ok(registry, runtime, *poll)
+    assert late_retry["replayed"] is True
+    assert set(late_retry["successor_todo_ids"]) == original_ids
+    assert state.read_bytes() == state_after
+    assert _quota_receipt(runtime).read_bytes() == receipt_after
+    assert _poll_events(runtime) == events_after
+    assert read_local_authority_shadow(
+        runtime_root=runtime, goal_id=GOAL, scan_limit=20,
+    )["proof"]["transactions"] == view["proof"]["transactions"]
 
 
 def test_committed_receipt_without_batch_version_still_replays(tmp_path: Path) -> None:

tests/control_plane_ts/shadow_drain.test.ts

@@ -85,6 +85,41 @@ test("deadline expiring in proof keeps recoverable evidence", async t => {
   assert.equal((await drainShadowOutbox(request(f))).replayed, 1);
 });
 
+test("inline deadline retains a markerless source commit for an independent drain", async t => {
+  const f = await fixture(t);
+  const entry = await pendingEntry(f, 1, {handoff_mode: "hard_lease", todos: [todo()]}, {marker: false});
+  const before = await inventory(f), source = await readFile(f.statePath);
+  const history = await f.store.scanCommitted(null, 10);
+  const limits = {max_entries: 16, budget_seconds: 2, lock_timeout_seconds: .25};
+  let delayed = false;
+  const first = await drainShadowOutbox(request(f, limits), {afterEffect: async phase => {
+    if (phase === "after_proof" && !delayed) {
+      delayed = true;
+      await new Promise(resolve => setTimeout(resolve, 2100));
+    }
+  }});
+  assert.equal(delayed, true);
+  assert.equal(first.budget_exhausted, true);
+  assert.equal(first.prepared_only_after, 1);
+  assert.equal(first.delivered, 0);
+  assert.equal(first.reclaimed_residue, 0);
+  assert.deepEqual(await inventory(f), before);
+  assert.deepEqual(await readFile(f.statePath), source);
+  assert.deepEqual(await f.store.scanCommitted(null, 10), history);
+
+  const second = await drainShadowOutbox(request(f, limits));
+  assert.equal(second.ok, true, JSON.stringify(second));
+  assert.equal(second.delivered, 1);
+  assert.equal(second.prepared_only_after, 0);
+  assert.equal(second.pending_after, 0);
+  assert.equal((second.entries as JsonObject[])[0].entry_id, (entry.entry as JsonObject).entry_id);
+  assert.equal((second.entries as JsonObject[])[0].resolution, "committed_proven_by_readback");
+  assert.deepEqual(await readFile(f.statePath), source);
+  const committed = await f.store.scanCommitted(null, 10);
+  assert.equal((await drainShadowOutbox(request(f, limits))).outcome, "nothing_pending");
+  assert.deepEqual(await f.store.scanCommitted(null, 10), committed);
+});
+
 test("one-entry budget leaves the remaining source writes intact", async t => {
   const f = await fixture(t);
   const one = {handoff_mode: "hard_lease", todos: [todo()]};