From c0de7613788849aff2f2fd4b092b21aa070ca759 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 13 Oct 2026 01:33:51 +0800 Subject: [PATCH 01/10] fix(quota): fence polls after unavailable Monitor closeout Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../quota/monitor_poll_commit.ts | 28 +++++++++++++++---- 1 file changed, 23 insertions(+), 5 deletions(-) diff --git a/loopx/control_plane/quota/monitor_poll_commit.ts b/loopx/control_plane/quota/monitor_poll_commit.ts index 3e12392ab7..2305436006 100644 --- a/loopx/control_plane/quota/monitor_poll_commit.ts +++ b/loopx/control_plane/quota/monitor_poll_commit.ts @@ -8,6 +8,7 @@ import { basename, dirname, join, resolve } from "node:path"; import { monitorSuccessorIntent, monitorSuccessorRoute } from "../scheduler/monitor_successor.ts"; import { normalizeTodoCapabilities } from "../todos/work_requirements.ts"; import { parseProjectionDelivery } from "../todos/projection_delivery.ts"; +import { isMonitorUnavailableWait } from "./blocked_wait.ts"; import { QUOTA_SETTLEMENT_READBACK_REQUEST_SCHEMA, readQuotaSettlementForAdmittedOwnerFromSnapshot, @@ -563,9 +564,10 @@ interface Admission { external: boolean; } -async function readAuxiliarySettlement(request: MonitorRequest): Promise { +async function readMonitorSettlement(request: MonitorRequest): Promise { + const todoId = request.observation.settlement_todo_id ?? request.observation.todo_id; if (!request.runtime_root || !request.turn_instance_id || - !request.observation.actor_agent_id || !request.observation.settlement_todo_id) return null; + !request.observation.actor_agent_id || !todoId) return null; const snapshot = await readQuotaSettlementSnapshot( request.runtime_root, request.goal_id, @@ -573,7 +575,7 @@ async function readAuxiliarySettlement(request: MonitorRequest): Promise typeof value !== "string") || todo.excluded_agents.includes(decision.agent_id)))) throw conflict(); if (!historicalAdmission) { - const settlement = await readAuxiliarySettlement(request); + const settlement = await readMonitorSettlement(request); const progress = jsonObject(settlement?.progress); // The advancement's single debit is not an observation-admission fence. // A settled exact binding still permits independently admitted due Monitors. @@ -632,6 +634,22 @@ async function auxiliaryMonitorAllowed( } async function admission(request: MonitorRequest, historicalAdmission = false): Promise { + const observation = request.observation; + if (observation.todo_id && + (observation.settlement_todo_id ?? observation.todo_id) === observation.todo_id) { + const settlement = await readMonitorSettlement(request); + // The readback owner verifies admission, exact identity and durable blocked + // evidence. A late observation cannot replace that frozen no-spend outcome. + // Durable poll replay returns before admission; independent auxiliary + // Monitors retain their advancement-bound admission below. + if (jsonObject(settlement?.progress)?.closeout_kind === "typed_blocked_writeback_no_spend" && + isMonitorUnavailableWait(jsonObject(settlement?.writeback_run)?.blocked_retry, observation.todo_id)) { + throw new EffectRuntimeRequestError( + "monitor-poll conflicts with the committed unavailable attempt; use a fresh Turn for a new observation", + "monitor_poll_admission_rejected", + ); + } + } const blocked = blockedSuccessorAllowed(request.decision); const external = externalMonitorAllowed(request.decision); const auxiliary = await auxiliaryMonitorAllowed(request, historicalAdmission); @@ -1562,7 +1580,7 @@ function successorReceipts(providerReceipt: JsonObject | null): JsonObject[] { async function projectAuxiliaryContinuation(request: MonitorRequest, payload: JsonObject): Promise { const continuation = jsonObject(payload.turn_continuation); if (!request.execute || continuation?.settlement_binding_matches_observation !== false) return; - const settlement = await readAuxiliarySettlement(request); + const settlement = await readMonitorSettlement(request); if (jsonObject(settlement?.progress)?.state !== "settled") return; // This is current readback, not a change to the historical Monitor receipt // or permission for another advancement. Replays must not reopen a paid Turn. From a94fb43c91cf7a3a1db07d359cb0f49c7323cec0 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 13 Oct 2026 01:33:51 +0800 Subject: [PATCH 02/10] test(quota): preserve unavailable receipts across late poll attempts Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../test_monitor_unavailable_closeout_cli.py | 50 ++++++++++++++++++- 1 file changed, 49 insertions(+), 1 deletion(-) diff --git a/tests/control_plane/test_monitor_unavailable_closeout_cli.py b/tests/control_plane/test_monitor_unavailable_closeout_cli.py index 406ab543b8..f68f6ae6c2 100644 --- a/tests/control_plane/test_monitor_unavailable_closeout_cli.py +++ b/tests/control_plane/test_monitor_unavailable_closeout_cli.py @@ -23,7 +23,9 @@ @pytest.mark.parametrize("provider", ["markdown", "file", "sqlite"]) -@pytest.mark.parametrize("scenario", ["unavailable", "capability_restore", "observed", "peer_git"]) +@pytest.mark.parametrize("scenario", [ + "unavailable", "capability_restore", "observed", "peer_git", "late_nonmaterial", "late_material", +]) def test_unavailable_monitor_closes_original_turn_without_observation_or_debit( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], provider: str, scenario: str, @@ -160,6 +162,52 @@ def cli(*args: str) -> tuple[int, dict]: assert rc == 0 and after["todo"]["status"] == "open", after assert {k: after["todo"].get(k) for k in OBSERVATION_FIELDS} == fields assert not after["todo"].get("resume_when") + if scenario.startswith("late_"): + poll_args = ("quota", "monitor-poll", "--goal-id", GOAL_ID, "--codex-app", *binding, + "--target-key", "synthetic-source", "--result-hash", "late-observation", + *lease_args) + if scenario == "late_material": + poll_args += ("--material-change", "--next-agent-todo", "Inspect the source transition.", + "--next-action-kind", "validate") + for execute in (True, False): + rc, late = cli(*poll_args, *(("--execute",) if execute else ())) + read_rc, current = cli("todo", "list", "--goal-id", GOAL_ID, "--todo-id", MONITOR_ID) + replay_rc, frozen = cli(*refresh) + assert rc != 0 and late["error_code"] == "monitor_poll_admission_rejected", json.dumps({ + "poll_rc": rc, "appended": late.get("appended"), "error_code": late.get("error_code"), + "observation_changed": read_rc == 0 and + {k: current["todo"].get(k) for k in OBSERVATION_FIELDS} != fields, + "refresh_rc": replay_rc, "refresh_error_code": frozen.get("error_code"), + }) + assert read_rc == 0 and {k: current["todo"].get(k) for k in OBSERVATION_FIELDS} == fields + assert replay_rc == 0 and frozen["idempotent_replay"] is True, frozen + assert frozen["blocked_retry"] == wait + rc, after_late = cli("todo", "list", "--goal-id", GOAL_ID) + assert rc == 0, after_late + monitor = next(todo for todo in after_late["todos"] if todo["todo_id"] == MONITOR_ID) + assert {k: monitor.get(k) for k in OBSERVATION_FIELDS} == fields + assert {todo["todo_id"] for todo in after_late["todos"]} == { + todo["todo_id"] for todo in listed["todos"] + } + assert _classification_count(runtime, "monitor_observation_unavailable") == 1 + assert _classification_count(runtime, "quota_monitor_poll") == 0 + assert _spend_run_count(runtime) == 0 + fresh_binding = (*binding[:-1], f"fresh-observation-{provider}") + rc, fresh_guard = cli("quota", "should-run", "--goal-id", GOAL_ID, "--codex-app", *fresh_binding) + assert rc == 0 and fresh_guard["should_run"] is True, fresh_guard + fresh_poll = list(poll_args) + fresh_poll[fresh_poll.index("--turn-instance-id") + 1] = fresh_binding[-1] + rc, observed = cli(*fresh_poll, "--execute") + assert rc == 0 and observed["appended"] is True, observed + rc, poll_replay = cli(*fresh_poll, "--execute") + assert rc == 0 and poll_replay["replayed"] is True, poll_replay + rc, original_replay = cli(*refresh) + assert rc == 0 and original_replay["idempotent_replay"] is True, original_replay + assert original_replay["blocked_retry"] == wait + assert _classification_count(runtime, "monitor_observation_unavailable") == 1 + assert _classification_count(runtime, "quota_monitor_poll") == 1 + assert _spend_run_count(runtime) == 0 + return if scenario == "peer_git": assert closed.get("delivery_workspace") is None return From 466ba85a28841614a68613482eb3226a272c786b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 13 Oct 2026 01:33:51 +0800 Subject: [PATCH 03/10] docs(quota): describe immutable unavailable Turn outcomes Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../protocols/quota-blocked-causal-closeout-v0.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/docs/reference/protocols/quota-blocked-causal-closeout-v0.md b/docs/reference/protocols/quota-blocked-causal-closeout-v0.md index 9124ccf4ab..0d05c4b71c 100644 --- a/docs/reference/protocols/quota-blocked-causal-closeout-v0.md +++ b/docs/reference/protocols/quota-blocked-causal-closeout-v0.md @@ -111,6 +111,11 @@ cadence and expiry remain unchanged. It creates neither a retry clock nor an advancement wait, material successor, completion or delivery claim. Existing advancement retry/causal-wait semantics and genuine committed polls retain their behavior. A committed poll cannot be rewritten as an unavailable attempt. +Conversely, a committed unavailable closeout rejects a later material or +non-material poll on that same exact Turn binding before provider writeback, +including dry-run provider plans. Use a freshly admitted Turn for a new +observation; its poll cannot change the original frozen closeout. Genuine poll +replay and independently admitted auxiliary Monitors remain valid. Exact replay returns the frozen receipt, with no duplicate writeback or debit. After closing the original effect, reenter the current recovery host Turn to @@ -207,6 +212,10 @@ excluded/bound 限制仍拒绝。恢复后，模式要求执行 fencing 时获 检查/到期时钟、hash、material generation、cadence 与 expiry 不变。不生成重试时钟、 advancement 等待、material successor、完成或投递声明。旧 advancement 重试/因果等待 与真实 poll 保持原行为；已提交 poll 不能改写为不可观察尝试。 +反向提交顺序也互斥：不可观察结算已提交后，同一精确 Turn 身份的 material 或 +non-material poll 在 provider 写回前拒绝，dry-run 也不返回可写入计划。 +新观察须使用重新准入的新 Turn；它的 poll 不改变原冻结结算。 +真实 poll 的幂等重放及独立准入的 auxiliary Monitor 仍可执行。 精确重放返回冻结回执，不重复写回或扣额；原 effect 关闭后，重新进入当前 recovery host Turn 可选择独立合法任务，不新造第三身份或重绑旧 Turn。缺准入能力、缺证或错绑 From 63defb5ece23a7c620823ee5a781b5ae97abb514 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 13 Oct 2026 01:36:06 +0800 Subject: [PATCH 04/10] test(quota): reenter the admitted recovery Turn for fresh observations Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../test_monitor_unavailable_closeout_cli.py | 25 ++++++++++++++++--- 1 file changed, 21 insertions(+), 4 deletions(-) diff --git a/tests/control_plane/test_monitor_unavailable_closeout_cli.py b/tests/control_plane/test_monitor_unavailable_closeout_cli.py index f68f6ae6c2..21ac1b3557 100644 --- a/tests/control_plane/test_monitor_unavailable_closeout_cli.py +++ b/tests/control_plane/test_monitor_unavailable_closeout_cli.py @@ -192,11 +192,28 @@ def cli(*args: str) -> tuple[int, dict]: assert _classification_count(runtime, "monitor_observation_unavailable") == 1 assert _classification_count(runtime, "quota_monitor_poll") == 0 assert _spend_run_count(runtime) == 0 - fresh_binding = (*binding[:-1], f"fresh-observation-{provider}") - rc, fresh_guard = cli("quota", "should-run", "--goal-id", GOAL_ID, "--codex-app", *fresh_binding) - assert rc == 0 and fresh_guard["should_run"] is True, fresh_guard + # Re-enter the current recovery Turn after closing the original attempt; + # a third identity would leave the already-admitted recovery unresolved. + # Its advancement lane observes this due Monitor through the existing + # auxiliary route, rather than selecting a Monitor as advancement work. + rc, independent = cli("todo", "add", "--goal-id", GOAL_ID, "--role", "agent", + "--text", "Inspect an independent artifact.", + "--task-class", "advancement_task", "--claimed-by", AGENT_ID, + *(("--operation-id", f"fresh-independent-{provider}") + if provider != "markdown" else ())) + assert rc == 0, independent + rc, fresh_guard = cli("quota", "should-run", "--goal-id", GOAL_ID, "--codex-app", + "--agent-id", AGENT_ID, "--turn-instance-id", recovery_turn) + assert rc == 0 and fresh_guard["should_run"] is True, json.dumps({ + "rc": rc, "error_code": fresh_guard.get("error_code"), + "error": fresh_guard.get("error"), "effective_action": fresh_guard.get("effective_action"), + "message": fresh_guard.get("message"), "reason": fresh_guard.get("reason"), + "selection": fresh_guard.get("action_selection"), + }) + assert fresh_guard["heartbeat_receipt"]["settlement_identity"]["todo_id"] != MONITOR_ID + assert fresh_guard["work_lane_contract"]["lane"] == "advancement_task" fresh_poll = list(poll_args) - fresh_poll[fresh_poll.index("--turn-instance-id") + 1] = fresh_binding[-1] + fresh_poll[fresh_poll.index("--turn-instance-id") + 1] = recovery_turn rc, observed = cli(*fresh_poll, "--execute") assert rc == 0 and observed["appended"] is True, observed rc, poll_replay = cli(*fresh_poll, "--execute") From 8587f77b397c8eaac84f98775861113b098a9d7b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 13 Oct 2026 02:19:14 +0800 Subject: [PATCH 05/10] fix(quota): preserve interrupted poll ownership before closeout Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../quota/monitor_poll_commit.ts | 44 ++++++++++++++++--- .../quota/settlement_readback.ts | 21 +++++++-- 2 files changed, 55 insertions(+), 10 deletions(-) diff --git a/loopx/control_plane/quota/monitor_poll_commit.ts b/loopx/control_plane/quota/monitor_poll_commit.ts index 2305436006..050c3dbe7b 100644 --- a/loopx/control_plane/quota/monitor_poll_commit.ts +++ b/loopx/control_plane/quota/monitor_poll_commit.ts @@ -17,11 +17,12 @@ import { import { parseQuotaAccountingOwner, quotaGoalRef, + quotaOwnerOwnsProjection, withBorrowedQuotaAccountingOwner, type QuotaAccountingOwner, } from "./source_admission.ts"; -import type { JsonObject } from "../effect_program.ts"; +import type { JsonObject, SettlementIdentity } from "../effect_program.ts"; import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; import { appendJsonLine, @@ -566,8 +567,9 @@ interface Admission { async function readMonitorSettlement(request: MonitorRequest): Promise { const todoId = request.observation.settlement_todo_id ?? request.observation.todo_id; + const agentId = request.observation.actor_agent_id ?? request.decision.agent_id; if (!request.runtime_root || !request.turn_instance_id || - !request.observation.actor_agent_id || !todoId) return null; + !agentId || !todoId) return null; const snapshot = await readQuotaSettlementSnapshot( request.runtime_root, request.goal_id, @@ -575,7 +577,7 @@ async function readMonitorSettlement(request: MonitorRequest): Promise { const observation = request.observation; - if (observation.todo_id && - (observation.settlement_todo_id ?? observation.todo_id) === observation.todo_id) { + const settlementTodo = observation.settlement_todo_id ?? observation.todo_id; + if (settlementTodo && (!observation.todo_id || settlementTodo === observation.todo_id)) { const settlement = await readMonitorSettlement(request); // The readback owner verifies admission, exact identity and durable blocked // evidence. A late observation cannot replace that frozen no-spend outcome. // Durable poll replay returns before admission; independent auxiliary // Monitors retain their advancement-bound admission below. if (jsonObject(settlement?.progress)?.closeout_kind === "typed_blocked_writeback_no_spend" && - isMonitorUnavailableWait(jsonObject(settlement?.writeback_run)?.blocked_retry, observation.todo_id)) { + isMonitorUnavailableWait(jsonObject(settlement?.writeback_run)?.blocked_retry, settlementTodo)) { throw new EffectRuntimeRequestError( "monitor-poll conflicts with the committed unavailable attempt; use a fresh Turn for a new observation", "monitor_poll_admission_rejected", @@ -1713,6 +1715,7 @@ function receiptObject(value: unknown): MonitorReceipt { "receipt.expected_index_digest", ), expected_index_bytes: expectedBytes, + ...(receipt.goal_ref === undefined ? {} : {goal_ref: requiredObject(receipt.goal_ref, "receipt.goal_ref")}), } as const; const status = requireStringLiteral( receipt.status, @@ -1763,6 +1766,34 @@ async function readReceipt(path: string): Promise { } } +/** A stored observation may already have reached its provider even before the + * run index is settled. Its existing receipt owner must recover or prove no + * effect before a competing unavailable closeout can be admitted. */ +export async function hasMonitorPollTransaction( + runtimeRoot: string, identity: SettlementIdentity, owner: QuotaAccountingOwner, +): Promise { + if (!identity.todo_id) return false; + const runsDir = join(runtimeRoot, "goals", identity.goal_id, "runs"); + const base = `quota-monitor-poll:${identity.goal_id}:${identity.agent_id}:${identity.turn_instance_id}`; + for (const effectId of [base, `${base}:todo:${identity.todo_id}`]) { + const receipt = await readReceipt(transactionPath(runsDir, effectId)); + if (!receipt) continue; + validateReceiptScope(receipt, runsDir, join(runsDir, "index.jsonl")); + const facts = receipt.status === "provider_pending" ? receipt.provider_plan : receipt.index_record; + if (receipt.effect_id !== effectId || facts.goal_id !== identity.goal_id || + facts.agent_id !== identity.agent_id || facts.todo_id !== identity.todo_id) continue; + const goalRef = receipt.status === "provider_pending" ? receipt.goal_ref : facts.goal_ref; + if (receipt.status === "provider_pending" && goalRef === undefined && owner.kind !== "alias") { + throw new EffectRuntimeRequestError( + "historical pending Monitor transaction lacks GoalRef scope; reconcile its original owner before unavailable closeout", + ); + } + if (!quotaOwnerOwnsProjection(owner, goalRef)) continue; + return true; + } + return false; +} + function validateReceiptScope( receipt: MonitorReceipt, runsDir: string, @@ -2329,6 +2360,7 @@ async function evaluateQuotaMonitorPollRequest( expected_index_bytes: indexBytes?.length ?? 0, provider_plan: plan, admitted_decision: request.decision, + ...(quotaGoalRef(request.owner) === null ? {} : {goal_ref: quotaGoalRef(request.owner)}), } satisfies PendingMonitorReceipt; await atomicWriteJson(receiptPath, pending); } diff --git a/loopx/control_plane/quota/settlement_readback.ts b/loopx/control_plane/quota/settlement_readback.ts index 1002d0e0b1..ac96650c23 100644 --- a/loopx/control_plane/quota/settlement_readback.ts +++ b/loopx/control_plane/quota/settlement_readback.ts @@ -21,6 +21,7 @@ import { type SettlementResult, } from "../effect_program.ts"; import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import { hasMonitorPollTransaction } from "./monitor_poll_commit.ts"; import { jsonObject, requireJsonObject, @@ -1373,9 +1374,21 @@ export async function readQuotaSettlement(value: unknown): Promise { : withQuotaAccountingOwner; return await withOwner( request.owner, - async () => readQuotaSettlementFromRequest( - request, - await readQuotaSettlementSnapshot(request.runtime_root, request.goal_id), - ), + async () => { + const result = readQuotaSettlementFromRequest(request, + await readQuotaSettlementSnapshot(request.runtime_root, request.goal_id)); + const identity = jsonObject(jsonObject(jsonObject(result.identity)?.result)?.value); + if (identity && request.refresh_retry && + jsonObject(result.refresh_recovery)?.decision !== "reject" && + isTurnScopedSettlementOutcome(request.refresh_retry.delivery_outcome, + request.refresh_retry.progress_observation, optionalString(identity.todo_id)) && + request.refresh_retry.delivery_outcome === "outcome_gap" && + await hasMonitorPollTransaction(request.runtime_root, identity as unknown as SettlementIdentity, request.owner)) { + throw new EffectRuntimeRequestError( + "a stored Monitor poll may already have committed its observation; recover the exact monitor-poll before an unavailable closeout", + ); + } + return result; + }, ); } From 471b6f22d2a2840dbfada0ea1b0c10011ac127f8 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 13 Oct 2026 02:19:14 +0800 Subject: [PATCH 06/10] test(quota): qualify interrupted observations and Goal owner fences Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../test_monitor_poll_unavailable_recovery.py | 57 +++++++++++++++++++ .../test_monitor_unavailable_closeout_cli.py | 14 +++++ .../quota_monitor_poll_commit.test.ts | 50 ++++++++++++++++ 3 files changed, 121 insertions(+) create mode 100644 tests/control_plane/test_monitor_poll_unavailable_recovery.py diff --git a/tests/control_plane/test_monitor_poll_unavailable_recovery.py b/tests/control_plane/test_monitor_poll_unavailable_recovery.py new file mode 100644 index 0000000000..79909dc108 --- /dev/null +++ b/tests/control_plane/test_monitor_poll_unavailable_recovery.py @@ -0,0 +1,57 @@ +"""An interrupted observation retains its owner before competing closeout.""" +from __future__ import annotations +import json +from pathlib import Path + +import pytest +from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime +from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted +from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection +from loopx.control_plane.testing.canary_harness import read_run_index +from loopx.todos import list_goal_todos +from test_monitor_poll_transaction_recovery import _fixture, _cli, _poll_events, _quota_receipt, GOAL, AGENT + +@pytest.mark.parametrize('provider', ['file', 'sqlite']) +@pytest.mark.parametrize('material', [False, True]) +def test_pending_poll_cannot_be_closed_as_unavailable( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, material: bool, +) -> None: + isolate_sqlite_runtime(tmp_path, monkeypatch) + registry, runtime, state, monitor, poll = _fixture(tmp_path) + todos = list_goal_todos(registry_path=registry, goal_id=GOAL)['todos'] + initialize_canonical_authority(runtime, GOAL, build_todo_runtime_shadow_projection( + goal_id=GOAL, todos=todos, handoff_mode='soft_claim'), state_path=state, provider=provider) + if not material: + args = list(poll) + args.remove('--material-change') + for flag in ('--next-agent-todo', '--next-action-kind', '--next-user-todo', '--next-user-task-class'): + index = args.index(flag) + del args[index:index + 2] + poll = tuple(args) + rc, result = _cli(registry, runtime, *poll, fault='quota_commit') + assert (rc, result) == (73, None) + committed = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL) + pending_path = _quota_receipt(runtime) + pending_bytes = pending_path.read_bytes() + assert json.loads(pending_bytes)['status'] == 'provider_pending' + refresh_rc, refresh = _cli(registry, runtime, 'refresh-state', '--goal-id', GOAL, + '--agent-id', AGENT, '--todo-id', monitor, '--turn-instance-id', 'synthetic-monitor-turn-1', + '--classification', 'synthetic_unavailable', '--delivery-batch-scale', 'single_surface', + '--delivery-outcome', 'outcome_gap', '--progress-result-class', 'blocked', + '--progress-blocker-id', 'blocker:synthetic-source', '--progress-evidence-id', 'evidence:synthetic-source', + '--vision-unchanged-reason', 'Source unavailable; retain the Monitor.', + '--no-global-sync', '--suppress-external-sinks') + assert refresh_rc != 0 and refresh['appended'] is False, refresh + assert pending_path.read_bytes() == pending_bytes + assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL) == committed + assert len(_poll_events(runtime)) == 0 + assert not any(row['classification'] == 'synthetic_unavailable' for row in read_run_index(runtime, GOAL)) + retry_rc, retry = _cli(registry, runtime, *poll) + assert retry_rc == 0, retry + assert len(_poll_events(runtime)) == 1 + assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL) == committed + index_after = read_run_index(runtime, GOAL) + replay_rc, replay = _cli(registry, runtime, *poll) + assert replay_rc == 0 and replay['replayed'] is True, replay + assert read_run_index(runtime, GOAL) == index_after + assert not any(row['classification'] == 'quota_slot_spent' for row in index_after) diff --git a/tests/control_plane/test_monitor_unavailable_closeout_cli.py b/tests/control_plane/test_monitor_unavailable_closeout_cli.py index 21ac1b3557..355cc67366 100644 --- a/tests/control_plane/test_monitor_unavailable_closeout_cli.py +++ b/tests/control_plane/test_monitor_unavailable_closeout_cli.py @@ -182,6 +182,20 @@ def cli(*args: str) -> tuple[int, dict]: assert read_rc == 0 and {k: current["todo"].get(k) for k in OBSERVATION_FIELDS} == fields assert replay_rc == 0 and frozen["idempotent_replay"] is True, frozen assert frozen["blocked_retry"] == wait + if scenario == "late_nonmaterial": + from loopx.control_plane.quota import monitor_poll as monitor_runtime + native = monitor_runtime._native_result + + def actor_optional_request(request): + request = {**request, "observation": {**request["observation"], "actor_agent_id": None}} + return native(request) + + # The native compatibility request permits an omitted actor; + # provider and settlement admission must use the same agent fallback. + with monkeypatch.context() as context: + context.setattr(monitor_runtime, "_native_result", actor_optional_request) + rc, missing_actor = cli(*poll_args, "--execute") + assert rc != 0 and missing_actor["error_code"] == "monitor_poll_admission_rejected", missing_actor rc, after_late = cli("todo", "list", "--goal-id", GOAL_ID) assert rc == 0, after_late monitor = next(todo for todo in after_late["todos"] if todo["todo_id"] == MONITOR_ID) diff --git a/tests/control_plane_ts/quota_monitor_poll_commit.test.ts b/tests/control_plane_ts/quota_monitor_poll_commit.test.ts index bdfdb6d506..b20ae70d34 100644 --- a/tests/control_plane_ts/quota_monitor_poll_commit.test.ts +++ b/tests/control_plane_ts/quota_monitor_poll_commit.test.ts @@ -8,15 +8,65 @@ import test from "node:test"; import { evaluateQuotaMonitorPollCommit, + hasMonitorPollTransaction, QUOTA_MONITOR_POLL_COMMIT_REQUEST_SCHEMA, QUOTA_LEASED_MONITOR_POLL_COMMIT_REQUEST_SCHEMA, } from "../../loopx/control_plane/quota/monitor_poll_commit.ts"; import { EffectRuntimeRequestError } from "../../loopx/control_plane/effect_runtime_errors.ts"; import { resolveTestPython } from "../../scripts/test-python.mjs"; import { monitorPollRequestHash } from "../../loopx/control_plane/coordination/todo_monitor_poll.ts"; +import { settlementIdentity } from "../../loopx/control_plane/effect_program.ts"; +import type { QuotaAccountingOwner } from "../../loopx/control_plane/quota/source_admission.ts"; +import { parseExactGoalRef } from "../../loopx/control_plane/goals/goal_instance_identity.ts"; const goalId = "monitor-native-goal"; +test("competing closeout reads pending and durable transactions under their Goal owner", async t => { + const runtime = await tempRuntime(t); + const agent = "codex-main-control", todo = "todo_public_monitor", turn = "owner-scoped-poll"; + const identity = settlementIdentity({goal_id: goalId, agent_id: agent, todo_id: todo, turn_instance_id: turn}); + const effect = `quota-monitor-poll:${goalId}:${agent}:${turn}:todo:${todo}`; + const params = request({phase: "preflight", runtime_root: runtime, execute: true, + effect_id: effect, turn_instance_id: turn, observation: observation({todo_id: todo, result_hash: "observed"})}); + assert.equal((await evaluateQuotaMonitorPollCommit(params)).status, "provider_required"); + const runs = join(runtime, "goals", goalId, "runs"); + const token = createHash("sha256").update(effect).digest("hex").slice(0, 24); + const path = join(runs, ".transactions", "quota-monitor-poll", `${token}.json`); + const pending = JSON.parse(await readFile(path, "utf8")); + const refA = {goal_id: goalId, goal_instance_id: `ginst_${"a".repeat(32)}`}; + const refB = {goal_id: goalId, goal_instance_id: `ginst_${"b".repeat(32)}`}; + const owner = (wire: typeof refA): QuotaAccountingOwner => { + const parsed = parseExactGoalRef(wire); + assert.equal(parsed.kind, "parsed"); + if (parsed.kind !== "parsed") throw new Error("fixture GoalRef rejected"); + const goalRef = parsed.value; + return {kind: "exact_source", goalRef, + admission: {registryPath: join(runtime, "registry.json"), plannedGoalRef: goalRef, authority: {}, + locks: [{role: "run_index", target: join(runs, "index.jsonl"), pid: process.pid, token: "fixture"}, + {role: "source_guard", target: join(runtime, "guard"), pid: process.pid, token: "fixture"}]}}; + }; + assert.equal(await hasMonitorPollTransaction(runtime, identity, {kind: "alias"}), true); + await assert.rejects(hasMonitorPollTransaction(runtime, identity, owner(refB)), /lacks GoalRef scope/); + for (const goalRef of [refA, refB]) { + await writeFile(path, JSON.stringify({...pending, goal_ref: goalRef})); + assert.equal(await hasMonitorPollTransaction(runtime, identity, owner(refA)), goalRef === refA); + assert.equal(await hasMonitorPollTransaction(runtime, identity, owner(refB)), goalRef === refB); + assert.equal(await hasMonitorPollTransaction(runtime, identity, {kind: "alias"}), false); + } + for (const status of ["prepared", "committed"]) { + const receipt = {...pending, schema_version: "quota_monitor_poll_commit_receipt_v0", status, + json_path: join(runs, `synthetic-quota-monitor-poll-${token}.json`), + markdown_path: join(runs, `synthetic-quota-monitor-poll-${token}.md`), record: {}, + index_record: {goal_id: goalId, agent_id: agent, todo_id: todo, goal_ref: refA}, + markdown: "Synthetic observation", payload: {}}; + await writeFile(path, JSON.stringify(receipt)); + assert.equal(await hasMonitorPollTransaction(runtime, identity, owner(refA)), true); + assert.equal(await hasMonitorPollTransaction(runtime, identity, owner(refB)), false); + await writeFile(path, JSON.stringify({...receipt, index_record: {...receipt.index_record, todo_id: "todo_other_monitor"}})); + assert.equal(await hasMonitorPollTransaction(runtime, identity, owner(refA)), false); + } +}); + function decision( extra: Record = {}, ): Record { From 465a052fe07110714c79dc29abf12da6cd361303 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 13 Oct 2026 02:19:14 +0800 Subject: [PATCH 07/10] docs(quota): preserve stored observation recovery and scope Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../quota-blocked-causal-closeout-v0.md | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/docs/reference/protocols/quota-blocked-causal-closeout-v0.md b/docs/reference/protocols/quota-blocked-causal-closeout-v0.md index 0d05c4b71c..0ea610a430 100644 --- a/docs/reference/protocols/quota-blocked-causal-closeout-v0.md +++ b/docs/reference/protocols/quota-blocked-causal-closeout-v0.md @@ -116,6 +116,16 @@ non-material poll on that same exact Turn binding before provider writeback, including dry-run provider plans. Use a freshly admitted Turn for a new observation; its poll cannot change the original frozen closeout. Genuine poll replay and independently admitted auxiliary Monitors remain valid. +A stored poll transaction may already have committed its provider observation +while its quota ledger is still pending. Unavailable closeout rejects that +competing outcome until the exact poll is recovered or its owner proves no +effect. Preserve the original transaction and business receipts; do not erase +them or close the attempt by claiming that its observation was unavailable. +New exact-source pending receipts retain their existing GoalRef. Foreign durable +or scoped pending receipts do not prove a current instance's observation. A +historical pending receipt lacking GoalRef in an exact-source context requires +owner reconciliation; it remains preserved. Alias-scope recovery keeps its +existing compatibility. Exact replay returns the frozen receipt, with no duplicate writeback or debit. After closing the original effect, reenter the current recovery host Turn to @@ -216,6 +226,12 @@ advancement 等待、material successor、完成或投递声明。旧 advancemen non-material poll 在 provider 写回前拒绝，dry-run 也不返回可写入计划。 新观察须使用重新准入的新 Turn；它的 poll 不改变原冻结结算。 真实 poll 的幂等重放及独立准入的 auxiliary Monitor 仍可执行。 +持久 poll 事务的 provider 观察可能已经提交，而 quota ledger 仍处于 pending。 +不可观察结算须拒绝这一竞争结果，先恢复原精确 poll 或由其 owner 证明无 effect。 +保留原事务及业务回执；不能删除它们或声称观察不可取得以关闭该尝试。 +新 exact-source pending 回执保留既有 GoalRef；其他实例的 durable 或带作用域的 +pending 回执不证明当前实例的观察。exact-source 环境下，历史 pending 缺 GoalRef +时须核对原 owner 并保留回执；alias 作用域沿用既有恢复兼容。 精确重放返回冻结回执，不重复写回或扣额；原 effect 关闭后，重新进入当前 recovery host Turn 可选择独立合法任务，不新造第三身份或重绑旧 Turn。缺准入能力、缺证或错绑 From c4aa543da58a16098dabf6e11c446fba29801b51 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 13 Oct 2026 02:28:23 +0800 Subject: [PATCH 08/10] fix(quota): bind poll readback to the accounting Agent Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../quota/monitor_poll_commit.ts | 4 +++- .../test_monitor_unavailable_closeout_cli.py | 19 ++++++++++--------- 2 files changed, 13 insertions(+), 10 deletions(-) diff --git a/loopx/control_plane/quota/monitor_poll_commit.ts b/loopx/control_plane/quota/monitor_poll_commit.ts index 050c3dbe7b..f0d820697d 100644 --- a/loopx/control_plane/quota/monitor_poll_commit.ts +++ b/loopx/control_plane/quota/monitor_poll_commit.ts @@ -567,7 +567,9 @@ interface Admission { async function readMonitorSettlement(request: MonitorRequest): Promise { const todoId = request.observation.settlement_todo_id ?? request.observation.todo_id; - const agentId = request.observation.actor_agent_id ?? request.decision.agent_id; + // The Turn/effect and run record belong to the decision Agent; provider actor + // metadata cannot select a different historical settlement to evade its fence. + const agentId = request.decision.agent_id; if (!request.runtime_root || !request.turn_instance_id || !agentId || !todoId) return null; const snapshot = await readQuotaSettlementSnapshot( diff --git a/tests/control_plane/test_monitor_unavailable_closeout_cli.py b/tests/control_plane/test_monitor_unavailable_closeout_cli.py index 355cc67366..fb486d8f69 100644 --- a/tests/control_plane/test_monitor_unavailable_closeout_cli.py +++ b/tests/control_plane/test_monitor_unavailable_closeout_cli.py @@ -186,16 +186,17 @@ def cli(*args: str) -> tuple[int, dict]: from loopx.control_plane.quota import monitor_poll as monitor_runtime native = monitor_runtime._native_result - def actor_optional_request(request): - request = {**request, "observation": {**request["observation"], "actor_agent_id": None}} - return native(request) - # The native compatibility request permits an omitted actor; - # provider and settlement admission must use the same agent fallback. - with monkeypatch.context() as context: - context.setattr(monitor_runtime, "_native_result", actor_optional_request) - rc, missing_actor = cli(*poll_args, "--execute") - assert rc != 0 and missing_actor["error_code"] == "monitor_poll_admission_rejected", missing_actor + # provider actor metadata cannot rebind the Turn's accounting Agent. + for actor in (None, "other-synthetic-actor"): + def actor_metadata_request(request): + request = {**request, "observation": {**request["observation"], "actor_agent_id": actor}} + return native(request) + + with monkeypatch.context() as context: + context.setattr(monitor_runtime, "_native_result", actor_metadata_request) + rc, actor_reply = cli(*poll_args, "--execute") + assert rc != 0 and actor_reply["error_code"] == "monitor_poll_admission_rejected", actor_reply rc, after_late = cli("todo", "list", "--goal-id", GOAL_ID) assert rc == 0, after_late monitor = next(todo for todo in after_late["todos"] if todo["todo_id"] == MONITOR_ID) From f44fe3d9e831c8137ec36a4d97a90caa79f351fc Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 13 Oct 2026 02:41:43 +0800 Subject: [PATCH 09/10] fix(quota): preserve pending accounting identity across provider actors Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../quota/monitor_poll_commit.ts | 13 ++++- .../test_monitor_poll_unavailable_recovery.py | 54 ++++++++++++++++--- .../quota_monitor_poll_commit.test.ts | 12 +++++ 3 files changed, 72 insertions(+), 7 deletions(-) diff --git a/loopx/control_plane/quota/monitor_poll_commit.ts b/loopx/control_plane/quota/monitor_poll_commit.ts index f0d820697d..f4e83a19e1 100644 --- a/loopx/control_plane/quota/monitor_poll_commit.ts +++ b/loopx/control_plane/quota/monitor_poll_commit.ts @@ -1783,7 +1783,18 @@ export async function hasMonitorPollTransaction( validateReceiptScope(receipt, runsDir, join(runsDir, "index.jsonl")); const facts = receipt.status === "provider_pending" ? receipt.provider_plan : receipt.index_record; if (receipt.effect_id !== effectId || facts.goal_id !== identity.goal_id || - facts.agent_id !== identity.agent_id || facts.todo_id !== identity.todo_id) continue; + facts.todo_id !== identity.todo_id) continue; + if (receipt.status === "provider_pending") { + // The provider plan names its actor, not the Turn's accounting Agent. + // Legacy pending receipts retain this exact accounting effect identity. + if (receipt.schema_version === MONITOR_PENDING_ADMISSION_SCHEMA && + (receipt.admitted_decision.goal_id !== identity.goal_id || + receipt.admitted_decision.agent_id !== identity.agent_id)) { + throw new EffectRuntimeRequestError( + "stored Monitor admission conflicts with its accounting identity", "malformed_transaction_receipt", + ); + } + } else if (facts.agent_id !== identity.agent_id) continue; const goalRef = receipt.status === "provider_pending" ? receipt.goal_ref : facts.goal_ref; if (receipt.status === "provider_pending" && goalRef === undefined && owner.kind !== "alias") { throw new EffectRuntimeRequestError( diff --git a/tests/control_plane/test_monitor_poll_unavailable_recovery.py b/tests/control_plane/test_monitor_poll_unavailable_recovery.py index 79909dc108..64aab772fe 100644 --- a/tests/control_plane/test_monitor_poll_unavailable_recovery.py +++ b/tests/control_plane/test_monitor_poll_unavailable_recovery.py @@ -1,6 +1,9 @@ """An interrupted observation retains its owner before competing closeout.""" from __future__ import annotations import json +import os +import subprocess +import sys from pathlib import Path import pytest @@ -9,16 +12,51 @@ from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection from loopx.control_plane.testing.canary_harness import read_run_index from loopx.todos import list_goal_todos -from test_monitor_poll_transaction_recovery import _fixture, _cli, _poll_events, _quota_receipt, GOAL, AGENT +from test_monitor_poll_transaction_recovery import _fixture, _cli, _poll_events, _quota_receipt, GOAL, AGENT, REPO + +_ACTOR_POLL_WRAPPER = """ +import os, runpy, sys +import loopx.control_plane.quota.monitor_poll as monitor +original = monitor._native_result +def actor_request(request): + request = {**request, 'observation': {**request['observation'], + 'actor_agent_id': os.environ['SYNTHETIC_MONITOR_ACTOR']}} + if os.environ['SYNTHETIC_QUOTA_FAULT'] == '1' and request.get('phase') == 'commit' and request.get('provider_receipt'): + os._exit(73) + return original(request) +monitor._native_result = actor_request +sys.argv = ['loopx.cli', *sys.argv[1:]] +runpy.run_module('loopx.cli', run_name='__main__') +""" + + +def _poll_cli(registry: Path, runtime: Path, poll: tuple[str, ...], actor: str | None, + *, fault: bool = False) -> tuple[int, dict | None]: + if actor is None: + return _cli(registry, runtime, *poll, fault='quota_commit' if fault else None) + result = subprocess.run([sys.executable, '-c', _ACTOR_POLL_WRAPPER, '--registry', str(registry), + '--format', 'json', '--runtime-root', str(runtime), *poll], cwd=REPO, + env={**os.environ, 'LOOPX_USAGE_PING': '0', 'SYNTHETIC_MONITOR_ACTOR': actor, + 'SYNTHETIC_QUOTA_FAULT': '1' if fault else '0'}, + capture_output=True, text=True, check=False, timeout=60) + return result.returncode, json.loads(result.stdout) if result.stdout.strip() else None @pytest.mark.parametrize('provider', ['file', 'sqlite']) @pytest.mark.parametrize('material', [False, True]) +@pytest.mark.parametrize('actor', [None, 'other-synthetic-observer']) def test_pending_poll_cannot_be_closed_as_unavailable( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, material: bool, + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, material: bool, actor: str | None, ) -> None: isolate_sqlite_runtime(tmp_path, monkeypatch) registry, runtime, state, monitor, poll = _fixture(tmp_path) todos = list_goal_todos(registry_path=registry, goal_id=GOAL)['todos'] + if actor is not None: + data = json.loads(registry.read_text()) + data['goals'][0]['coordination']['registered_agents'].append(actor) + registry.write_text(json.dumps(data)) + for todo in todos: + if todo['todo_id'] == monitor: + todo['claimed_by'] = None initialize_canonical_authority(runtime, GOAL, build_todo_runtime_shadow_projection( goal_id=GOAL, todos=todos, handoff_mode='soft_claim'), state_path=state, provider=provider) if not material: @@ -28,12 +66,16 @@ def test_pending_poll_cannot_be_closed_as_unavailable( index = args.index(flag) del args[index:index + 2] poll = tuple(args) - rc, result = _cli(registry, runtime, *poll, fault='quota_commit') + rc, result = _poll_cli(registry, runtime, poll, actor, fault=True) assert (rc, result) == (73, None) committed = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL) pending_path = _quota_receipt(runtime) pending_bytes = pending_path.read_bytes() - assert json.loads(pending_bytes)['status'] == 'provider_pending' + pending = json.loads(pending_bytes) + assert pending['status'] == 'provider_pending' + assert pending['admitted_decision']['agent_id'] == AGENT + assert pending['provider_plan']['agent_id'] == (actor or AGENT) + assert next(todo for todo in committed['todos'] if todo['todo_id'] == monitor)['last_actor_agent_id'] == (actor or AGENT) refresh_rc, refresh = _cli(registry, runtime, 'refresh-state', '--goal-id', GOAL, '--agent-id', AGENT, '--todo-id', monitor, '--turn-instance-id', 'synthetic-monitor-turn-1', '--classification', 'synthetic_unavailable', '--delivery-batch-scale', 'single_surface', @@ -46,12 +88,12 @@ def test_pending_poll_cannot_be_closed_as_unavailable( assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL) == committed assert len(_poll_events(runtime)) == 0 assert not any(row['classification'] == 'synthetic_unavailable' for row in read_run_index(runtime, GOAL)) - retry_rc, retry = _cli(registry, runtime, *poll) + retry_rc, retry = _poll_cli(registry, runtime, poll, actor) assert retry_rc == 0, retry assert len(_poll_events(runtime)) == 1 assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL) == committed index_after = read_run_index(runtime, GOAL) - replay_rc, replay = _cli(registry, runtime, *poll) + replay_rc, replay = _poll_cli(registry, runtime, poll, actor) assert replay_rc == 0 and replay['replayed'] is True, replay assert read_run_index(runtime, GOAL) == index_after assert not any(row['classification'] == 'quota_slot_spent' for row in index_after) diff --git a/tests/control_plane_ts/quota_monitor_poll_commit.test.ts b/tests/control_plane_ts/quota_monitor_poll_commit.test.ts index b20ae70d34..7ce8c52752 100644 --- a/tests/control_plane_ts/quota_monitor_poll_commit.test.ts +++ b/tests/control_plane_ts/quota_monitor_poll_commit.test.ts @@ -46,6 +46,18 @@ test("competing closeout reads pending and durable transactions under their Goal {role: "source_guard", target: join(runtime, "guard"), pid: process.pid, token: "fixture"}]}}; }; assert.equal(await hasMonitorPollTransaction(runtime, identity, {kind: "alias"}), true); + // Provider actor metadata does not change the pending accounting Agent. + const alternateActor = {...pending, provider_plan: {...pending.provider_plan, agent_id: "other-observer"}}; + await writeFile(path, JSON.stringify(alternateActor)); + assert.equal(await hasMonitorPollTransaction(runtime, identity, {kind: "alias"}), true); + await writeFile(path, JSON.stringify({...alternateActor, + admitted_decision: {...pending.admitted_decision, agent_id: "other-accounting-agent"}})); + await assert.rejects(hasMonitorPollTransaction(runtime, identity, {kind: "alias"}), /accounting identity/); + const legacyActor = {...alternateActor, schema_version: "quota_monitor_poll_commit_receipt_v0"}; + delete legacyActor.admitted_decision; + await writeFile(path, JSON.stringify(legacyActor)); + assert.equal(await hasMonitorPollTransaction(runtime, identity, {kind: "alias"}), true); + await writeFile(path, JSON.stringify(pending)); await assert.rejects(hasMonitorPollTransaction(runtime, identity, owner(refB)), /lacks GoalRef scope/); for (const goalRef of [refA, refB]) { await writeFile(path, JSON.stringify({...pending, goal_ref: goalRef})); From 8d7b0f127c06dd2fde393f0fe9f98341bbee7a90 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 13 Oct 2026 04:44:14 +0800 Subject: [PATCH 10/10] test(runtime): diagnose copied locator retirement only on failure Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../runtime_retirement_diagnostics.py | 128 ++++++++++++++++++ .../test_effect_runtime_integration.py | 55 +++++++- 2 files changed, 181 insertions(+), 2 deletions(-) create mode 100644 tests/control_plane/runtime_retirement_diagnostics.py diff --git a/tests/control_plane/runtime_retirement_diagnostics.py b/tests/control_plane/runtime_retirement_diagnostics.py new file mode 100644 index 0000000000..300fdd5358 --- /dev/null +++ b/tests/control_plane/runtime_retirement_diagnostics.py @@ -0,0 +1,128 @@ +"""Failure-only observations for the copied runtime retirement fixture.""" +from __future__ import annotations + +import json +import os +from pathlib import Path + +from loopx.control_plane import effect_runtime + +PREFIX = "loopx_fixture_retirement:" +EVENTS = frozenset({"shutdown_requested", "close", "flush_enter", "flush_done", "locator_read_done", + "locator_read_error", "locator_unlink_enter", "locator_unlink_done", "locator_unlink_error", + "cleanup_finally", "lock_attempt", "lock_acquired", "lock_acquire_error", "lock_release_error", + "lock_release_result"}) +CODES = frozenset({"EACCES", "EPERM", "ENOENT", "EEXIST", "EIO", "EBUSY", "ENOTEMPTY", "EBADF", + "ENOSPC", "EMFILE", "ENFILE", "ETIMEDOUT", "EINTR", "ENOTDIR", "EROFS", "mutation_lock_timeout"}) +TRACE = ''' +function fixtureTrace(event: string, details: Record = {}): void { + try { console.error("loopx_fixture_retirement:" + JSON.stringify({ + event, pid: process.pid, ms: performance.now(), ...details})); } catch {} +} +function fixtureCode(error: unknown): string | null { + const code = (error as {code?: unknown})?.code; + return typeof code === "string" ? code : null; +} +''' + + +def _replace(source: str, before: str, after: str) -> str: + assert source.count(before) == 1, "copied runtime diagnostic anchor changed" + return source.replace(before, after, 1) + + +def instrument_copied_runtime(root: Path) -> None: + """Observe the private fixture copy; production modules are never edited.""" + server = root / "effect_runtime_server.ts" + text = server.read_text() + text = _replace(text, 'const REQUEST_SCHEMA =', TRACE + '\nconst REQUEST_SCHEMA =') + for before, after in ( + ('shutdownRequested = true;', 'shutdownRequested = true; fixtureTrace("shutdown_requested");'), + ('server.on("close", () => {', 'server.on("close", () => { fixtureTrace("close", {pending: pendingRequests});'), + ('flushCompileCache();', 'fixtureTrace("flush_enter"); flushCompileCache(); fixtureTrace("flush_done");'), + ('published = JSON.parse(await readFile(infoPath, "utf8"));', + 'published = JSON.parse(await readFile(infoPath, "utf8")); fixtureTrace("locator_read_done");'), + ('} catch {\n return;\n }\n // A timed-out client', + '} catch (error) {\n fixtureTrace("locator_read_error", {code: fixtureCode(error)});\n return;\n }\n // A timed-out client'), + ('await rm(infoPath, { force: true });', + 'fixtureTrace("locator_unlink_enter");\n try { await rm(infoPath, { force: true }); fixtureTrace("locator_unlink_done"); }\n' + ' catch (error) { fixtureTrace("locator_unlink_error", {code: fixtureCode(error)}); throw error; }'), + ('})().finally(() => process.exit(0));', + '})().finally(() => { fixtureTrace("cleanup_finally"); process.exit(0); });'), + ): + text = _replace(text, before, after) + server.write_text(text) + io = root / "effect_runtime_io.ts" + text = io.read_text() + text = _replace(text, 'const MUTATION_LOCK_TIMEOUT_MS =', TRACE + '\nconst MUTATION_LOCK_TIMEOUT_MS =') + text = _replace(text, 'const lock = await acquireFileMutationLock(targetPath, process.pid, timeoutMs);', + 'fixtureTrace("lock_attempt");\n let lock: FileMutationLock;\n try { lock = await acquireFileMutationLock(targetPath, process.pid, timeoutMs); }\n' + ' catch (error) { fixtureTrace("lock_acquire_error", {code: fixtureCode(error)}); throw error; }\n fixtureTrace("lock_acquired");') + text = _replace(text, 'if (suppressErrors) return false;', + 'fixtureTrace("lock_release_error", {code: fixtureCode(error)});\n if (suppressErrors) return false;') + text = _replace(text, 'await releaseFileMutationLock(lock.targetPath, lock.token, null, true);', + 'const released = await releaseFileMutationLock(lock.targetPath, lock.token, null, true);\n' + ' fixtureTrace("lock_release_result", {released});') + io.write_text(text) + + +def trace_events(raw: bytes) -> list[dict]: + """Never publish raw stderr or unknown fields, event names, or error text.""" + result = [] + for line in raw.decode("utf-8", errors="replace").splitlines(): + if not line.startswith(PREFIX): + continue + try: + value = json.loads(line[len(PREFIX):]) + except ValueError: + continue + if not isinstance(value, dict) or not isinstance(value.get("event"), str) or value["event"] not in EVENTS: + continue + item = {"event": value["event"]} + for key in ("pid", "ms", "pending"): + number = value.get(key) + if type(number) in (int, float): + item[key] = number + if type(value.get("released")) is bool: + item["released"] = value["released"] + if "code" in value: + item["code"] = value["code"] if isinstance(value["code"], str) and value["code"] in CODES else None + result.append(item) + return result[-64:] + + +def file_state(path: Path, pid: int) -> dict: + """Return only existence, identity matches, liveness and numeric errno.""" + try: + value = json.loads(path.read_text()) + except OSError as error: + return {"read": "missing" if isinstance(error, FileNotFoundError) else "error", + "errno": error.errno, "winerror": getattr(error, "winerror", None)} + except ValueError: + return {"read": "invalid_json"} + owner = value.get("pid") if isinstance(value, dict) else None + return {"read": "ok", "owner_matches_runtime": owner == pid, + "owner_alive": effect_runtime._pid_is_alive(owner)} + + +def retained_stderr(process_args: list, options: dict, root: Path, captures: list[int]) -> None: + if str(root / "effect_runtime_server.ts") in process_args and "stderr" in options: + captures.append(os.dup(options["stderr"].fileno())) + + +def failure_trace(captures: list[int]) -> list[dict]: + events = [] + for descriptor in captures: + try: + # Observe only after the original assertion failed. dup shares the + # child's file offset: seek/read/restore can race a concurrent write. + # This bounded trace is evidence available now, not a complete log. + position = os.lseek(descriptor, 0, os.SEEK_CUR) + try: + os.lseek(descriptor, 0, os.SEEK_SET) + events.extend(trace_events(os.read(descriptor, 16 * 1024))) + finally: + os.lseek(descriptor, position, os.SEEK_SET) + except OSError as error: + events.append({"trace_read_errno": error.errno}) + return events diff --git a/tests/control_plane/test_effect_runtime_integration.py b/tests/control_plane/test_effect_runtime_integration.py index fd295a2a15..ce7e46ec64 100644 --- a/tests/control_plane/test_effect_runtime_integration.py +++ b/tests/control_plane/test_effect_runtime_integration.py @@ -11,6 +11,9 @@ from types import SimpleNamespace import pytest +from runtime_retirement_diagnostics import ( + failure_trace, file_state, instrument_copied_runtime, retained_stderr, trace_events, +) from loopx.control_plane import effect_runtime from loopx.control_plane.coordination.runtime_shadow import ( @@ -598,6 +601,7 @@ def test_identical_release_does_not_reuse_a_retired_lazy_loader( def test_runtime_decode_change_rotates_identity_and_starts_replacement( tmp_path: Path, monkeypatch, + request, ) -> None: source_root = Path(effect_runtime.__file__).resolve().parent staged_root = tmp_path / "control_plane" @@ -606,6 +610,24 @@ def test_runtime_decode_change_rotates_identity_and_starts_replacement( target = staged_root / relative target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source, target) + instrument_copied_runtime(staged_root) + captures: list[int] = [] + popen = effect_runtime.subprocess.Popen + + def close_captures(): + for descriptor in captures: + try: + os.close(descriptor) + except OSError: + pass + + request.addfinalizer(close_captures) + + def capture_runtime_stderr(args, *positional, **options): + retained_stderr(args, options, staged_root, captures) + return popen(args, *positional, **options) + + monkeypatch.setattr(effect_runtime.subprocess, "Popen", capture_runtime_stderr) runtime_dir = tmp_path / "runtime" monkeypatch.setattr(effect_runtime, "_control_plane_root", lambda: staged_root) @@ -626,14 +648,43 @@ def test_runtime_decode_change_rotates_identity_and_starts_replacement( assert original_info != replacement_info assert replacement_info.exists() + shutdown_started = time.monotonic() effect_runtime.effect_runtime_result("runtime.shutdown", {}, retry_safe=False) + shutdown_acknowledged = time.monotonic() deadline = time.monotonic() + 2 while ( original_info.exists() or replacement_info.exists() ) and time.monotonic() < deadline: time.sleep(0.025) - assert not original_info.exists() - assert not replacement_info.exists() + try: + assert not original_info.exists() + assert not replacement_info.exists() + except AssertionError: + try: + print("runtime retirement diagnostic: " + json.dumps({ + "shutdown_request_seconds": shutdown_started, + "shutdown_ack_seconds": shutdown_acknowledged, + "assertion_seconds": time.monotonic(), + "deadline_seconds": deadline, + "original_pid": int(original["pid"]), + "replacement_pid": int(replacement["pid"]), + "original": file_state(original_info, int(original["pid"])), + "replacement": file_state(replacement_info, int(replacement["pid"])), + "replacement_alive": effect_runtime._pid_is_alive(replacement["pid"]), + "replacement_lock": file_state(Path(str(replacement_info) + ".ts-effect.lock"), int(replacement["pid"])), + "events": failure_trace(captures), + "event_clock": "performance.now milliseconds per PID; not aligned with Python monotonic seconds", + "trace_limit": "first 16 KiB / last 64 events per stderr; shared-offset reads may race writes; later cleanup and Windows sharing unobserved", + })) + except Exception: + # Diagnostics must never replace or suppress the original failure. + pass + raise + + +def test_runtime_retirement_trace_filters_private_stderr() -> None: + raw = b'private path/token\nloopx_fixture_retirement:{"event":{}}\nloopx_fixture_retirement:{"event":"lock_release_error","code":"EPERM","token":"private","path":"private","pid":12,"ms":1}\n' + assert trace_events(raw) == [{"event": "lock_release_error", "pid": 12, "ms": 1, "code": "EPERM"}] def test_typed_write_rejects_cross_effect_overwrite(