refactor(quota): make TypeScript own prior-Turn closeout recovery

Move the prior-host-Turn closeout transaction behind one TypeScript
request/response boundary so the Python coordinator keeps only transport
and public-payload projection.

The typed owner now decides which prior must-attempt Turn still needs a
closeout, whether its settlement already validates, which closeout is
accepted, and what the recovery obligation is.  Python reads exactly the
two provider facts the preflight names - the bound Todo and the committed
monitor-poll receipt - and projects the typed verdict.

Two requests, not one per receipt: a fail-closed preflight that reads the
persisted guards and the settlement readback itself, then one final
reduction over the checkpointed bound facts.  The preflight reads the
rollout log from runtime_root because a real goal persists 3.25 MiB of
receipt facts for its busiest Agent, and the runtime bridge rejects any
request over its 2 MiB bound.

Semantics tightened while sharing the rule with the settlement readback:

- the heartbeat receipt identity rule has one owner instead of one
  TypeScript and one Python copy;
- a receipt that declares a Todo binding the settlement authority cannot
  address now fails the read instead of minting a recovery obligation for
  an identity no other reader can reproduce;
- a receipt that names a settlement effect without a Todo or autonomous
  replan binding keeps failing closed, and the declared effect id is no
  longer silently replaced by the derived one.

Python deletion: the closeout policy, the receipt-selection rule, the
identity rule, and the settlement-readback transport for this path.

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>

Author: huangruiteng

loopx/control_plane/effect_runtime_handlers.ts

@@ -52,6 +52,10 @@ import { evaluateDeliveryWorkspaceCausality } from "./quota/settlement_workspace
 import { evaluateQuotaSpendCommit } from "./quota/spend_commit.ts";
 import { evaluateQuotaVoidCommit } from "./quota/void_commit.ts";
 import { readQuotaSettlement } from "./quota/settlement_readback.ts";
+import {
+  preflightPriorHostTurnCloseout,
+  reduceUnsettledHostTurnRecovery,
+} from "./quota/unsettled_host_turn_recovery.ts";
 import { evaluateTurnEnvelope } from "./quota/turn_envelope.ts";
 import { evaluateQuotaMonitorPollCommit } from "./quota/monitor_poll_commit.ts";
 import { planMonitorSuccessor, selectMonitorTodoRequest } from "./scheduler/monitor_successor.ts";
@@ -463,6 +467,14 @@ export function createEffectRuntimeHandlers(
     ["quota.spend.commit", evaluateQuotaSpendCommit],
     ["quota.void.commit", evaluateQuotaVoidCommit],
     ["quota.settlement.read", readQuotaSettlement],
+    [
+      "quota.prior_host_turn_closeout.preflight",
+      preflightPriorHostTurnCloseout,
+    ],
+    [
+      "quota.unsettled_host_turn_recovery.reduce",
+      reduceUnsettledHostTurnRecovery,
+    ],
     ["quota.turn_envelope.evaluate", evaluateTurnEnvelope],
     ["task_lease.owner_eligibility", evaluateTaskLeaseOwnerEligibility],
     ["task_lease.acquire.decide", evaluateTaskLeaseAcquireDecision],

loopx/control_plane/quota/heartbeat_receipt.py

@@ -153,51 +153,6 @@ def find_heartbeat_receipt(
     )
 
 
-def prior_closeout_required_heartbeat_receipts(
-    runtime_root: Path,
-    *,
-    goal_id: str,
-    agent_id: str,
-    exclude_turn_instance_id: str | None = None,
-) -> list[dict[str, object]]:
-    """Return prior heartbeat guards that explicitly require host closeout.
-
-    The expectation is opt-in on the persisted guard so older receipts cannot
-    become false-positive recovery obligations after an upgrade. Results are
-    newest first and contain at most one effective receipt per Turn.
-    """
-
-    events = load_rollout_events(rollout_event_log_path(runtime_root, goal_id))
-    matching_by_turn: dict[str, list[dict[str, object]]] = {}
-    newest_turns: list[str] = []
-    excluded = str(exclude_turn_instance_id or "").strip()
-    for event in events:
-        if (
-            event.get("event_kind") != "quota_should_run"
-            or str(event.get("goal_id") or "") != goal_id
-            or str(event.get("agent_id") or "") != agent_id
-        ):
-            continue
-        turn_id = str(event.get("run_id") or "").strip()
-        if not turn_id or turn_id == excluded:
-            continue
-        matching_by_turn.setdefault(turn_id, []).append(event)
-        if turn_id in newest_turns:
-            newest_turns.remove(turn_id)
-        newest_turns.append(turn_id)
-
-    required: list[dict[str, object]] = []
-    for turn_id in reversed(newest_turns):
-        effective = _effective_heartbeat_receipt(matching_by_turn[turn_id])
-        if effective is None:
-            continue
-        details_value = effective.get("details")
-        details = details_value if isinstance(details_value, Mapping) else {}
-        if details.get("closeout_required") is True:
-            required.append(effective)
-    return required
-
-
 def ensure_turn_heartbeat_settlement_receipt(
     runtime_root: Path,
     identity: SettlementIdentity,

loopx/control_plane/quota/heartbeat_receipt_identity.ts

@@ -0,0 +1,182 @@
+/**
+ * One owner for the persisted heartbeat-receipt settlement identity rule.
+ *
+ * A Turn can persist several `quota_should_run` events. The effective receipt
+ * is the one that binds a settlement identity; a Turn whose events disagree on
+ * that binding is an identity conflict and must fail closed instead of letting
+ * a caller infer, upgrade, or silently prefer one binding.
+ *
+ * Both the settlement readback and the prior-host-Turn closeout selection read
+ * this rule, so it lives here rather than in either caller.
+ */
+import type { JsonObject } from "../effect_program.ts";
+import {settlementIdentity, type SettlementIdentity} from "../effect_program.ts";
+import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
+import { jsonObject } from "../runtime_decode.ts";
+
+const TODO_ID_PATTERN = /^todo_[a-z0-9_-]{3,64}$/;
+const REPLAN_OBLIGATION_ID_PATTERN = /^replan-[a-f0-9]{16}$/;
+/** Public diagnostic preserved from the retired Python identity rule. */
+const IDENTITY_CONFLICT_CODE = "heartbeat_receipt_identity_conflict";
+
+function receiptIdentityConflict(message: string): EffectRuntimeRequestError {
+  return new EffectRuntimeRequestError(message, IDENTITY_CONFLICT_CODE);
+}
+
+export function heartbeatReceiptDetails(event: JsonObject | null): JsonObject {
+  return jsonObject(event?.details) ?? {};
+}
+
+export function optionalHeartbeatString(value: unknown): string | null {
+  if (value === null || value === undefined) return null;
+  return String(value).trim() || null;
+}
+
+export function normalizeHeartbeatTodoId(value: unknown): string | null {
+  const candidate = String(value ?? "").trim().toLowerCase();
+  return candidate && TODO_ID_PATTERN.test(candidate) ? candidate : null;
+}
+
+export function normalizeHeartbeatReplanObligationId(
+  value: unknown,
+): string | null {
+  const candidate = String(value ?? "").trim();
+  return candidate && REPLAN_OBLIGATION_ID_PATTERN.test(candidate)
+    ? candidate
+    : null;
+}
+
+export interface HeartbeatReceiptBinding extends JsonObject {
+  binding_kind: "todo" | "autonomous_replan";
+  /** The Todo id or autonomous replan obligation id, per `binding_kind`. */
+  binding_id: string;
+  /**
+   * The effect id the receipt declares, kept verbatim for projection.
+   *
+   * It is intentionally not repaired from the identity's derived effect id: a
+   * receipt that names an effect the settlement authority could not derive
+   * stays visible in the projected payload instead of being silently rewritten.
+   */
+  settlement_effect_id: string | null;
+  /** The conflict-detection key: one Turn may declare only one of these. */
+  identity_key: string;
+}
+
+/**
+ * The only receipt fields the settlement binding rule can read.
+ *
+ * The persisted rollout event and the settlement readback's own projection both
+ * decode into this shape, so one rule serves both readers.  The goal and Agent
+ * a receipt belongs to are not receipt facts: they are the scope the reader
+ * already selected.
+ */
+export interface HeartbeatReceiptFact extends JsonObject {
+  event_id: string | null;
+  run_id: string | null;
+  todo_id: string | null;
+  replan_obligation_id: string | null;
+  settlement_effect_id: string | null;
+  closeout_required: boolean;
+}
+
+export function heartbeatReceiptFactFromEvent(
+  event: JsonObject,
+): HeartbeatReceiptFact {
+  const eventDetails = heartbeatReceiptDetails(event);
+  return {
+    event_id: optionalHeartbeatString(event.event_id),
+    run_id: optionalHeartbeatString(event.run_id),
+    todo_id: optionalHeartbeatString(eventDetails.todo_id),
+    replan_obligation_id: optionalHeartbeatString(
+      eventDetails.replan_obligation_id,
+    ),
+    settlement_effect_id: optionalHeartbeatString(
+      eventDetails.settlement_effect_id,
+    ),
+    closeout_required: eventDetails.closeout_required === true,
+  };
+}
+
+/**
+ * Resolve the settlement binding a persisted receipt declares.
+ *
+ * A receipt without a binding is not a settlement identity; it is reported as
+ * `null` so the caller can decide whether an unbound receipt is admissible.
+ */
+export function heartbeatReceiptBinding(
+  goalId: string,
+  agentId: string,
+  fact: HeartbeatReceiptFact,
+): HeartbeatReceiptBinding | null {
+  const declaredTodoId = optionalHeartbeatString(fact.todo_id);
+  const replanObligationId = normalizeHeartbeatReplanObligationId(
+    fact.replan_obligation_id,
+  );
+  const declaredEffectId = optionalHeartbeatString(fact.settlement_effect_id);
+  if (declaredTodoId && replanObligationId) {
+    throw receiptIdentityConflict(
+      "heartbeat receipt has conflicting Todo and autonomous replan bindings",
+    );
+  }
+  if (declaredEffectId && !declaredTodoId && !replanObligationId) {
+    throw receiptIdentityConflict(
+      "heartbeat receipt has an effect identity without a Todo or autonomous replan binding; refuse to infer or upgrade it",
+    );
+  }
+  if (!declaredTodoId && !replanObligationId) return null;
+  // A Todo-binding the settlement authority cannot address is not a binding we
+  // may act on: acting on the raw string would create a recovery obligation
+  // whose identity no other reader can reproduce, and dropping it would let a
+  // required closeout disappear. Refuse the read instead.
+  const todoId = declaredTodoId;
+  if (todoId !== null && normalizeHeartbeatTodoId(todoId) !== todoId) {
+    throw receiptIdentityConflict(
+      "heartbeat receipt declares a Todo binding that is not a legal Todo id",
+    );
+  }
+  const identity = settlementIdentity({
+    goal_id: goalId,
+    agent_id: agentId,
+    todo_id: todoId,
+    turn_instance_id: fact.run_id ?? "",
+    replan_obligation_id: replanObligationId,
+  });
+  return {
+    binding_kind: identity.binding_kind === "todo"
+      ? "todo"
+      : "autonomous_replan",
+    binding_id: identity.binding_id,
+    settlement_effect_id: declaredEffectId,
+    identity_key: `${identity.binding_kind}\u0000${identity.binding_id}\u0000${
+      declaredEffectId ?? identity.effect_id
+    }`,
+  };
+}
+
+/**
+ * Reduce one Turn's receipts to its effective receipt.
+ *
+ * Receipts without a settlement binding are admissible only while the Turn
+ * declares no binding at all; they cannot outrank a bound receipt, and they
+ * cannot silently turn a conflicting Turn into a valid one.
+ */
+export function selectEffectiveHeartbeatReceipt<Value>(
+  goalId: string,
+  agentId: string,
+  entries: readonly { fact: HeartbeatReceiptFact; value: Value }[],
+): Value | null {
+  if (entries.length === 0) return null;
+  const identities = new Map<string, Value>();
+  for (const entry of entries) {
+    const binding = heartbeatReceiptBinding(goalId, agentId, entry.fact);
+    if (binding) identities.set(binding.identity_key, entry.value);
+  }
+  if (identities.size > 1) {
+    throw receiptIdentityConflict(
+      "heartbeat receipt has conflicting settlement identities for the same goal, agent, and turn",
+    );
+  }
+  return identities.size === 1
+    ? [...identities.values()][0]!
+    : entries.at(-1)!.value;
+}