fix(quota): reject late polls after unavailable Monitor closeout

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。缺准入能力、缺证或错绑

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,17 +565,21 @@ interface Admission {
   external: boolean;
 }
 
-async function readAuxiliarySettlement(request: MonitorRequest): Promise<JsonObject | null> {
+async function readMonitorSettlement(request: MonitorRequest): Promise<JsonObject | null> {
+  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,
   );
   return readQuotaSettlementForAdmittedOwnerFromSnapshot({
     schema_version: QUOTA_SETTLEMENT_READBACK_REQUEST_SCHEMA,
     runtime_root: request.runtime_root, goal_id: request.goal_id,
-    agent_id: request.observation.actor_agent_id, todo_id: request.observation.settlement_todo_id,
+    agent_id: agentId, todo_id: todoId,
     turn_instance_id: request.turn_instance_id, replan_obligation_id: null,
     infer_turn_instance_id: false, allow_unbound_binding: false,
   }, request.owner, snapshot);
@@ -604,7 +610,7 @@ async function auxiliaryMonitorAllowed(
         todo.excluded_agents.some(value => 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<Admission> {
+  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<void> {
   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<MonitorReceipt | null> {
   }
 }
 
+/** 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<boolean> {
+  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);
       }

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<JsonObject> {
     : 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;
+    },
   );
 }

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<string, unknown> = {}): 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