From 2f6ae6b620b2c4a486a9755981d9fc052902e256 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Fri, 25 Sep 2026 18:31:58 +0800
Subject: [PATCH 1/4] fix(quota): reuse verified history and unify closeout
 receipt reads

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 .../quota/settlement_readback.ts              |  75 ++++-----
 .../quota/unsettled_host_turn.py              |  82 ++++------
 .../quota/unsettled_host_turn_recovery.ts     |  10 ++
 loopx/control_plane/rollout_receipt_log.ts    |  36 ++---
 .../runtime/receipt_log_snapshot.ts           | 144 ++++++++++++++++++
 .../test_prior_closeout_preflight_budget.py   |  50 ++++++
 .../receipt_log_snapshot.test.ts              | 113 ++++++++++++++
 .../unsettled_host_turn_recovery.test.ts      |  55 ++++++-
 8 files changed, 442 insertions(+), 123 deletions(-)
 create mode 100644 loopx/control_plane/runtime/receipt_log_snapshot.ts
 create mode 100644 tests/control_plane_ts/receipt_log_snapshot.test.ts

diff --git a/loopx/control_plane/quota/settlement_readback.ts b/loopx/control_plane/quota/settlement_readback.ts
index e6f5c5f2f2..1f60cdc369 100644
--- a/loopx/control_plane/quota/settlement_readback.ts
+++ b/loopx/control_plane/quota/settlement_readback.ts
@@ -1,4 +1,4 @@
-import { readFile } from "node:fs/promises";
+import { readReceiptLogSnapshot } from "../runtime/receipt_log_snapshot.ts";
 import { isAbsolute, join } from "node:path";
 
 import {
@@ -185,40 +185,15 @@ function decodeRequest(value: unknown): ReadbackRequest {
   };
 }
 
-async function readJsonLines(path: string, schemaVersion?: string): Promise<JsonObject[]> {
-  let content: string;
-  try {
-    content = await readFile(path, "utf8");
-  } catch (error) {
-    if (
-      error !== null &&
-      typeof error === "object" &&
-      "code" in error &&
-      error.code === "ENOENT"
-    ) {
-      return [];
-    }
-    throw error;
-  }
-  const records: JsonObject[] = [];
-  for (const [index, line] of content.split(/\r?\n/).entries()) {
-    if (!line.trim()) continue;
-    try {
-      const parsed: unknown = JSON.parse(line);
-      const record = jsonObject(parsed);
-      if (!record) throw new Error("record must be a JSON object");
-      if (schemaVersion !== undefined && record.schema_version !== schemaVersion) {
-        throw new Error(`schema must be ${schemaVersion}`);
-      }
-      records.push(record);
-    } catch {
-      throw new EffectRuntimeRequestError(
-        `settlement readback line ${index + 1} is malformed`,
-        "malformed_settlement_state",
-      );
-    }
+async function readRunReceipts(path: string): Promise<readonly JsonObject[]> {
+  const snapshot = await readReceiptLogSnapshot(path);
+  if (snapshot?.firstErrorLine != null) {
+    throw new EffectRuntimeRequestError(
+      `settlement readback line ${snapshot.firstErrorLine} is malformed`,
+      "malformed_settlement_state",
+    );
   }
-  return records;
+  return snapshot?.records ?? [];
 }
 
 function turnKey(
@@ -313,7 +288,7 @@ export async function readQuotaSettlementSnapshot(
     rolloutSnapshot === undefined
       ? readGoalRolloutEventSnapshot(runtimeRoot, goalId)
       : Promise.resolve(rolloutSnapshot),
-    readJsonLines(join(goalRoot, "runs", "index.jsonl")),
+    readRunReceipts(join(goalRoot, "runs", "index.jsonl")),
   ]);
   return indexSettlementSnapshot(
     runtimeRoot,
@@ -343,6 +318,27 @@ function indexedRuns(
   ) ?? [];
 }
 
+/** One committed-poll rule for settlement and prior-Turn closeout. */
+export function committedMonitorPollFromSnapshot(
+  snapshot: QuotaSettlementReadbackSnapshot,
+  identity: Pick<SettlementIdentity, "goal_id" | "agent_id" | "turn_instance_id" | "todo_id">,
+): JsonObject | null {
+  if (snapshot.goalId !== identity.goal_id) {
+    throw new EffectRuntimeRequestError("monitor poll snapshot scope mismatch");
+  }
+  const runs = snapshot.runsByTurn.get(
+    turnKey(identity.goal_id, identity.agent_id, identity.turn_instance_id)!,
+  ) ?? [];
+  return [...runs].reverse().find((run) =>
+    run.classification === "quota_monitor_poll" &&
+    optionalString(run.goal_id) === identity.goal_id &&
+    optionalString(run.agent_id) === identity.agent_id &&
+    optionalString(run.turn_instance_id) === identity.turn_instance_id &&
+    normalizeTodoId(run.todo_id) === identity.todo_id &&
+    isCommittedMonitorPollEffect(jsonObject(run.quota_monitor_poll_commit)?.effect_id, identity)
+  ) ?? null;
+}
+
 function spendCandidateRuns(
   snapshot: QuotaSettlementReadbackSnapshot,
   identity: SettlementIdentity,
@@ -990,14 +986,7 @@ function readQuotaSettlementFromRequest(
   const withWriteback = settlementBindReduce(identityResult, writeback);
   const settled = blockedNoSpend ? withWriteback : settlementBindReduce(withWriteback, spend);
   const terminalSettlement = settlementBindReduce(settled, terminalCloseout);
-  const monitorPoll = [...runs].reverse().find((run) =>
-    run.classification === "quota_monitor_poll" &&
-    optionalString(run.goal_id) === identity.goal_id &&
-    optionalString(run.agent_id) === identity.agent_id &&
-    optionalString(run.turn_instance_id) === identity.turn_instance_id &&
-    normalizeTodoId(run.todo_id) === identity.todo_id &&
-    isCommittedMonitorPollEffect(jsonObject(run.quota_monitor_poll_commit)?.effect_id, identity)
-  ) ?? null;
+  const monitorPoll = committedMonitorPollFromSnapshot(snapshot, identity);
   const nestedCausality = typeof receiptDetails.delivery_workspace_causality === "object" &&
       receiptDetails.delivery_workspace_causality !== null &&
       !Array.isArray(receiptDetails.delivery_workspace_causality)
diff --git a/loopx/control_plane/quota/unsettled_host_turn.py b/loopx/control_plane/quota/unsettled_host_turn.py
index 6b79028b2c..9a5d706948 100644
--- a/loopx/control_plane/quota/unsettled_host_turn.py
+++ b/loopx/control_plane/quota/unsettled_host_turn.py
@@ -1,7 +1,6 @@
 """Transport for the TypeScript-owned prior-host-Turn closeout recovery.
 
-Python reads two provider facts - the exact bound Todo and the committed
-monitor-poll receipt for the prior Turn the typed preflight names - hands them
+Python reads the exact bound Todo the typed preflight names, hands it
 to the typed transaction, and projects the typed verdict back into the
 existing public payload.  It owns no closeout policy: which prior Turn needs a
 closeout, whether its settlement validates, which closeout is accepted, and
@@ -15,15 +14,18 @@
 from pathlib import Path
 from typing import Any
 
-from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
+from ..effect_runtime import (
+    EffectRuntimeRejected,
+    EffectRuntimeResponseAmbiguous,
+    EffectRuntimeStartupError,
+    effect_runtime_result,
+)
 from ..scheduler.execution_context import SchedulerExecutionContextResolution
-from ..todos.contract import TODO_TASK_CLASS_MONITOR
 from ..todos.todo_semantics import todo_item_task_class
 from ..work_items.interaction_contract import (
     build_interaction_contract,
 )
 from .error_codes import HeartbeatReceiptIdentityConflictError
-from .monitor_poll import find_quota_monitor_poll_turn
 
 UNSETTLED_HOST_TURN_RECOVERY_SCHEMA_VERSION = "unsettled_host_turn_recovery_v0"
 
@@ -76,40 +78,6 @@ def _bound_todo_item(
     return dict(item)
 
 
-def _committed_monitor_poll_fact(
-    *,
-    runtime_root: Path,
-    goal_id: str,
-    agent_id: str,
-    todo_id: str | None,
-    prior_turn_instance_id: str,
-    todo_item: Mapping[str, Any] | None,
-) -> dict[str, Any]:
-    """Read the persisted monitor-poll receipt for one prior heartbeat Turn."""
-
-    # Only a monitor-bound Turn can carry this closeout, so the read is elided
-    # for every other Turn.  The transaction still owns the acceptance rule.
-    if (
-        not todo_id
-        or todo_item is None
-        or todo_item_task_class(todo_item) != TODO_TASK_CLASS_MONITOR
-    ):
-        return {}
-    receipt = find_quota_monitor_poll_turn(
-        runtime_root,
-        goal_id=goal_id,
-        agent_id=agent_id,
-        todo_id=todo_id,
-        turn_instance_id=prior_turn_instance_id,
-    )
-    if receipt is None:
-        return {}
-    commit_metadata = receipt.get("quota_monitor_poll_commit")
-    if not isinstance(commit_metadata, Mapping):
-        return {}
-    return {"effect_id": commit_metadata.get("effect_id")}
-
-
 def _todo_binding_facts(item: Mapping[str, Any] | None) -> dict[str, Any] | None:
     if item is None:
         return None
@@ -134,7 +102,7 @@ def _prior_closeout_preflight(
     goal_id: str,
     agent_id: str,
     current_turn_instance_id: str | None,
-) -> tuple[dict[str, Any], list[str]] | None:
+) -> tuple[dict[str, Any], list[str], dict[str, Any]] | None:
     """Ask the typed owner which prior Turn must still be closed out.
 
     The preflight reads the goal's persisted guards and the selected Turn's
@@ -154,6 +122,18 @@ def _prior_closeout_preflight(
             },
             timeout=PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_TIMEOUT_SECONDS,
         )
+    except EffectRuntimeResponseAmbiguous as exc:
+        # This method only reads receipts. A lost query response is not a
+        # possibly committed mutation, and must not send the operator hunting
+        # for a nonexistent preflight write receipt. Do not infer a verdict or
+        # automatically restart/retry the shared runtime.
+        raise EffectRuntimeStartupError(
+            f"Read-only {PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_METHOD} returned no "
+            f"verifiable response within {PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_TIMEOUT_SECONDS:g}s; "
+            "closeout state is unknown. Retry the query after checking runtime health; "
+            "the preflight itself performs no durable writes",
+            diagnostic_code="closeout_query_unavailable",
+        ) from exc
     except EffectRuntimeRejected as exc:
         # Keep the public diagnostic the identity rule has always published,
         # even though the rule now lives in the typed owner.
@@ -174,7 +154,13 @@ def _prior_closeout_preflight(
     missing_receipts = result.get("missing_receipts")
     if not isinstance(candidate, Mapping) or not isinstance(missing_receipts, list):
         raise RuntimeError("TypeScript closeout preflight result shape mismatch")
-    return dict(candidate), [str(name) for name in missing_receipts]
+    monitor_poll = result.get("committed_monitor_poll")
+    if monitor_poll is not None and not isinstance(monitor_poll, Mapping):
+        raise RuntimeError("TypeScript closeout monitor-poll fact shape mismatch")
+    return (
+        dict(candidate), [str(name) for name in missing_receipts],
+        dict(monitor_poll) if isinstance(monitor_poll, Mapping) else {},
+    )
 
 
 def _unsettled_host_turn_recovery(
@@ -195,7 +181,7 @@ def _unsettled_host_turn_recovery(
     )
     if preflight is None:
         return None
-    selected, missing_receipts = preflight
+    selected, missing_receipts, monitor_poll = preflight
     # A candidate carries exactly one binding: the Todo it must read, or the
     # autonomous replan obligation that has no Todo to read.
     todo_id = (
@@ -203,9 +189,8 @@ def _unsettled_host_turn_recovery(
         if selected.get("binding_kind") == "todo"
         else ""
     ) or None
-    prior_turn_id = str(selected.get("prior_turn_instance_id") or "")
     # The preflight named this Turn as the one whose bound facts decide the
-    # verdict, so these are the only provider reads this side still performs.
+    # verdict; this is the only provider read this side still performs.
     todo_item = _bound_todo_item(
         registry_path=registry_path,
         runtime_root=runtime_root,
@@ -215,14 +200,7 @@ def _unsettled_host_turn_recovery(
     binding_facts: dict[str, Any] = {
         "status": "read",
         "todo": _todo_binding_facts(todo_item),
-        "committed_monitor_poll": _committed_monitor_poll_fact(
-            runtime_root=runtime_root,
-            goal_id=goal_id,
-            agent_id=agent_id,
-            todo_id=todo_id,
-            prior_turn_instance_id=prior_turn_id,
-            todo_item=todo_item,
-        ),
+        "committed_monitor_poll": monitor_poll,
     }
     verdict = effect_runtime_result(
         UNSETTLED_HOST_TURN_RECOVERY_METHOD,
diff --git a/loopx/control_plane/quota/unsettled_host_turn_recovery.ts b/loopx/control_plane/quota/unsettled_host_turn_recovery.ts
index e03cb1b8f1..c892b37277 100644
--- a/loopx/control_plane/quota/unsettled_host_turn_recovery.ts
+++ b/loopx/control_plane/quota/unsettled_host_turn_recovery.ts
@@ -37,6 +37,7 @@ import {
   type HeartbeatReceiptFact,
 } from "./heartbeat_receipt_identity.ts";
 import {
+  committedMonitorPollFromSnapshot,
   QUOTA_SETTLEMENT_READBACK_REQUEST_SCHEMA,
   readQuotaSettlementFromSnapshot,
   readQuotaSettlementSnapshot,
@@ -267,9 +268,18 @@ export async function preflightPriorHostTurnCloseout(
     if (readback.found !== true || bundleFailed(readback, "spend")) {
       missingReceipts.push(SPEND_RECEIPT);
     }
+    const monitorPoll = selected.binding_kind === "todo"
+      ? committedMonitorPollFromSnapshot(settlementSnapshot, {
+        goal_id: request.goal_id, agent_id: request.agent_id,
+        turn_instance_id: selected.prior_turn_instance_id,
+        todo_id: selected.binding_id,
+      }) : null;
     return {
       schema_version: PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_RESULT_SCHEMA,
       status: "candidate",
+      committed_monitor_poll: monitorPoll === null ? null : {
+        effect_id: jsonObject(monitorPoll.quota_monitor_poll_commit)!.effect_id,
+      },
       turns_validated: turnsValidated,
       candidate: selected,
       missing_receipts: missingReceipts,
diff --git a/loopx/control_plane/rollout_receipt_log.ts b/loopx/control_plane/rollout_receipt_log.ts
index 3dc3e528a6..ed58a87984 100644
--- a/loopx/control_plane/rollout_receipt_log.ts
+++ b/loopx/control_plane/rollout_receipt_log.ts
@@ -7,12 +7,12 @@
  * receipts lives here so a reader cannot invent a second path rule or a
  * different tolerance for malformed lines.
  */
-import { readFile } from "node:fs/promises";
 import { relative, resolve, sep } from "node:path";
 
 import type { JsonObject } from "./effect_program.ts";
 import { EffectRuntimeRequestError } from "./effect_runtime_errors.ts";
-import { jsonObject, requireNonEmptyString } from "./runtime_decode.ts";
+import { requireNonEmptyString } from "./runtime_decode.ts";
+import { readReceiptLogSnapshot } from "./runtime/receipt_log_snapshot.ts";
 
 export const ROLLOUT_EVENT_SCHEMA_VERSION = "loopx_rollout_event_v0";
 export const HEARTBEAT_RECEIPT_EVENT_KIND = "quota_should_run";
@@ -79,31 +79,13 @@ export async function readGoalRolloutEventSnapshot(
   runtimeRoot: string,
   goalId: string,
 ): Promise<GoalRolloutEventSnapshot | null> {
-  let text: string;
-  try {
-    text = await readFile(goalRolloutEventLogPath(runtimeRoot, goalId), "utf8");
-  } catch (error) {
-    if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
-    throw error;
-  }
-  const events: JsonObject[] = [];
-  let firstStrictErrorLine: number | null = null;
-  for (const [index, line] of text.split(/\r?\n/).entries()) {
-    if (!line.trim()) continue;
-    try {
-      const event = jsonObject(JSON.parse(line));
-      if (
-        event === null ||
-        event.schema_version !== ROLLOUT_EVENT_SCHEMA_VERSION
-      ) {
-        throw new Error("rollout event has an unsupported schema");
-      }
-      events.push(event);
-    } catch {
-      firstStrictErrorLine ??= index + 1;
-    }
-  }
-  return {runtimeRoot, goalId, events, firstStrictErrorLine};
+  const snapshot = await readReceiptLogSnapshot(
+    goalRolloutEventLogPath(runtimeRoot, goalId), ROLLOUT_EVENT_SCHEMA_VERSION,
+  );
+  return snapshot === null ? null : {
+    runtimeRoot, goalId, events: snapshot.records,
+    firstStrictErrorLine: snapshot.firstErrorLine,
+  };
 }
 
 /** Return strict settlement input, or fail on the first malformed line. */
diff --git a/loopx/control_plane/runtime/receipt_log_snapshot.ts b/loopx/control_plane/runtime/receipt_log_snapshot.ts
new file mode 100644
index 0000000000..da0495f76e
--- /dev/null
+++ b/loopx/control_plane/runtime/receipt_log_snapshot.ts
@@ -0,0 +1,144 @@
+/** Disposable parsing acceleration; the bytes on disk remain the authority. */
+import { createHash } from "node:crypto";
+import { readFile } from "node:fs/promises";
+import { resolve } from "node:path";
+import { setImmediate } from "node:timers/promises";
+
+import type { JsonObject } from "../effect_program.ts";
+import { jsonObject } from "../runtime_decode.ts";
+
+export interface ReceiptLogSnapshot {
+  readonly records: readonly JsonObject[];
+  readonly firstErrorLine: number | null;
+}
+
+interface Prefix extends ReceiptLogSnapshot {
+  readonly bytes: number;
+  readonly digest: string;
+  readonly lines: number;
+}
+
+// Bound retained source volume and the number of histories. This is not a
+// promise about JS heap bytes: decoded objects can exceed their JSON size.
+const MAX_RETAINED_BYTES = 128 * 1024 * 1024;
+const MAX_RETAINED_LOGS = 4;
+const prefixes = new Map<string, Prefix>();
+let retainedBytes = 0;
+
+function digest(bytes: Buffer): string {
+  return createHash("sha256").update(bytes).digest("hex");
+}
+
+function forget(key: string): void {
+  const previous = prefixes.get(key);
+  if (previous) retainedBytes -= previous.bytes;
+  prefixes.delete(key);
+}
+
+function remember(key: string, prefix: Prefix): void {
+  forget(key);
+  if (prefix.bytes > MAX_RETAINED_BYTES) return;
+  while (prefixes.size >= MAX_RETAINED_LOGS ||
+    retainedBytes + prefix.bytes > MAX_RETAINED_BYTES) {
+    forget(prefixes.keys().next().value!);
+  }
+  prefixes.set(key, prefix);
+  retainedBytes += prefix.bytes;
+}
+
+/** `readonly` is erased; freeze nested values so a caller cannot poison reuse. */
+function freezeRecord(record: JsonObject): JsonObject {
+  const pending: object[] = [record];
+  while (pending.length > 0) {
+    const value = pending.pop()!;
+    for (const child of Object.values(value)) {
+      if (child !== null && typeof child === "object") pending.push(child);
+    }
+    Object.freeze(value);
+  }
+  return record;
+}
+
+function decodeLine(line: string, schemaVersion?: string): JsonObject {
+  const record = jsonObject(JSON.parse(line));
+  if (record === null || (schemaVersion !== undefined &&
+    record.schema_version !== schemaVersion)) {
+    throw new Error("receipt log row is malformed");
+  }
+  return freezeRecord(record);
+}
+
+/**
+ * Read fresh bytes on EVERY invocation. Only reuse a parsed newline-terminated
+ * prefix after hashing those exact bytes; size/mtime/inode are not evidence of
+ * unchanged history. Rewrites, rotation, truncation and same-size edits all
+ * remain observable, including edits to an old identity-conflicting receipt.
+ *
+ * Never cache an unterminated tail: an append can finish its UTF-8 character or
+ * JSON token. The tail still participates in this read, including strict errors.
+ * Cache loss/eviction/restart affects cost only, never the returned facts.
+ */
+export async function readReceiptLogSnapshot(
+  path: string,
+  schemaVersion?: string,
+): Promise<ReceiptLogSnapshot | null> {
+  const key = JSON.stringify([resolve(path), schemaVersion ?? null]);
+  let bytes: Buffer;
+  try {
+    bytes = await readFile(path);
+  } catch (error) {
+    forget(key);
+    if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
+    throw error;
+  }
+  const completeBytes = bytes.lastIndexOf(10) + 1;
+  const prior = prefixes.get(key);
+  const reusable = prior !== undefined && prior.bytes <= completeBytes &&
+    digest(bytes.subarray(0, prior.bytes)) === prior.digest ? prior : null;
+  let prefix: Prefix;
+  if (reusable !== null && reusable.bytes === completeBytes) {
+    prefix = reusable;
+  } else {
+    const records = reusable ? [...reusable.records] : [];
+    let firstErrorLine = reusable?.firstErrorLine ?? null;
+    let lines = reusable?.lines ?? 0;
+    let offset = reusable?.bytes ?? 0;
+    let batchStart = offset;
+    while (offset < completeBytes) {
+      const end = bytes.indexOf(10, offset);
+      const line = bytes.subarray(offset, end).toString("utf8");
+      lines += 1;
+      if (line.trim()) {
+        try {
+          records.push(decodeLine(line, schemaVersion));
+        } catch {
+          firstErrorLine ??= lines;
+        }
+      }
+      offset = end + 1;
+      // Cold/rebuilt histories must let the shared server service other sockets.
+      if (offset - batchStart >= 1024 * 1024) {
+        await setImmediate();
+        batchStart = offset;
+      }
+    }
+    prefix = Object.freeze({
+      records: Object.freeze(records), firstErrorLine, lines,
+      bytes: completeBytes, digest: digest(bytes.subarray(0, completeBytes)),
+    });
+  }
+  remember(key, prefix);
+  const tail = bytes.subarray(completeBytes).toString("utf8");
+  if (!tail.trim()) return prefix;
+  try {
+    return Object.freeze({
+      records: Object.freeze([...prefix.records, decodeLine(tail, schemaVersion)]),
+      firstErrorLine: prefix.firstErrorLine,
+    });
+  } catch {
+    return Object.freeze({
+      records: prefix.records,
+      firstErrorLine: prefix.firstErrorLine ?? prefix.lines + 1,
+    });
+  }
+}
diff --git a/tests/control_plane/test_prior_closeout_preflight_budget.py b/tests/control_plane/test_prior_closeout_preflight_budget.py
index cdbc91dc93..07f38f3a8a 100644
--- a/tests/control_plane/test_prior_closeout_preflight_budget.py
+++ b/tests/control_plane/test_prior_closeout_preflight_budget.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import argparse
+import json
 from pathlib import Path
 from unittest.mock import patch
 
@@ -13,6 +14,7 @@
 from loopx.cli_commands.quota_failure_report import quota_failure_payload
 from loopx.control_plane.effect_runtime import (
     EffectRuntimeRejected,
+    EffectRuntimeResponseAmbiguous,
     EffectRuntimeStartupError,
 )
 from loopx.control_plane.quota.unsettled_host_turn import (
@@ -78,6 +80,54 @@ def rejected(method, params, **kwargs):
     assert type(raised.value).__name__ == "HeartbeatReceiptIdentityConflictError"
 
 
+def test_a_lost_preflight_response_is_an_unknown_query_not_an_ambiguous_write():
+    with patch.object(
+        unsettled_host_turn, "effect_runtime_result",
+        side_effect=EffectRuntimeResponseAmbiguous(
+            PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_METHOD, timeout=5,
+        ),
+    ) as request:
+        with pytest.raises(EffectRuntimeStartupError) as raised:
+            _preflight()
+    assert request.call_count == 1
+    assert raised.value.diagnostic_code == "closeout_query_unavailable"
+    assert "closeout state is unknown" in str(raised.value)
+    assert "may have committed" not in str(raised.value)
+    assert isinstance(raised.value.__cause__, EffectRuntimeResponseAmbiguous)
+
+
+def test_native_preflight_carries_monitor_evidence_through_python_adapter(tmp_path):
+    """Real files and RPC; Python only supplies the exact current Todo fact."""
+    goal, agent, turn, todo = "goal-fixture", "agent-fixture", "old-turn", "todo_monitor"
+    goal_root = tmp_path / "goals" / goal
+    (goal_root / "runs").mkdir(parents=True)
+    (goal_root / "rollout-event-log.jsonl").write_text(json.dumps({
+        "schema_version": "loopx_rollout_event_v0", "event_kind": "quota_should_run",
+        "goal_id": goal, "agent_id": agent, "run_id": turn,
+        "details": {"closeout_required": True, "todo_id": todo},
+    }) + "\n", encoding="utf-8")
+    (goal_root / "runs" / "index.jsonl").write_text(json.dumps({
+        "classification": "quota_monitor_poll", "goal_id": goal, "agent_id": agent,
+        "turn_instance_id": turn, "todo_id": todo,
+        "quota_monitor_poll_commit": {
+            "effect_id": f"quota-monitor-poll:{goal}:{agent}:{turn}:todo:{todo}",
+        },
+    }) + "\n", encoding="utf-8")
+    with (
+        patch.object(unsettled_host_turn, "_bound_todo_item", return_value={
+            "todo_id": todo, "task_class": "continuous_monitor", "status": "open",
+        }),
+        patch(
+            "loopx.control_plane.quota.monitor_poll.find_quota_monitor_poll_turn",
+            side_effect=AssertionError("Python must not scan the run log again"),
+        ),
+    ):
+        assert unsettled_host_turn._unsettled_host_turn_recovery(
+            registry_path=tmp_path / "registry.json", runtime_root=tmp_path,
+            goal_id=goal, agent_id=agent, current_turn_instance_id="new-turn",
+        ) is None
+
+
 def test_a_runtime_timeout_names_the_method_and_the_budget():
     """A caller cannot repair "request failed"; it can repair a budget."""
 
diff --git a/tests/control_plane_ts/receipt_log_snapshot.test.ts b/tests/control_plane_ts/receipt_log_snapshot.test.ts
new file mode 100644
index 0000000000..a5119950d2
--- /dev/null
+++ b/tests/control_plane_ts/receipt_log_snapshot.test.ts
@@ -0,0 +1,113 @@
+import assert from "node:assert/strict";
+import { appendFile, mkdtemp, rename, rm, stat, utimes, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import test from "node:test";
+
+import { readReceiptLogSnapshot as read } from "../../loopx/control_plane/runtime/receipt_log_snapshot.ts";
+
+async function fixture(run: (path: string) => Promise<void>): Promise<void> {
+  const root = await mkdtemp(join(tmpdir(), "loopx-receipt-history-"));
+  try { await run(join(root, "history.jsonl")); }
+  finally { await rm(root, {recursive: true, force: true}); }
+}
+
+test("fresh disk bytes own reuse: append retains old records without changing old snapshots", async () => {
+  await fixture(async (path) => {
+    await writeFile(path, '{"turn":"old","nested":{"value":1}}\n');
+    const before = (await read(path))!;
+    const again = (await read(path))!;
+    assert.equal(again.records, before.records);
+    assert.throws(() => { before.records[0].turn = "poison"; }, TypeError);
+    assert.throws(() => { (before.records[0].nested as {value: number}).value = 2; }, TypeError);
+    await appendFile(path, '{"turn":"new"}\n');
+    const after = (await read(path))!;
+    assert.deepEqual(after.records.map((r) => r.turn), ["old", "new"]);
+    assert.equal(after.records[0], before.records[0]);
+    assert.equal(before.records.length, 1);
+    assert.equal(after.firstErrorLine, null);
+  });
+});
+
+test("same-size rewrite with restored mtime cannot hide a changed old receipt", async () => {
+  await fixture(async (path) => {
+    await writeFile(path, '{"binding":"aaa"}\n');
+    const metadata = await stat(path);
+    await read(path);
+    await writeFile(path, '{"binding":"bbb"}\n');
+    await utimes(path, metadata.atime, metadata.mtime);
+    assert.deepEqual((await read(path))!.records, [{binding: "bbb"}]);
+    // Growth is not proof of append-only history either.
+    await writeFile(path, '{"binding":"ccc"}\n{"later":true}\n');
+    assert.deepEqual((await read(path))!.records, [{binding: "ccc"}, {later: true}]);
+  });
+});
+
+test("truncation, replacement, deletion and recreation discard stale facts", async () => {
+  await fixture(async (path) => {
+    await writeFile(path, '{"old":true}\n{"old":2}\n');
+    await read(path);
+    await writeFile(path, "");
+    assert.deepEqual((await read(path))!.records, []);
+    await writeFile(path + ".next", '{"replacement":true}\n');
+    await rename(path + ".next", path);
+    assert.deepEqual((await read(path))!.records, [{replacement: true}]);
+    await rm(path);
+    assert.equal(await read(path), null);
+    await writeFile(path, '{"recreated":true}\n');
+    assert.deepEqual((await read(path))!.records, [{recreated: true}]);
+  });
+});
+
+test("unfinished JSON and split UTF-8 tails are reparsed when completed", async () => {
+  await fixture(async (path) => {
+    const full = Buffer.from('{"label":"界"}\n');
+    const split = full.indexOf(Buffer.from("界")) + 1;
+    await writeFile(path, full.subarray(0, split));
+    assert.equal((await read(path))!.firstErrorLine, 1);
+    await appendFile(path, full.subarray(split));
+    assert.deepEqual(await read(path).then((s) => s!.records), [{label: "界"}]);
+    assert.equal((await read(path))!.firstErrorLine, null);
+    await appendFile(path, '{"last":1}');
+    const unterminated = (await read(path))!;
+    assert.equal(unterminated.records.length, 2);
+    await appendFile(path, 'broken\n');
+    const malformed = (await read(path))!;
+    assert.equal(malformed.records.length, 1);
+    assert.equal(malformed.firstErrorLine, 2);
+    assert.equal(unterminated.records.length, 2);
+  });
+});
+
+test("schema, malformed line numbers and tolerant valid rows survive reuse", async () => {
+  await fixture(async (path) => {
+    await writeFile(path, '\r\n {"schema_version":"v1"}\r\n[]\n{"schema_version":"v2"}\n');
+    const strict = (await read(path, "v1"))!;
+    assert.deepEqual(strict.records, [{schema_version: "v1"}]);
+    assert.equal(strict.firstErrorLine, 3);
+    assert.equal((await read(path))!.records.length, 2);
+    await appendFile(path, '{"schema_version":"v1","new":true}\n');
+    const appended = (await read(path, "v1"))!;
+    assert.equal(appended.firstErrorLine, 3);
+    assert.equal(appended.records.length, 2);
+    await writeFile(path, '{"schema_version":"v1","repaired":true}\n');
+    assert.equal((await read(path, "v1"))!.firstErrorLine, null);
+  });
+});
+
+test("eviction and concurrent cold readers affect cost, never evidence", async () => {
+  await fixture(async (path) => {
+    const row = JSON.stringify({payload: "x".repeat(2048)}) + "\n";
+    await writeFile(path, row.repeat(1024));
+    const [a, b] = await Promise.all([read(path), read(path)]);
+    assert.deepEqual(a!.records, b!.records);
+    for (let index = 0; index < 5; index++) {
+      const other = path + index;
+      await writeFile(other, '{"other":true}\n');
+      await read(other);
+    }
+    const evicted = (await read(path))!;
+    assert.deepEqual(evicted.records, a!.records);
+    assert.notEqual(evicted.records, a!.records);
+  });
+});
diff --git a/tests/control_plane_ts/unsettled_host_turn_recovery.test.ts b/tests/control_plane_ts/unsettled_host_turn_recovery.test.ts
index 82fae03862..e65f268922 100644
--- a/tests/control_plane_ts/unsettled_host_turn_recovery.test.ts
+++ b/tests/control_plane_ts/unsettled_host_turn_recovery.test.ts
@@ -1,5 +1,5 @@
 import assert from "node:assert/strict";
-import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
+import { appendFile, mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { performance } from "node:perf_hooks";
@@ -478,6 +478,59 @@ test("settled history is read and indexed once instead of rescanned per Turn", a
   }
 });
 
+test("warm history still discovers an appended older identity conflict and repaired rewrite", async () => {
+  const runtime = await runtimeWithSettledTurns(1000);
+  const path = join(runtime.root, "goals", GOAL, "rollout-event-log.jsonl");
+  try {
+    assert.equal((await preflight(runtime.root)).status, "none");
+    await appendFile(path, JSON.stringify(
+      receipt("turn-0000", closeoutRequired("turn-0000", "todo_changed")),
+    ) + "\n");
+    await assert.rejects(() => preflight(runtime.root), /conflicting settlement identities/);
+    // An authoritative repair/rewrite must also retire the cached conflict.
+    await writeFile(path, JSON.stringify(receipt("repaired-turn", {closeout_required: false})) + "\n");
+    const repaired = await preflight(runtime.root);
+    assert.equal(repaired.status, "none");
+    assert.equal(repaired.turns_validated, 1);
+  } finally { await runtime.close(); }
+});
+
+test("monitor closeout comes from exact committed history, not a second Python scan", async () => {
+  const todo = "todo_monitor";
+  const runtime = await runtimeWith([receipt("turn-a", closeoutRequired("turn-a", todo))]);
+  const runs = join(runtime.root, "goals", GOAL, "runs");
+  const effectId = `quota-monitor-poll:${GOAL}:${AGENT}:turn-a:todo:${todo}`;
+  const poll = {
+    classification: "quota_monitor_poll", goal_id: GOAL, agent_id: AGENT,
+    turn_instance_id: "turn-a", todo_id: todo,
+    quota_monitor_poll_commit: {effect_id: effectId},
+  };
+  try {
+    await mkdir(runs);
+    assert.equal((await preflight(runtime.root)).committed_monitor_poll, null);
+    for (const override of [
+      {goal_id: "other-goal"}, {agent_id: "other-agent"},
+      {turn_instance_id: "other-turn"}, {todo_id: "todo_other"},
+      {quota_monitor_poll_commit: {effect_id: "uncommitted"}},
+    ]) {
+      await writeFile(join(runs, "index.jsonl"), JSON.stringify({...poll, ...override}) + "\n");
+      assert.equal((await preflight(runtime.root)).committed_monitor_poll, null);
+    }
+    await writeFile(join(runs, "index.jsonl"), JSON.stringify(poll) + "\n");
+    // A later observation without a commit cannot revoke a durable exact poll.
+    await appendFile(join(runs, "index.jsonl"), JSON.stringify({...poll, quota_monitor_poll_commit: {}}) + "\n");
+    const result = await preflight(runtime.root);
+    assert.deepEqual(result.committed_monitor_poll, {effect_id: effectId});
+    const verdict = reduce(candidateFrom(result), result.missing_receipts as string[], READ_TODO(
+      {...ADVANCEMENT_OPEN, task_class: "continuous_monitor"},
+      result.committed_monitor_poll as JsonObject,
+    ));
+    assert.equal(verdict.accepted_closeout, "exact_committed_quota_monitor_poll");
+    await appendFile(join(runs, "index.jsonl"), "malformed\n");
+    await assert.rejects(() => preflight(runtime.root), /malformed/);
+  } finally { await runtime.close(); }
+});
+
 test("an unsettled Turn cannot be decided without its bound Todo facts", async () => {
   const runtime = await runtimeWith([
     receipt("turn-a", closeoutRequired("turn-a", "todo_alpha")),

From ba58ab2ce4c322bd5066af91132c118a0f3425df Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Fri, 25 Sep 2026 18:32:16 +0800
Subject: [PATCH 2/4] docs(rfc): account for closeout liveness without moving
 cutover gates

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 ...26-09-24-default-cutover-reconciliation.md | 38 ++++++++++++++++++-
 ...24-default-cutover-reconciliation.zh-CN.md | 30 ++++++++++++++-
 ...shared-goal-authority-state-provider-v0.md |  5 +++
 ...-goal-authority-state-provider-v0.zh-CN.md |  4 ++
 .../typescript-control-plane-migration-v0.md  |  6 +++
 ...script-control-plane-migration-v0.zh-CN.md |  4 ++
 6 files changed, 83 insertions(+), 4 deletions(-)

diff --git a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-default-cutover-reconciliation.md b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-default-cutover-reconciliation.md
index bbddf9433f..7ef0be0b3e 100644
--- a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-default-cutover-reconciliation.md
+++ b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-default-cutover-reconciliation.md
@@ -1,8 +1,9 @@
 # Default cutover: reconciled implementation frontier
 
-- Baseline: `37bbaec79` on `main`, 2026-09-25; open PR states are a snapshot, not merge promises.
+- Baseline: `41ba6f4d9` on `main`, 2026-09-25; open PR states are a snapshot, not merge promises.
 - Owners: overall roadmap #4574 R5/G2; shared authority L2–L9/D1–D3; TS migration T1–T4.
-- Delivery: current registration admission, complete saved migration intent and truthful fence recovery.
+- Delivered #5040: current registration admission, complete saved migration intent and truthful fence recovery.
+- Current increment: long-history closeout reuse and TS-owned monitor evidence; the migration packages below remain open.
 - This checkpoint supersedes numerical remaining-PR estimates in earlier delivery entries.
 
 ## Correct the accounting
@@ -63,6 +64,39 @@ capacity qualification remain its separate medium-term path. Local default
 does not wait for PostgreSQL deployment; a passing conformance suite does not
 establish production service readiness.
 
+## Long-history closeout: this repair and its remaining boundary
+
+A live long-running lane lost the response to the five-second
+`quota.prior_host_turn_closeout.preflight` query; later read-only inspection and
+same-Turn retry recovered. Per-request indexing already exists. This repair
+removes repeated JSON decoding across reads: always read fresh bytes and hash
+the entire retained newline-terminated prefix before reuse; decode only appended
+lines when it matches. Rewrites, truncation, replacement, malformed rows and
+unfinished tails remain visible, as do conflicts in old Turns. Retain at most
+four logs and prefixes representing 128 MiB of source bytes; oversized histories
+use uncached parsing. This bounds retained source volume, not exact JS heap size.
+The cache is disposable and introduces no durable index, format or authority.
+
+Cold parsing yields between data batches to share the runtime event loop. Full
+byte reads remain necessary: this is not constant-time arbitrary-history support
+or D2 retention/capacity qualification. The five-second budget is unchanged. The
+incident's transient process/host scheduling cause was not reproduced reliably;
+validation establishes reduced duplicate work and concurrency headroom, not the
+absence of every possible environmental timeout.
+
+Monitor closeout now consumes the existing TS settlement rule for an exact
+committed poll. Python no longer scans the run log a second time and adapts only
+current Todo facts. Later uncommitted observations cannot hide earlier exact
+commit evidence; foreign identities and wrong effects cannot settle a Turn.
+A lost read-only preflight response reports `closeout_query_unavailable`, without
+asking for a nonexistent preflight write receipt. Unknown queries remain closed;
+there is no automatic mutation replay or shared-runtime restart.
+
+This is an evidenced R1/R5/S7 liveness repair and bounded Python retirement, not
+completion of implementation package 2. The three named boundaries above and
+separate #4931/D2 evidence gates remain. Existing quota CLI/heartbeat entrypoints
+adopt the change; no new setting or separate frontend/Lark policy is needed.
+
 ## Delivered #5013: complete-source transport and budget decision
 
 Before #5013, `test_canonical_snapshot_integration` failed before provider
diff --git a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-default-cutover-reconciliation.zh-CN.md b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-default-cutover-reconciliation.zh-CN.md
index 483db469ec..dd7bb82427 100644
--- a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-default-cutover-reconciliation.zh-CN.md
+++ b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-24-default-cutover-reconciliation.zh-CN.md
@@ -1,8 +1,9 @@
 # 默认切换：按实现证据重算交付边界
 
-- 核对基线：2026-09-25 `main` 的 `37bbaec79`；开放 PR 状态是快照，不是合入承诺。
+- 核对基线：2026-09-25 `main` 的 `41ba6f4d9`；开放 PR 状态是快照，不是合入承诺。
 - 归属：总目标 #4574 R5/G2；shared authority L2–L9/D1–D3；TS 迁移 T1–T4。
-- 本次交付：当前注册事实约束晋升，保存的模式意图完整执行，准确恢复 fence 状态。
+- 已交付 #5040：当前注册事实约束晋升，保存的模式意图完整执行，准确恢复 fence 状态。
+- 当前增量：长历史 closeout 读取复用与 TS monitor 回执归一；没有完成下列迁移工作包。
 - 本检查点取代此前交付记录中的剩余 PR 数量估算。
 
 ## 先纠正统计口径
@@ -53,6 +54,31 @@ PostgreSQL 复用 typed command 和 AuthorityStore，部署 transport、认证/t
 策略、restore identity、运维和 capacity 资格仍是独立中期路线。本地默认不等待
 PostgreSQL 部署，conformance 通过也不等于生产服务已合格。
 
+## 长历史收尾检查：本次修复与剩余边界
+
+真实长期运行暴露了 `quota.prior_host_turn_closeout.preflight` 的 5 秒超时；
+后续只读检查和同 Turn 重试恢复。历史已经在单次请求内建索引，不能把该优化当作
+未做。本次去除跨请求重复 JSON 解码：每次仍读取并 SHA-256 核验完整既有前缀，
+只复用字节相同且以换行结尾的解析结果，追加只解析新行。截断、同长度改写、替换、
+损坏、未完成尾行均重新验证；旧 Turn 冲突仍阻止推进。缓存最多保留四份日志、
+128 MiB 原始输入对应的解析前缀，超限回到普通读取。它不是持久索引或新 authority，
+不改变日志格式；原始字节预算不等于 JS heap 的硬上限。
+
+冷解析按数据批次让出事件循环，避免长历史独占共享 runtime。仍需读取全部字节，
+因此不宣称任意历史长度恒定耗时，也不替代 retention、D2 容量/恢复资格。5 秒预算
+保持不变，原事故的瞬时进程/机器调度原因未能稳定重现；回归证明的是重复工作降低
+以及长历史/并发下的可用余量，不声称消灭所有环境超时。
+
+Monitor 的精确已提交回执复用 TS settlement 的同一个查询规则；Python 删除第二遍
+run log 扫描，只适配当前 Todo 事实。后来的未提交观察不再遮住较早的精确提交证据；
+跨 Goal/Agent/Turn/Todo 或错误 effect 仍不能结算。只读 preflight 丢失响应报告
+`closeout_query_unavailable`，不会建议寻找不存在的 preflight 写回执；查询失败依旧
+阻止推断准入，不自动重试 mutation 或重启共享进程。
+
+这是 R1/R5/S7 的实测阻塞修复与有界 Python 退役，不是上表第 2 项整体完成。
+三个后续实现边界和 #4931/D2 的独立证据门不变。CLI/heartbeat 收益来自原有配额
+入口；没有新增设置，frontend/Lark 也无需各维护一套策略。
+
 ## 已交付 #5013：完整来源传输与预算决定
 
 在 #5013 之前，`test_canonical_snapshot_integration` 曾在 provider 准入之前失败：完整
diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
index 92150aaf13..3276631f87 100644
--- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
+++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
@@ -50,6 +50,11 @@ shadow lineage, still default-off and subject to explicit bootstrap.
 
 ## Current implementation checkpoint
 
+Long-history closeout reuse retains only re-verifiable parsed prefixes; it adds
+no durable authority and relaxes no writer fence or D2 gate. This runtime repair
+does not mechanically subtract one of the three remaining implementation packages.
+[Evidence and boundary](ledger/shared-goal-authority-state-provider-v0/2026-09-24-default-cutover-reconciliation.md#long-history-closeout-this-repair-and-its-remaining-boundary).
+
 Promotion admission now binds complete sources to a current registry witness
 and rechecks it inside the TS lock scope. Saved execution retains the reviewed
 handoff policy, and failures report durable fence presence. Recovery of a
diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
index a5fac9a549..a51de6138d 100644
--- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
+++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
@@ -43,6 +43,10 @@
 
 ## 当前实现检查点
 
+长历史 closeout 读取复用仅缓存可重新验证的解析前缀，不新增持久 authority 或放宽
+writer fence/D2。剩余三个实现边界不因该运行缺陷修复而机械减一。
+[证据与边界](ledger/shared-goal-authority-state-provider-v0/2026-09-24-default-cutover-reconciliation.zh-CN.md#长历史收尾检查本次修复与剩余边界)。
+
 晋升准入现将完整来源绑定到当前 registry witness，并在 TS 持锁范围内重新校验；
 保存计划执行保留已审核的 handoff 策略，失败结果如实报告持久 fence。
 已提交事务的恢复仍按原 fence/receipt，不要求失去权威的旧来源重新有效。
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
index d1ffe7790a..e0d1d9fecf 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
@@ -42,6 +42,12 @@ Retain T0 caller/parity inventory, T1/T2 transaction/effect convergence, T3 comp
 
 ## Current implementation checkpoint
 
+Long-history closeout now reuses byte-verified TS receipt prefixes and the
+single committed-monitor rule. Python retires its duplicate run-log scan and
+adapts Todo facts only; lost queries are distinguished from ambiguous writes.
+This is bounded retirement within closeout, not complete Python removal.
+[Delivery and limits](ledger/shared-goal-authority-state-provider-v0/2026-09-24-default-cutover-reconciliation.md#long-history-closeout-this-repair-and-its-remaining-boundary).
+
 Promotion admission now binds complete sources to a current registry witness
 and rechecks it inside the TS lock scope. Saved execution retains the reviewed
 handoff policy, and failures report durable fence presence. Recovery of a
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
index 623f91e129..f84b005a69 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
@@ -45,6 +45,10 @@ shared-authority 的当前核对表区分已合入实现、在途 PR、新代码
 
 ## 当前实现检查点
 
+长历史 closeout 现在复用经原始字节校验的 TS 日志前缀与统一 monitor 提交规则，
+Python 删除重复 run log 读取，只适配 Todo 事实。只读失败与提交不确定性分开报告。
+这是收尾边界内的有界退役，不是全量 Python 移除；[当前交付与限制](ledger/shared-goal-authority-state-provider-v0/2026-09-24-default-cutover-reconciliation.zh-CN.md#长历史收尾检查本次修复与剩余边界)。
+
 晋升准入现将完整来源绑定到当前 registry witness，并在 TS 持锁范围内重新校验；
 保存计划执行保留已审核的 handoff 策略，失败结果如实报告持久 fence。
 已提交事务的恢复仍按原 fence/receipt，不要求失去权威的旧来源重新有效。

From 26677918314d999bd81148ddb380a1eb31a590bc Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Fri, 25 Sep 2026 18:50:47 +0800
Subject: [PATCH 3/4] fix(quota): preserve read-only closeout failure through
 CLI projection

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 loopx/cli_commands/quota.py                   | 13 ++-
 loopx/cli_commands/quota_action_selection.py  | 19 ----
 loopx/cli_commands/quota_failure_report.py    | 15 ++-
 loopx/control_plane/quota/error_codes.py      |  9 ++
 .../control_plane/quota/heartbeat_receipt.py  | 18 ++++
 .../quota/unsettled_host_turn.py              |  9 +-
 .../test_quota_closeout_query_failure.py      | 92 +++++++++++++++++++
 .../test_prior_closeout_preflight_budget.py   |  3 +-
 8 files changed, 149 insertions(+), 29 deletions(-)
 create mode 100644 tests/cli_commands/test_quota_closeout_query_failure.py

diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py
index 8e4d8f1573..578d797c26 100644
--- a/loopx/cli_commands/quota.py
+++ b/loopx/cli_commands/quota.py
@@ -21,8 +21,12 @@
 )
 from ..control_plane.quota.effective_action import EffectiveAction
 from ..control_plane.quota.effect_program import SettlementIdentity
-from ..control_plane.quota.error_codes import QuotaCommandValidationError
+from ..control_plane.quota.error_codes import (
+    CloseoutQueryUnavailableError,
+    QuotaCommandValidationError,
+)
 from ..control_plane.quota.heartbeat_receipt import (
+    attach_uncommitted_heartbeat_receipt,
     fail_heartbeat_receipt,
     find_heartbeat_receipt,
     heartbeat_receipt_view,
@@ -65,7 +69,6 @@
 )
 from .quota_action_selection import (
     RequestedQuotaActionSelection,
-    attach_uncommitted_action_selection_receipt,
     commit_requested_action_selection,
     load_requested_quota_action_selection,
     reconcile_requested_quota_action_selection,
@@ -333,6 +336,7 @@ def handle_quota_command(
     heartbeat_receipt_existing_appended = False
     heartbeat_receipt_ready = False
     action_selection_preflight_failed = False
+    closeout_query_unavailable = False
     action_selection: RequestedQuotaActionSelection | None = None
     heartbeat_stall_observation = "not_evaluated"
     detail_sections: frozenset[str] = frozenset()
@@ -551,6 +555,7 @@ def handle_quota_command(
             runtime_root_arg=runtime_root_arg,
         )
     except Exception as exc:  # noqa: BLE001 - CLI fail-safe boundary; error_code is typed below.
+        closeout_query_unavailable = isinstance(exc, CloseoutQueryUnavailableError)
         payload = quota_failure_payload(
             args,
             registry_path=registry_path,
@@ -571,7 +576,7 @@ def handle_quota_command(
             replan_obligation_id=rollout_replan_obligation_id,
         )
         if heartbeat_turn_id and args.quota_command == "should-run":
-            if action_selection_preflight_failed:
+            if action_selection_preflight_failed or closeout_query_unavailable:
                 if heartbeat_receipt_existing:
                     render_existing_heartbeat_receipt_payload(
                         payload,
@@ -581,7 +586,7 @@ def handle_quota_command(
                         appended=heartbeat_receipt_existing_appended,
                     )
                 else:
-                    attach_uncommitted_action_selection_receipt(
+                    attach_uncommitted_heartbeat_receipt(
                         payload,
                         turn_instance_id=heartbeat_turn_id,
                     )
diff --git a/loopx/cli_commands/quota_action_selection.py b/loopx/cli_commands/quota_action_selection.py
index be785309e9..f0e5690ea1 100644
--- a/loopx/cli_commands/quota_action_selection.py
+++ b/loopx/cli_commands/quota_action_selection.py
@@ -11,7 +11,6 @@
     QuotaActionSelectionConflictKind,
 )
 from ..control_plane.quota.heartbeat_receipt import (
-    HEARTBEAT_RECEIPT_SCHEMA_VERSION,
     find_heartbeat_receipt,
     heartbeat_receipt_pending_action_todo_id,
     heartbeat_receipt_settlement_replan_obligation_id,
@@ -282,24 +281,6 @@ def reconcile_requested_quota_action_selection(
     )
 
 
-def attach_uncommitted_action_selection_receipt(
-    payload: dict[str, object],
-    *,
-    turn_instance_id: str,
-) -> None:
-    """Expose an accurate non-durable receipt for a rejected preflight."""
-
-    payload["heartbeat_receipt"] = {
-        "schema_version": HEARTBEAT_RECEIPT_SCHEMA_VERSION,
-        "turn_instance_id": turn_instance_id,
-        "status": "not_committed",
-        "stall_observation": "not_evaluated",
-        "reason_code": str(
-            payload.get("error_code") or "quota_action_selection_rejected"
-        ),
-    }
-
-
 def commit_requested_action_selection(
     payload: Mapping[str, object],
     *,
diff --git a/loopx/cli_commands/quota_failure_report.py b/loopx/cli_commands/quota_failure_report.py
index 8c7936441d..cc3b156150 100644
--- a/loopx/cli_commands/quota_failure_report.py
+++ b/loopx/cli_commands/quota_failure_report.py
@@ -19,7 +19,9 @@
     LocalCoordinationAuthorityUnavailable,
 )
 from ..control_plane.effect_runtime import EffectRuntimeStartupError
+from ..control_plane.quota.effective_action import EffectiveAction
 from ..control_plane.quota.error_codes import (
+    CloseoutQueryUnavailableError,
     HeartbeatReceiptIdentityConflictError,
     QuotaActionSelectionConflictError,
     QuotaCommandValidationError,
@@ -112,7 +114,8 @@ def quota_failure_payload(
     public_reason = (
         str(error)
         if isinstance(
-            error, (HeartbeatReceiptIdentityConflictError, EffectRuntimeStartupError)
+            error,
+            (CloseoutQueryUnavailableError, HeartbeatReceiptIdentityConflictError, EffectRuntimeStartupError),
         )
         else "quota collection failed"
     )
@@ -134,6 +137,16 @@ def quota_failure_payload(
         **verbose_debug,
         **lock_timeout_fields,
     }
+    if isinstance(error, CloseoutQueryUnavailableError):
+        payload.update({
+            "status": error.diagnostic_code,
+            "effective_action": EffectiveAction.CONTROL_PLANE_HEALTH_REPAIR.value,
+            "recommended_action": (
+                "check runtime health, then retry quota should-run with the same "
+                "Turn identity to read the closeout state; do not infer settlement "
+                "or replay work from a missing query response"
+            ),
+        })
     if isinstance(error, QuotaActionSelectionConflictError):
         # The requested Todo could not be reconciled with the projection. Report
         # the real conflict and the next read to make, rather than the generic
diff --git a/loopx/control_plane/quota/error_codes.py b/loopx/control_plane/quota/error_codes.py
index e8296e7bb0..bc9c71273e 100644
--- a/loopx/control_plane/quota/error_codes.py
+++ b/loopx/control_plane/quota/error_codes.py
@@ -4,6 +4,13 @@
 from enum import StrEnum
 
 
+class CloseoutQueryUnavailableError(RuntimeError):
+    """A read-only closeout query returned no verified result; no verdict exists."""
+
+    error_code = "quota_closeout_query_unavailable"
+    diagnostic_code = "closeout_query_unavailable"
+
+
 class QuotaCommandValidationError(ValueError):
     """Public-safe diagnostic for an invalid ``loopx quota`` invocation."""
 
@@ -199,6 +206,8 @@ def __init__(
 
 
 def quota_error_code(exc: BaseException) -> str:
+    if isinstance(exc, CloseoutQueryUnavailableError):
+        return exc.error_code
     if isinstance(exc, json.JSONDecodeError):
         return "quota_state_invalid_json"
     if isinstance(exc, QuotaCommandValidationError):
diff --git a/loopx/control_plane/quota/heartbeat_receipt.py b/loopx/control_plane/quota/heartbeat_receipt.py
index 4680c50bac..f02fec2e14 100644
--- a/loopx/control_plane/quota/heartbeat_receipt.py
+++ b/loopx/control_plane/quota/heartbeat_receipt.py
@@ -552,6 +552,24 @@ def heartbeat_receipt_view(
     return receipt
 
 
+def attach_uncommitted_heartbeat_receipt(
+    payload: dict[str, object],
+    *,
+    turn_instance_id: str,
+) -> None:
+    """Expose an accurate non-durable receipt for an incomplete preflight."""
+
+    payload["heartbeat_receipt"] = {
+        "schema_version": HEARTBEAT_RECEIPT_SCHEMA_VERSION,
+        "turn_instance_id": turn_instance_id,
+        "status": "not_committed",
+        "stall_observation": "not_evaluated",
+        "reason_code": str(
+            payload.get("error_code") or "quota_preflight_incomplete"
+        ),
+    }
+
+
 def fail_heartbeat_receipt(
     payload: dict[str, object],
     *,
diff --git a/loopx/control_plane/quota/unsettled_host_turn.py b/loopx/control_plane/quota/unsettled_host_turn.py
index 9a5d706948..62f95266d6 100644
--- a/loopx/control_plane/quota/unsettled_host_turn.py
+++ b/loopx/control_plane/quota/unsettled_host_turn.py
@@ -17,7 +17,6 @@
 from ..effect_runtime import (
     EffectRuntimeRejected,
     EffectRuntimeResponseAmbiguous,
-    EffectRuntimeStartupError,
     effect_runtime_result,
 )
 from ..scheduler.execution_context import SchedulerExecutionContextResolution
@@ -25,7 +24,10 @@
 from ..work_items.interaction_contract import (
     build_interaction_contract,
 )
-from .error_codes import HeartbeatReceiptIdentityConflictError
+from .error_codes import (
+    CloseoutQueryUnavailableError,
+    HeartbeatReceiptIdentityConflictError,
+)
 
 UNSETTLED_HOST_TURN_RECOVERY_SCHEMA_VERSION = "unsettled_host_turn_recovery_v0"
 
@@ -127,12 +129,11 @@ def _prior_closeout_preflight(
         # possibly committed mutation, and must not send the operator hunting
         # for a nonexistent preflight write receipt. Do not infer a verdict or
         # automatically restart/retry the shared runtime.
-        raise EffectRuntimeStartupError(
+        raise CloseoutQueryUnavailableError(
             f"Read-only {PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_METHOD} returned no "
             f"verifiable response within {PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_TIMEOUT_SECONDS:g}s; "
             "closeout state is unknown. Retry the query after checking runtime health; "
             "the preflight itself performs no durable writes",
-            diagnostic_code="closeout_query_unavailable",
         ) from exc
     except EffectRuntimeRejected as exc:
         # Keep the public diagnostic the identity rule has always published,
diff --git a/tests/cli_commands/test_quota_closeout_query_failure.py b/tests/cli_commands/test_quota_closeout_query_failure.py
new file mode 100644
index 0000000000..94cce70b9a
--- /dev/null
+++ b/tests/cli_commands/test_quota_closeout_query_failure.py
@@ -0,0 +1,92 @@
+"""The public quota response distinguishes a lost read from a lost write."""
+from __future__ import annotations
+
+import json
+from unittest.mock import Mock
+
+import pytest
+
+from loopx.cli import build_parser
+import loopx.cli_commands.quota as command
+from loopx.cli_commands.quota_context import QuotaCommandContext
+from loopx.control_plane.effect_runtime import EffectRuntimeResponseAmbiguous
+import loopx.control_plane.quota.unsettled_host_turn as closeout
+from loopx.rollout_event_log import append_rollout_event, build_rollout_event, rollout_event_log_path
+
+
+@pytest.mark.parametrize("existing", [False, True])
+@pytest.mark.parametrize("turn_envelope", [False, True])
+@pytest.mark.parametrize("failure", ["query", "write"])
+def test_cli_preserves_query_failure_and_existing_receipt(
+    tmp_path, monkeypatch, existing, turn_envelope, failure,
+):
+    goal, agent, turn = "fixture-goal", "fixture-agent", "fixture-turn"
+    registry = tmp_path / "registry.json"
+    registry.write_text('{}')
+    if existing:
+        event = build_rollout_event(
+            event_kind="quota_should_run", goal_id=goal, agent_id=agent,
+            run_id=turn, status="run", summary="synthetic prior admission",
+            details={"turn_instance_id": turn, "stall_observation": "not_applicable"},
+        )
+        append_rollout_event(rollout_event_log_path(tmp_path, goal), event)
+    args = build_parser().parse_args([
+        "--format", "json", "quota", "should-run", "--goal-id", goal,
+        "--agent-id", agent, "--turn-instance-id", turn,
+    ] + (["--turn-envelope"] if turn_envelope else []))
+    context = QuotaCommandContext(
+        runtime_root=tmp_path, scan_roots=[], status_limit=1, status_goal_id=goal,
+        status_payload={}, cache_metadata=None, scheduler_context=None,
+        operator_inbox_urgency_projector=lambda **kwargs: {},
+        detail_sections=frozenset(), heartbeat_turn_id=turn,
+    )
+    # Only isolate unrelated collection/hooks and inject the transport loss.
+    # Receipt lookup, exception translation, CLI failure projection, and final
+    # compact/envelope rendering run through their production implementations.
+    monkeypatch.setattr(command, "_dispatch_quota_turn_start_hooks", lambda *a, **k: (None, False))
+    monkeypatch.setattr(command, "prepare_quota_command_context", lambda *a, **k: context)
+    transport = Mock(side_effect=EffectRuntimeResponseAmbiguous(
+        closeout.PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_METHOD if failure == "query"
+        else "quota.heartbeat.commit", timeout=5,
+    ))
+    monkeypatch.setattr(closeout, "effect_runtime_result", transport)
+
+    def decision(*a, **k):
+        if failure == "write":
+            return transport()
+        return closeout._prior_closeout_preflight(
+            runtime_root=tmp_path, goal_id=goal, agent_id=agent,
+            current_turn_instance_id=turn,
+        )
+
+    monkeypatch.setattr(command, "build_live_quota_should_run_decision", decision)
+    output = []
+    append = Mock(side_effect=AssertionError("failed preflight cannot admit a Turn"))
+    before = {p: p.read_bytes() for p in tmp_path.rglob("*") if p.is_file()}
+    result = command.handle_quota_command(
+        args, registry_path=registry, runtime_root_arg=str(tmp_path),
+        print_payload=lambda payload, *_: output.append(json.loads(json.dumps(payload))),
+        append_cli_rollout_event=append,
+    )
+    assert result == 1
+    payload = output[-1]
+    assert payload["ok"] is False
+    assert payload["should_run"] is False
+    receipt = payload["heartbeat_receipt"]
+    assert transport.call_count == 1
+    append.assert_not_called()
+    assert {p: p.read_bytes() for p in tmp_path.rglob("*") if p.is_file()} == before
+    if failure == "write":
+        assert payload["effective_action"] == "heartbeat_receipt_write_failed"
+        assert receipt["status"] == "write_failed"
+        return
+    assert payload["error_code"] == "quota_closeout_query_unavailable"
+    assert payload["effective_action"] == "control_plane_health_repair"
+    assert "closeout state is unknown" in payload["reason"]
+    assert "same Turn identity" in payload["recommended_action"]
+    assert "repairing heartbeat receipt" not in payload["recommended_action"]
+    assert receipt["status"] == ("replayed" if existing else "not_committed")
+    if existing:
+        assert receipt["event_id"] == event["event_id"]
+    else:
+        assert receipt["reason_code"] == payload["error_code"]
diff --git a/tests/control_plane/test_prior_closeout_preflight_budget.py b/tests/control_plane/test_prior_closeout_preflight_budget.py
index 07f38f3a8a..2fccb47b1c 100644
--- a/tests/control_plane/test_prior_closeout_preflight_budget.py
+++ b/tests/control_plane/test_prior_closeout_preflight_budget.py
@@ -17,6 +17,7 @@
     EffectRuntimeResponseAmbiguous,
     EffectRuntimeStartupError,
 )
+from loopx.control_plane.quota.error_codes import CloseoutQueryUnavailableError
 from loopx.control_plane.quota.unsettled_host_turn import (
     PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_METHOD,
     PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_REQUEST_SCHEMA,
@@ -87,7 +88,7 @@ def test_a_lost_preflight_response_is_an_unknown_query_not_an_ambiguous_write():
             PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_METHOD, timeout=5,
         ),
     ) as request:
-        with pytest.raises(EffectRuntimeStartupError) as raised:
+        with pytest.raises(CloseoutQueryUnavailableError) as raised:
             _preflight()
     assert request.call_count == 1
     assert raised.value.diagnostic_code == "closeout_query_unavailable"

From 723fc33cadd977b7bfbd7c8f9ec9a4dec6cc2fe3 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Fri, 25 Sep 2026 18:53:31 +0800
Subject: [PATCH 4/4] fix(quota): declare closeout diagnostic action producer

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 loopx/semantics/vocabulary_v0.json | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/loopx/semantics/vocabulary_v0.json b/loopx/semantics/vocabulary_v0.json
index df27a6d433..422be32d17 100644
--- a/loopx/semantics/vocabulary_v0.json
+++ b/loopx/semantics/vocabulary_v0.json
@@ -470,7 +470,7 @@
         "blocked_wait": "Existing result of quota_effective_action; previously missed because the literal scan did not inspect declared return functions.",
         "boundary_projection_repair": "A required write scope is missing from the projected goal boundary; trigger required_write_scope_missing_from_goal_boundary.",
         "capability_bridge_repair": "Capability repair is allowed once workspace repair and self-repair are not; the capability bridge is repaired before delivery.",
-        "control_plane_health_repair": "Stall repair raised by a health blocker; recommended_mode repair_control_plane_health. One of the five self-repair spend actions that stand in for the generic control_plane_repair.",
+        "control_plane_health_repair": "Control-plane health repair for an unavailable closeout query or a stall health blocker; recommended_mode repair_control_plane_health. One of the five self-repair spend actions that stand in for the generic control_plane_repair.",
         "control_plane_projection_repair": "Stall repair for a lane left waiting with no owner projection; trigger waiting_without_owner_projection.",
         "control_plane_repair": "Existing result of quota_effective_action; previously missed because the literal scan did not inspect declared return functions.",
         "coordinate_task_bundle": "Replaces normal_run when a ready task-orchestration contract makes this lane the coordinator: admitted or explicitly selected peer lanes are activated or resumed before its own worker-lane delivery.",
@@ -500,6 +500,7 @@
         "loopx/control_plane/quota/decision_summary.py::_task_orchestration_effective_action",
         "loopx/control_plane/quota/decision_summary.py::quota_effective_action",
         "loopx/control_plane/quota/decision_summary.py::resolve_quota_run_decision",
+        "loopx/cli_commands/quota_failure_report.py::quota_failure_payload",
         "loopx/control_plane/quota/heartbeat_receipt.py::fail_heartbeat_receipt",
         "loopx/control_plane/quota/live_decision.py::_apply_pending_capability_intent_precedence",
         "loopx/control_plane/work_items/action_portfolio.ts::reconcileRetainedActionSelection",