From 69bcfdabc3fca361d5bf5e86c7302c2949bf2e21 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 13 Oct 2026 02:04:58 +0800 Subject: [PATCH 1/2] test(shadow): separate Monitor replay from bounded catch-up Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../test_local_authority_shadow_drain.py | 37 ++++++++++++ .../test_monitor_poll_transaction_recovery.py | 57 +++++++++++++++++-- tests/control_plane_ts/shadow_drain.test.ts | 35 ++++++++++++ 3 files changed, 123 insertions(+), 6 deletions(-) diff --git a/tests/control_plane/test_local_authority_shadow_drain.py b/tests/control_plane/test_local_authority_shadow_drain.py index 9998e4de0c..7874ad2229 100644 --- a/tests/control_plane/test_local_authority_shadow_drain.py +++ b/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): diff --git a/tests/control_plane/test_monitor_poll_transaction_recovery.py b/tests/control_plane/test_monitor_poll_transaction_recovery.py index 802eca1ac5..e7c4c3c175 100644 --- a/tests/control_plane/test_monitor_poll_transaction_recovery.py +++ b/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: diff --git a/tests/control_plane_ts/shadow_drain.test.ts b/tests/control_plane_ts/shadow_drain.test.ts index 40618069cd..cfabe61607 100644 --- a/tests/control_plane_ts/shadow_drain.test.ts +++ b/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()]}; From 032f8fbb83882bad827515e39a55246689f41fcb Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 13 Oct 2026 02:04:58 +0800 Subject: [PATCH 2/2] docs(shadow): clarify independent catch-up after receipt replay Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../rfcs/typescript-control-plane-migration-v0.md | 7 ++++++- .../rfcs/typescript-control-plane-migration-v0.zh-CN.md | 6 +++++- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index 6ce43b579a..e896e74ba8 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/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). diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md index 217db63173..bfc60649c4 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md +++ b/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)。