diff --git a/docs/reference/protocols/quota-blocked-causal-closeout-v0.md b/docs/reference/protocols/quota-blocked-causal-closeout-v0.md index 9124ccf4ab..0ea610a430 100644 --- a/docs/reference/protocols/quota-blocked-causal-closeout-v0.md +++ b/docs/reference/protocols/quota-blocked-causal-closeout-v0.md @@ -111,6 +111,21 @@ 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. +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 @@ -207,6 +222,16 @@ 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 仍可执行。 +持久 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。缺准入能力、缺证或错绑 diff --git a/loopx/control_plane/quota/monitor_poll_commit.ts b/loopx/control_plane/quota/monitor_poll_commit.ts index 3e12392ab7..f4e83a19e1 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, @@ -16,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, @@ -563,9 +565,13 @@ 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; + // 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 || - !request.observation.actor_agent_id || !request.observation.settlement_todo_id) return null; + !agentId || !todoId) return null; const snapshot = await readQuotaSettlementSnapshot( request.runtime_root, request.goal_id, @@ -573,7 +579,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 +638,22 @@ async function auxiliaryMonitorAllowed( } async function admission(request: MonitorRequest, historicalAdmission = false): Promise { + const observation = request.observation; + 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, settlementTodo)) { + 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 +1584,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. @@ -1695,6 +1717,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, @@ -1745,6 +1768,45 @@ 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.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( + "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, @@ -2311,6 +2373,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; + }, ); } 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( 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..64aab772fe --- /dev/null +++ b/tests/control_plane/test_monitor_poll_unavailable_recovery.py @@ -0,0 +1,99 @@ +"""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 +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, 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, 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: + 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 = _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() + 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', + '--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 = _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 = _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/test_monitor_unavailable_closeout_cli.py b/tests/control_plane/test_monitor_unavailable_closeout_cli.py index 406ab543b8..fb486d8f69 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,84 @@ 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 + if scenario == "late_nonmaterial": + from loopx.control_plane.quota import monitor_poll as monitor_runtime + native = monitor_runtime._native_result + + # The native compatibility request permits an omitted 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) + 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 + # 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] = recovery_turn + 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 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..7ce8c52752 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,77 @@ 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); + // 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})); + 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 {