fix(scheduler): fence Python follow-ups on superseded heartbeat receipts

loopx/cli_commands/quota_context.py

@@ -13,6 +13,7 @@
     cached_goal_run_index_is_current,
     load_status_projection_cache,
     resolve_status_projection_cache_runtime_root,
+    status_projection_cache_metadata,
     write_status_projection_cache,
 )
 from ..control_plane.scheduler.execution_context import (
@@ -362,20 +363,48 @@ def prepare_quota_command_context(
             available_capabilities=args.available_capabilities,
             agent_lane_id=args.agent_id,
             include_presentation_projections=False,
+            # A scheduler ACK/failure return writes cadence, not settlement
+            # facts. One damaged audit row must not fail the whole status read
+            # before the tolerant native fence can answer it.
+            tolerate_malformed_settlement_history=(
+                command in QUOTA_SCHEDULER_FOLLOWUP_COMMANDS
+            ),
         )
         if bool(getattr(args, "write_projection_cache", False)):
-            cache_metadata = write_status_projection_cache(
-                registry_path=registry_path,
-                runtime_root=runtime_root,
-                limit=status_limit,
-                include_task_graph=False,
-                goal_id=status_goal_id,
-                payload=status_payload,
-                max_age_seconds=projection_cache_ttl_seconds,
-                available_capabilities=args.available_capabilities,
-                agent_lane_id=args.agent_id,
-                include_presentation_projections=False,
-            )
+            if command in QUOTA_SCHEDULER_FOLLOWUP_COMMANDS:
+                # The follow-up collector tolerates malformed history so a
+                # valid scheduler return can still be recorded. That partial
+                # view must never become the cache consumed by strict status
+                # readers under the same key.
+                cache_metadata = status_projection_cache_metadata(
+                    registry_path=registry_path,
+                    runtime_root=runtime_root,
+                    limit=status_limit,
+                    include_task_graph=False,
+                    goal_id=status_goal_id,
+                    max_age_seconds=projection_cache_ttl_seconds,
+                    available_capabilities=args.available_capabilities,
+                    agent_lane_id=args.agent_id,
+                    include_presentation_projections=False,
+                )
+                cache_metadata.update({
+                    "hit": False,
+                    "written": False,
+                    "skipped_reason": "tolerant_scheduler_followup_view",
+                })
+            else:
+                cache_metadata = write_status_projection_cache(
+                    registry_path=registry_path,
+                    runtime_root=runtime_root,
+                    limit=status_limit,
+                    include_task_graph=False,
+                    goal_id=status_goal_id,
+                    payload=status_payload,
+                    max_age_seconds=projection_cache_ttl_seconds,
+                    available_capabilities=args.available_capabilities,
+                    agent_lane_id=args.agent_id,
+                    include_presentation_projections=False,
+                )
     elif isinstance(status_payload.get("projection_cache"), dict):
         cache_metadata = dict(status_payload["projection_cache"])
 

loopx/cli_commands/quota_scheduler_followup.py

@@ -76,6 +76,11 @@ def _build_scheduler_followup_decision(
             repository_delivery_interaction_hook(repo_path=Path.cwd()),
         ),
         goal_ref=goal_ref,
+        # The compatibility ACK/failure route writes scheduler cadence, not
+        # settlement facts: one damaged audit row must not turn a valid return
+        # into a failure before the tolerant native fence runs. Exact
+        # settlement branches keep the strict readback.
+        tolerate_malformed_settlement_history=True,
     )
 
 
@@ -105,35 +110,6 @@ def build_scheduler_followup_payload(
         if turn_instance_id
         else None
     )
-    if turn_instance_id and heartbeat_receipt is None:
-        return {
-            "ok": False,
-            "schema_version": "quota_scheduler_followup_receipt_failure_v0",
-            "mode": args.quota_command,
-            "goal_id": args.goal_id,
-            "agent_id": args.agent_id,
-            "turn_instance_id": turn_instance_id,
-            "decision": "skip",
-            "should_run": False,
-            "status": "heartbeat_receipt_missing",
-            "state": "blocked_receipt",
-            "error_code": "SCHEDULER_FOLLOWUP_HEARTBEAT_RECEIPT_MISSING",
-            "reason": (
-                "scheduler follow-up requires the originating heartbeat receipt; "
-                "refusing to rebuild authority from the current live frontier"
-            ),
-            "retry_guidance": (
-                "retry with the exact Turn id from a committed quota should-run "
-                "heartbeat receipt"
-            ),
-            "write_performed": False,
-            "appended": False,
-            "registry_mutated": False,
-            "scheduler_state_mutated": False,
-            "quota_spend_performed": False,
-            "delivery_outcome": "surface_only",
-        }
-
     observed_rrule = str(
         getattr(args, "app_automation_current_rrule", None)
         or getattr(args, "codex_app_current_rrule", None)
@@ -228,6 +204,8 @@ def build_scheduler_followup_payload(
             failed_rrule=args.failed_rrule,
             observed_host_rrule=observed_rrule,
             failure_kind=args.failure_kind,
+            turn_instance_id=turn_instance_id,
+            goal_ref=goal_ref,
         )
     return record_quota_scheduler_ack(
         status_payload,
@@ -248,4 +226,6 @@ def build_scheduler_followup_payload(
         scheduler_execution_context=scheduler_context,
         operator_inbox_urgency_projector=operator_inbox_urgency_projector,
         before_decision=before_decision,
+        turn_instance_id=turn_instance_id,
+        goal_ref=goal_ref,
     )

loopx/control_plane/quota/live_decision.py

@@ -685,6 +685,7 @@ def build_live_quota_should_run_decision(
     workspace_path: Path | None = None,
     goal_ref: Mapping[str, object] | None = None,
     inline_goal_source: bool = False,
+    tolerate_malformed_settlement_history: bool = False,
 ) -> dict[str, Any]:
     """Build one live CLI decision while keeping host observation injectable."""
     resolved_context = resolve_scheduler_execution_context(scheduler_execution_context)
@@ -742,6 +743,7 @@ def build_live_quota_should_run_decision(
         ),
         registry_path=registry_path,
         goal_ref=goal_ref,
+        tolerate_malformed_history=tolerate_malformed_settlement_history,
     )
     receipt_bound_monitor_phase = (
         settlement_readback.monitor_phase if settlement_readback else None

loopx/control_plane/quota/scheduler_ack.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Mapping
 from typing import Any, cast
 
 import json
@@ -184,14 +185,17 @@ def _run_native_followup(
     host_facts: dict[str, Any],
     use_current_hint: bool,
     reason_summary: str | None,
+    turn_instance_id: str | None = None,
+    goal_ref: Mapping[str, object] | None = None,
 ) -> dict[str, Any]:
     """Decision-free compatibility transport for in-process Python callers."""
 
     request = {
         "schema_version": _FOLLOWUP_REQUEST_SCHEMA,
         "runtime_root": str(runtime_root),
-        "turn_instance_id": None,
-        "require_heartbeat_receipt": False,
+        "goal_ref": dict(goal_ref) if goal_ref is not None else None,
+        "turn_instance_id": turn_instance_id,
+        "require_heartbeat_receipt": bool(turn_instance_id),
         "before": before,
         "host_facts": host_facts,
         "use_current_hint": use_current_hint,
@@ -298,6 +302,8 @@ def record_quota_scheduler_ack_for_decision(
     generated_at: str | None = None,
     use_current_hint: bool = False,
     host_match_observed: bool = False,
+    turn_instance_id: str | None = None,
+    goal_ref: Mapping[str, object] | None = None,
 ) -> dict[str, Any]:
     state_key = _followup_state_key(before, surface=surface, state_key=state_key)
     safe_agent_id = normalize_todo_claimed_by(agent_id)
@@ -367,6 +373,8 @@ def record_quota_scheduler_ack_for_decision(
             host_facts=facts,
             use_current_hint=use_current_hint,
             reason_summary=reason_summary,
+            turn_instance_id=turn_instance_id,
+            goal_ref=goal_ref,
         )
     except ValueError as exc:
         return _failure_payload(
@@ -396,6 +404,8 @@ def record_quota_scheduler_failure_for_decision(
     observed_host_rrule: str | None = None,
     failure_kind: str = "host_tool_failure",
     generated_at: str | None = None,
+    turn_instance_id: str | None = None,
+    goal_ref: Mapping[str, object] | None = None,
 ) -> dict[str, Any]:
     state_key = _followup_state_key(before, surface=surface, state_key=state_key)
     safe_agent_id = normalize_todo_claimed_by(agent_id)
@@ -443,6 +453,8 @@ def record_quota_scheduler_failure_for_decision(
             host_facts=facts,
             use_current_hint=False,
             reason_summary=None,
+            turn_instance_id=turn_instance_id,
+            goal_ref=goal_ref,
         )
     except ValueError as exc:
         return _failure_payload(

loopx/control_plane/quota/settlement.py

@@ -467,6 +467,7 @@ def read_heartbeat_settlement(
     borrow_source_admission: bool = False,
     heartbeat_reentry_guard: dict[str, Any] | None = None,
     include_record_bodies: bool = True,
+    tolerate_malformed_history: bool = False,
 ) -> QuotaSettlementReadback | None:
     """Read one complete heartbeat settlement through the TS domain owner."""
 
@@ -485,6 +486,11 @@ def read_heartbeat_settlement(
                 if include_record_bodies is not True
                 else {}
             ),
+            **(
+                {"tolerate_malformed_history": True}
+                if tolerate_malformed_history
+                else {}
+            ),
             **({"resolve_original_binding": True} if resolve_original_binding else {}),
             **({"refresh_retry": refresh_retry} if refresh_retry is not None else {}),
             **({"borrow_source_admission": True} if borrow_source_admission else {}),

loopx/control_plane/quota/settlement_readback.ts

@@ -110,6 +110,7 @@ interface ReadbackRequest {
   resolve_original_binding: boolean;
   borrow_source_admission: boolean;
   include_record_bodies: boolean;
+  tolerate_malformed_history: boolean;
   refresh_retry: RefreshRetryRequest | null;
   heartbeat_reentry_guard: JsonObject | null;
   owner: QuotaAccountingOwner;
@@ -224,6 +225,14 @@ function decodeRequest(
   if (request.include_record_bodies !== undefined && typeof request.include_record_bodies !== "boolean") {
     throw new EffectRuntimeRequestError("include_record_bodies must be a boolean");
   }
+  if (
+    request.tolerate_malformed_history !== undefined &&
+    typeof request.tolerate_malformed_history !== "boolean"
+  ) {
+    throw new EffectRuntimeRequestError(
+      "tolerate_malformed_history must be a boolean",
+    );
+  }
   if (request.resolve_original_binding === true && request.infer_turn_instance_id) {
     throw new EffectRuntimeRequestError("original binding requires an explicit Turn identity");
   }
@@ -283,16 +292,20 @@ function decodeRequest(
     resolve_original_binding: request.resolve_original_binding === true,
     borrow_source_admission: request.borrow_source_admission === true,
     include_record_bodies: request.include_record_bodies !== false,
+    tolerate_malformed_history: request.tolerate_malformed_history === true,
     refresh_retry: decodeRefreshRetry(request.refresh_retry),
     heartbeat_reentry_guard: request.heartbeat_reentry_guard == null ? null
       : requireJsonObject(request.heartbeat_reentry_guard, "heartbeat_reentry_guard"),
     owner,
   };
 }
 
-async function readRunReceipts(path: string): Promise<readonly JsonObject[]> {
+async function readRunReceipts(
+  path: string,
+  tolerateMalformedHistory: boolean = false,
+): Promise<readonly JsonObject[]> {
   const snapshot = await readReceiptLogSnapshot(path);
-  if (snapshot?.firstErrorLine != null) {
+  if (!tolerateMalformedHistory && snapshot?.firstErrorLine != null) {
     throw new EffectRuntimeRequestError(
       `settlement readback line ${snapshot.firstErrorLine} is malformed`,
       "malformed_settlement_state",
@@ -373,6 +386,7 @@ export async function readQuotaSettlementSnapshot(
   runtimeRoot: string,
   goalId: string,
   rolloutSnapshot?: GoalRolloutEventSnapshot | null,
+  tolerateMalformedHistory: boolean = false,
 ): Promise<QuotaSettlementReadbackSnapshot> {
   settlementScope(runtimeRoot, goalId);
   if (
@@ -393,12 +407,17 @@ export async function readQuotaSettlementSnapshot(
     rolloutSnapshot === undefined
       ? readGoalRolloutEventSnapshot(runtimeRoot, goalId)
       : Promise.resolve(rolloutSnapshot),
-    readRunReceipts(join(goalRoot, "runs", "index.jsonl")),
+    readRunReceipts(join(goalRoot, "runs", "index.jsonl"), tolerateMalformedHistory),
   ]);
+  // Settlement-authority callers keep failing closed on the first damaged row.
+  // The scheduler follow-up route only needs the receipts it can actually read,
+  // so it opts into the same tolerant view the heartbeat reader already uses.
   return indexSettlementSnapshot(
     runtimeRoot,
     goalId,
-    strictGoalRolloutEvents(snapshot),
+    tolerateMalformedHistory
+      ? snapshot?.events ?? []
+      : strictGoalRolloutEvents(snapshot),
     runs,
   );
 }
@@ -1446,7 +1465,12 @@ export async function readQuotaSettlement(value: unknown): Promise<JsonObject> {
     request.owner,
     async () => {
       const result = readQuotaSettlementFromRequest(request,
-        await readQuotaSettlementSnapshot(request.runtime_root, request.goal_id));
+        await readQuotaSettlementSnapshot(
+          request.runtime_root,
+          request.goal_id,
+          undefined,
+          request.tolerate_malformed_history,
+        ));
       const identity = jsonObject(jsonObject(jsonObject(result.identity)?.result)?.value);
       if (identity && request.refresh_retry &&
           jsonObject(result.refresh_recovery)?.decision !== "reject" &&

loopx/control_plane/rollout_receipt_log.ts

@@ -11,7 +11,7 @@ import { relative, resolve, sep } from "node:path";
 
 import type { JsonObject } from "./effect_program.ts";
 import { EffectRuntimeRequestError } from "./effect_runtime_errors.ts";
-import { requireNonEmptyString } from "./runtime_decode.ts";
+import { jsonObject, requireNonEmptyString } from "./runtime_decode.ts";
 import { readReceiptLogSnapshot } from "./runtime/receipt_log_snapshot.ts";
 
 export const ROLLOUT_EVENT_SCHEMA_VERSION = "loopx_rollout_event_v0";
@@ -154,12 +154,23 @@ export async function readGoalHeartbeatReceipts(
   );
 }
 
+/** A receipt without an exact Goal instance belongs to the legacy goal scope. */
+function receiptMatchesGoalRef(event: JsonObject, goalRef: JsonObject | null): boolean {
+  if (goalRef === null) return !Object.hasOwn(event, "goal_ref");
+  const eventGoalRef = jsonObject(event.goal_ref);
+  return eventGoalRef !== null &&
+    eventGoalRef.goal_id === goalRef.goal_id &&
+    eventGoalRef.goal_instance_id === goalRef.goal_instance_id;
+}
+
 export async function heartbeatReceiptStatus(params: {
   runtimeRoot: string;
   goalId: string;
   agentId: string;
   turnInstanceId: string;
+  goalRef?: JsonObject | null;
 }): Promise<HeartbeatReceiptStatus> {
+  const goalRef = params.goalRef ?? null;
   const snapshot = await readGoalRolloutEventSnapshot(
     params.runtimeRoot,
     params.goalId,
@@ -171,7 +182,8 @@ export async function heartbeatReceiptStatus(params: {
   );
   if (snapshot === null || receipts === null) return "missing";
   const firstReceipt = receipts.find(
-    (event) => event.run_id === params.turnInstanceId,
+    (event) => event.run_id === params.turnInstanceId &&
+      receiptMatchesGoalRef(event, goalRef),
   );
   const firstMatch = firstReceipt === undefined
     ? -1
@@ -181,7 +193,8 @@ export async function heartbeatReceiptStatus(params: {
     if (
       event.event_kind !== HEARTBEAT_RECEIPT_EVENT_KIND ||
       event.goal_id !== params.goalId ||
-      event.agent_id !== params.agentId
+      event.agent_id !== params.agentId ||
+      !receiptMatchesGoalRef(event, goalRef)
     ) return false;
     if (typeof event.run_id === "string" && event.run_id.trim().length > 0) {
       return event.run_id !== params.turnInstanceId;