diff --git a/loopx/cli_commands/quota_context.py b/loopx/cli_commands/quota_context.py index e7b84085f9..a26fe1a6b9 100644 --- a/loopx/cli_commands/quota_context.py +++ b/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"]) diff --git a/loopx/cli_commands/quota_scheduler_followup.py b/loopx/cli_commands/quota_scheduler_followup.py index a8ebbe64a1..f960c828bc 100644 --- a/loopx/cli_commands/quota_scheduler_followup.py +++ b/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, ) diff --git a/loopx/control_plane/quota/live_decision.py b/loopx/control_plane/quota/live_decision.py index 5498b6ea28..3cc871f140 100644 --- a/loopx/control_plane/quota/live_decision.py +++ b/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 diff --git a/loopx/control_plane/quota/scheduler_ack.py b/loopx/control_plane/quota/scheduler_ack.py index b9bf45a87d..34cb5f4b12 100644 --- a/loopx/control_plane/quota/scheduler_ack.py +++ b/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( diff --git a/loopx/control_plane/quota/settlement.py b/loopx/control_plane/quota/settlement.py index d8e1710326..3791537532 100644 --- a/loopx/control_plane/quota/settlement.py +++ b/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 {}), diff --git a/loopx/control_plane/quota/settlement_readback.ts b/loopx/control_plane/quota/settlement_readback.ts index d77a381bc3..4002a55d4f 100644 --- a/loopx/control_plane/quota/settlement_readback.ts +++ b/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,6 +292,7 @@ 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"), @@ -290,9 +300,12 @@ function decodeRequest( }; } -async function readRunReceipts(path: string): Promise { +async function readRunReceipts( + path: string, + tolerateMalformedHistory: boolean = false, +): Promise { 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 { 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 { 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" && diff --git a/loopx/control_plane/rollout_receipt_log.ts b/loopx/control_plane/rollout_receipt_log.ts index a02c8464cc..494b1a317c 100644 --- a/loopx/control_plane/rollout_receipt_log.ts +++ b/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 { + 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; diff --git a/loopx/control_plane/runtime/receipt_log_snapshot.ts b/loopx/control_plane/runtime/receipt_log_snapshot.ts index da0495f76e..d5e9f18ba8 100644 --- a/loopx/control_plane/runtime/receipt_log_snapshot.ts +++ b/loopx/control_plane/runtime/receipt_log_snapshot.ts @@ -59,6 +59,19 @@ function freezeRecord(record: JsonObject): JsonObject { return record; } +// One strict decoder for the whole read: replacement characters must never +// become a different, valid-looking receipt identity. `ignoreBOM` matches +// Buffer#toString, so a leading BOM still fails JSON parsing as before. +const STRICT_UTF8 = new TextDecoder("utf-8", { fatal: true, ignoreBOM: true }); + +function strictUtf8(bytes: Uint8Array): string | null { + try { + return STRICT_UTF8.decode(bytes); + } catch { + return null; + } +} + function decodeLine(line: string, schemaVersion?: string): JsonObject { const record = jsonObject(JSON.parse(line)); if (record === null || (schemaVersion !== undefined && @@ -106,9 +119,11 @@ export async function readReceiptLogSnapshot( let batchStart = offset; while (offset < completeBytes) { const end = bytes.indexOf(10, offset); - const line = bytes.subarray(offset, end).toString("utf8"); + const line = strictUtf8(bytes.subarray(offset, end)); lines += 1; - if (line.trim()) { + if (line === null) { + firstErrorLine ??= lines; + } else if (line.trim()) { try { records.push(decodeLine(line, schemaVersion)); } catch { @@ -128,7 +143,13 @@ export async function readReceiptLogSnapshot( }); } remember(key, prefix); - const tail = bytes.subarray(completeBytes).toString("utf8"); + const tail = strictUtf8(bytes.subarray(completeBytes)); + if (tail === null) { + return Object.freeze({ + records: prefix.records, + firstErrorLine: prefix.firstErrorLine ?? prefix.lines + 1, + }); + } if (!tail.trim()) return prefix; try { return Object.freeze({ diff --git a/loopx/control_plane/scheduler/heartbeat_commit.ts b/loopx/control_plane/scheduler/heartbeat_commit.ts index 2017c07143..f58b7c8650 100644 --- a/loopx/control_plane/scheduler/heartbeat_commit.ts +++ b/loopx/control_plane/scheduler/heartbeat_commit.ts @@ -3,6 +3,7 @@ import { readFile } from "node:fs/promises"; import type { JsonObject } from "../effect_program.ts"; import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import { parseExactGoalRef } from "../goals/goal_instance_identity.ts"; import { atomicWriteJson, withFileMutationLock, @@ -171,6 +172,7 @@ export interface SchedulerHeartbeatCommitResult extends JsonObject { export interface SchedulerHeartbeatCommitOptions { readonly heartbeatReceiptTurnInstanceId?: string | null; + readonly heartbeatReceiptGoalRef?: JsonObject | null; } function stableValue(value: unknown): unknown { @@ -831,6 +833,27 @@ export async function evaluateSchedulerHeartbeatCommit( options.heartbeatReceiptTurnInstanceId, "heartbeatReceiptTurnInstanceId", ).trim(); + let receiptGoalRef: JsonObject | null = null; + if (options.heartbeatReceiptGoalRef !== undefined && + options.heartbeatReceiptGoalRef !== null) { + const parsed = parseExactGoalRef(options.heartbeatReceiptGoalRef); + if (parsed.kind === "invalid") { + throw new EffectRuntimeRequestError( + `heartbeat receipt goal_ref is invalid: ${parsed.issue}`, + "heartbeat_receipt_goal_ref_invalid", + ); + } + receiptGoalRef = { + goal_id: parsed.value.goalId.value, + goal_instance_id: parsed.value.goalInstanceId.value, + }; + if (receiptGoalRef.goal_id !== request.scope.goalId) { + throw new EffectRuntimeRequestError( + "heartbeat receipt goal_ref does not match the scheduler scope", + "heartbeat_receipt_goal_ref_mismatch", + ); + } + } const path = schedulerStatePath(request.runtime_root, request.scope); const fingerprint = requestDigest(request); // Legacy state stays read-only until the commit holds every applicable @@ -853,6 +876,7 @@ export async function evaluateSchedulerHeartbeatCommit( goalId: request.scope.goalId, agentId: request.scope.agentId, turnInstanceId: receiptTurnInstanceId, + goalRef: receiptGoalRef, }); if (status !== "fresh") { return result( diff --git a/loopx/control_plane/scheduler/heartbeat_followup.ts b/loopx/control_plane/scheduler/heartbeat_followup.ts index d1047f0454..5b0cb72efa 100644 --- a/loopx/control_plane/scheduler/heartbeat_followup.ts +++ b/loopx/control_plane/scheduler/heartbeat_followup.ts @@ -1,5 +1,6 @@ import type { JsonObject } from "../effect_program.ts"; import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import { parseExactGoalRef } from "../goals/goal_instance_identity.ts"; import { goalPathSegment, heartbeatReceiptStatus, @@ -32,6 +33,7 @@ const FAILURE_CLASSIFICATION = "quota_scheduler_host_update_failure"; export interface SchedulerHeartbeatFollowupRequest extends JsonObject { schema_version: typeof SCHEDULER_HEARTBEAT_FOLLOWUP_REQUEST_SCHEMA; runtime_root: string; + goal_ref: JsonObject | null; turn_instance_id: string | null; require_heartbeat_receipt: boolean; before: JsonObject; @@ -81,6 +83,27 @@ function requestObject(value: unknown): SchedulerHeartbeatFollowupRequest { goalPathSegment(hostFacts.goal_id); requireNonEmptyString(hostFacts.agent_id, "agent_id"); followupOperation(hostFacts); + const rawGoalRef = input.goal_ref; + let goalRef: JsonObject | null = null; + if (rawGoalRef !== undefined && rawGoalRef !== null) { + const parsed = parseExactGoalRef(rawGoalRef); + if (parsed.kind === "invalid") { + throw new EffectRuntimeRequestError( + `scheduler follow-up goal_ref is invalid: ${parsed.issue}`, + "scheduler_followup_goal_ref_invalid", + ); + } + goalRef = { + goal_id: parsed.value.goalId.value, + goal_instance_id: parsed.value.goalInstanceId.value, + }; + if (goalRef.goal_id !== String(hostFacts.goal_id)) { + throw new EffectRuntimeRequestError( + "scheduler follow-up goal_ref does not match goal_id", + "scheduler_followup_goal_ref_mismatch", + ); + } + } const turnInstanceId = optionalText(input.turn_instance_id); const requireReceipt = input.require_heartbeat_receipt === undefined ? turnInstanceId !== null @@ -94,6 +117,7 @@ function requestObject(value: unknown): SchedulerHeartbeatFollowupRequest { return { schema_version: SCHEDULER_HEARTBEAT_FOLLOWUP_REQUEST_SCHEMA, runtime_root: runtimeRoot, + goal_ref: goalRef, turn_instance_id: turnInstanceId, require_heartbeat_receipt: requireReceipt, before: requireJsonObject(input.before ?? {}, "before"), @@ -341,6 +365,7 @@ export async function evaluateSchedulerHeartbeatFollowup( goalId: String(facts.goal_id), agentId: String(facts.agent_id), turnInstanceId: String(request.turn_instance_id), + goalRef: request.goal_ref, }); if (status !== "fresh") return receiptFailure(request, status); } @@ -351,6 +376,9 @@ export async function evaluateSchedulerHeartbeatFollowup( heartbeatReceiptTurnInstanceId: request.require_heartbeat_receipt ? request.turn_instance_id : null, + heartbeatReceiptGoalRef: request.require_heartbeat_receipt + ? request.goal_ref + : null, }); const receiptStatus = commitReceiptFailureStatus(commit); if (receiptStatus !== null) return receiptFailure(request, receiptStatus); diff --git a/loopx/control_plane/status/collection.py b/loopx/control_plane/status/collection.py index 7ddd538a4b..3ad3f1973b 100644 --- a/loopx/control_plane/status/collection.py +++ b/loopx/control_plane/status/collection.py @@ -71,6 +71,7 @@ class StatusCollectionContext: build_goal_channel_notification_projection: StatusCallback status_control_plane_context_limit: int max_todo_index_items: int + tolerate_malformed_settlement_history: bool = False def collect_status( @@ -89,6 +90,7 @@ def collect_status( agent_lane_id: str | None = None, allow_missing_registry: bool = False, include_presentation_projections: bool = True, + tolerate_malformed_settlement_history: bool = False, ) -> dict[str, Any]: """Collect the status read model. @@ -187,6 +189,10 @@ def canonical_runtime_for_goal(current_goal_id: str) -> Path: current_registry=registry, registry_path=registry_path, include_goal_channel_projection=include_presentation_projections, + tolerate_malformed_settlement_history=( + tolerate_malformed_settlement_history + or context.tolerate_malformed_settlement_history + ), ) # No later projection consumes the shared status snapshot. del todo_snapshot diff --git a/loopx/control_plane/work_items/attention_queue.py b/loopx/control_plane/work_items/attention_queue.py index fb9976b9b9..fb7a76a115 100644 --- a/loopx/control_plane/work_items/attention_queue.py +++ b/loopx/control_plane/work_items/attention_queue.py @@ -50,6 +50,7 @@ class AttentionQueueContext: monitor_signal_waiting_on: str registry_path: Path | None = None external_progress_review_context: Optional[Callable[..., dict[str, Any] | None]] = None + tolerate_malformed_settlement_history: bool = False def merge_global_registry_findings( @@ -335,7 +336,14 @@ def build_attention_queue( next_action_projection_warning=context.next_action_projection_warning, autonomous_replan_obligation_from_runs=context.autonomous_replan_obligation_from_runs, external_progress_review=external_progress_review, - effective_turn_cadence=effective_turn_cadence_context(goal, runtime_root, registry_path=context.registry_path), + effective_turn_cadence=effective_turn_cadence_context( + goal, + runtime_root, + registry_path=context.registry_path, + tolerate_malformed_history=( + context.tolerate_malformed_settlement_history + ), + ), ) item["quota"] = context.quota_status( goal, diff --git a/loopx/control_plane/work_items/replan_history_codec.py b/loopx/control_plane/work_items/replan_history_codec.py index c2f2d446d9..5d15b051cd 100644 --- a/loopx/control_plane/work_items/replan_history_codec.py +++ b/loopx/control_plane/work_items/replan_history_codec.py @@ -35,6 +35,7 @@ def effective_turn_cadence_context( registry_path: Path | None = None, goal_ref: dict[str, str] | None = None, source_admission: dict[str, Any] | None = None, + tolerate_malformed_history: bool = False, ) -> dict[str, Any] | None: """Encode a Goal whose live cadence was resolved by the caller.""" profile = goal.get("execution_profile") or {} @@ -55,6 +56,11 @@ def effective_turn_cadence_context( "settlement_source": { "runtime_root": str(runtime_root.resolve()), "goal_id": goal["id"], + **( + {"tolerate_malformed_history": True} + if tolerate_malformed_history + else {} + ), }, } diff --git a/loopx/control_plane/work_items/replan_history_settlement.ts b/loopx/control_plane/work_items/replan_history_settlement.ts index b09652f49e..1327730403 100644 --- a/loopx/control_plane/work_items/replan_history_settlement.ts +++ b/loopx/control_plane/work_items/replan_history_settlement.ts @@ -22,6 +22,11 @@ export async function projectSettledReplanHistory(value: unknown): Promise quotaOwnerOwnsProjection(owner, jsonObject(raw)?.goal_ref)); const rolloutSnapshot = await readGoalRolloutEventSnapshot(runtimeRoot, goalId); - const events = strictGoalRolloutEvents(rolloutSnapshot); + // Scheduler writeback returns only need the receipts they can read; a + // damaged row is never a receipt. Settlement-authority callers keep the + // strict view. + const events = tolerateMalformedHistory === true + ? rolloutSnapshot?.events ?? [] + : strictGoalRolloutEvents(rolloutSnapshot); const receiptKeys = new Set(events.filter(event => event.event_kind === "quota_should_run" && event.goal_id === goalId && typeof event.agent_id === "string" && typeof event.run_id === "string" && @@ -49,7 +59,12 @@ export async function projectSettledReplanHistory(value: unknown): Promise(); const seen = new Set(); for (const raw of settlementRuns) { diff --git a/loopx/quota.py b/loopx/quota.py index eded708cd4..7b175a83c9 100644 --- a/loopx/quota.py +++ b/loopx/quota.py @@ -1213,6 +1213,8 @@ def record_quota_scheduler_ack( | None = None, operator_inbox_urgency_projector: Callable[..., dict[str, Any]] | None = None, before_decision: Mapping[str, Any] | None = None, + turn_instance_id: str | None = None, + goal_ref: Mapping[str, Any] | None = None, ) -> dict[str, Any]: safe_goal_id = _validate_goal_id_path_segment(str(goal_id or "")) safe_agent_id = normalize_todo_claimed_by(agent_id) @@ -1247,6 +1249,8 @@ def record_quota_scheduler_ack( reason_summary=reason_summary, use_current_hint=use_current_hint, host_match_observed=host_match_observed, + turn_instance_id=turn_instance_id, + goal_ref=goal_ref, ) diff --git a/loopx/status.py b/loopx/status.py index c64de931c4..07be2e4da3 100644 --- a/loopx/status.py +++ b/loopx/status.py @@ -1093,6 +1093,7 @@ def build_attention_queue( current_registry: dict[str, Any] | None = None, registry_path: Path | None = None, include_goal_channel_projection: bool = True, + tolerate_malformed_settlement_history: bool = False, ) -> dict[str, Any]: def request_active_state_todo_fields( goal: dict[str, Any], @@ -1155,6 +1156,9 @@ def request_active_state_todo_fields( monitor_signal_waiting_on=MONITOR_SIGNAL_WAITING_ON, external_progress_review_context=external_progress_review_context, registry_path=registry_path, + tolerate_malformed_settlement_history=( + tolerate_malformed_settlement_history + ), ), runtime_root=runtime_root, include_task_graph=include_task_graph, @@ -1325,7 +1329,10 @@ def build_status_runtime_summaries( ) -def build_status_collection_context() -> StatusCollectionContext: +def build_status_collection_context( + *, + tolerate_malformed_settlement_history: bool = False, +) -> StatusCollectionContext: return StatusCollectionContext( load_registry=load_registry, resolve_runtime_root=resolve_runtime_root, @@ -1341,6 +1348,9 @@ def build_status_collection_context() -> StatusCollectionContext: build_goal_channel_notification_projection=build_goal_channel_notification_projection, status_control_plane_context_limit=STATUS_CONTROL_PLANE_CONTEXT_LIMIT, max_todo_index_items=MAX_TODO_INDEX_ITEMS, + tolerate_malformed_settlement_history=( + tolerate_malformed_settlement_history + ), ) @@ -1359,6 +1369,7 @@ def collect_status( agent_lane_id: str | None = None, allow_missing_registry: bool = False, include_presentation_projections: bool = True, + tolerate_malformed_settlement_history: bool = False, ) -> dict[str, Any]: with effect_runtime_request_scope(): return _collect_status_read_model( @@ -1377,5 +1388,9 @@ def collect_status( activation_state_filter=activation_state_filter, agent_lane_id=agent_lane_id, include_presentation_projections=include_presentation_projections, - context=build_status_collection_context(), + context=build_status_collection_context( + tolerate_malformed_settlement_history=( + tolerate_malformed_settlement_history + ), + ), ) diff --git a/tests/control_plane/test_scheduler_heartbeat_receipt_fence.py b/tests/control_plane/test_scheduler_heartbeat_receipt_fence.py new file mode 100644 index 0000000000..d23c018270 --- /dev/null +++ b/tests/control_plane/test_scheduler_heartbeat_receipt_fence.py @@ -0,0 +1,436 @@ +from __future__ import annotations + +import json +from argparse import Namespace +from pathlib import Path + +import pytest + +from examples.control_plane.quota_plan_fixtures import ( + SCOPED_AGENT_ID, + write_cli_fixture, +) +from loopx.cli_commands import quota_scheduler_followup +from loopx.control_plane.quota.scheduler_ack import _run_native_followup +from loopx.control_plane.testing.canary_harness import run_json_cli_result +from loopx.rollout_event_log import build_rollout_event, rollout_event_log_path + + +GOAL_ID = "receipt-fence" +AGENT_ID = "agent-a" + + +def _append(runtime_root: Path, **fields: object) -> None: + path = rollout_event_log_path(runtime_root, GOAL_ID) + path.parent.mkdir(parents=True, exist_ok=True) + event = build_rollout_event(goal_id=GOAL_ID, agent_id=AGENT_ID, **fields) + with path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(event, ensure_ascii=False) + "\n") + + +def test_native_followup_request_carries_the_turn_fence( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + captured: dict[str, object] = {} + + class _Completed: + returncode = 0 + stdout = '{"ok": true}' + stderr = "" + + class _Probe: + returncode = 0 + stdout = "v24.19.0" + stderr = "" + + def _run(command, **kwargs): + if "input" not in kwargs: + return _Probe() + captured["request"] = json.loads(kwargs["input"]) + return _Completed() + + monkeypatch.setattr( + "loopx.control_plane.quota.scheduler_ack.subprocess.run", _run + ) + goal_ref = { + "goal_id": "receipt-fence", + "goal_instance_id": "ginst_" + "a" * 32, + } + _run_native_followup( + runtime_root=tmp_path, + before={}, + host_facts={}, + use_current_hint=False, + reason_summary=None, + turn_instance_id="turn-1", + goal_ref=goal_ref, + ) + request = captured["request"] + assert request["turn_instance_id"] == "turn-1" + assert request["goal_ref"] == goal_ref + assert request["require_heartbeat_receipt"] is True + + _run_native_followup( + runtime_root=tmp_path, + before={}, + host_facts={}, + use_current_hint=False, + reason_summary=None, + ) + request = captured["request"] + assert request["turn_instance_id"] is None + assert request["goal_ref"] is None + assert request["require_heartbeat_receipt"] is False + + +def test_ack_route_forwards_the_turn_fence( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from loopx import quota + + captured: dict[str, object] = {} + + def _capture(before, **kwargs): + captured.update(kwargs) + return {"ok": True} + + monkeypatch.setattr( + quota, "record_quota_scheduler_ack_for_decision", _capture + ) + goal_ref = { + "goal_id": "goals", + "goal_instance_id": "ginst_" + "b" * 32, + } + quota.record_quota_scheduler_ack( + {"runtime_root": str(tmp_path)}, + goal_id="goals", + agent_id="agent-a", + before_decision={"decision": {}}, + turn_instance_id="turn-1", + goal_ref=goal_ref, + ) + assert captured["turn_instance_id"] == "turn-1" + assert captured["goal_ref"] == goal_ref + + +def _followup_args(**overrides: object) -> Namespace: + values: dict[str, object] = { + "quota_command": "scheduler-ack-current", + "goal_id": GOAL_ID, + "agent_id": AGENT_ID, + "codex_app_current_rrule": None, + "applied_rrule": "FREQ=MINUTELY;INTERVAL=3", + "host_match_observed": False, + "execute": True, + "surface": "trae_app", + "state_key": None, + "available_capabilities": [], + "reset_token": None, + "identity_signature": None, + "reason_summary": None, + "use_current_hint": False, + } + values.update(overrides) + return Namespace(**values) + + +def _capture_ack(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, object]]: + calls: list[dict[str, object]] = [] + + def _ack(_status, **kwargs): + calls.append(kwargs) + return {"ok": True} + + monkeypatch.setattr( + quota_scheduler_followup, "record_quota_scheduler_ack", _ack + ) + monkeypatch.setattr( + quota_scheduler_followup, + "_build_scheduler_followup_decision", + lambda *_args, **_kwargs: {"effective_action": "execute_todo"}, + ) + return calls + + +def test_superseded_receipt_is_decided_by_the_native_fence( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + runtime_root = tmp_path / "runtime" + _append(runtime_root, event_kind="quota_should_run", run_id="turn-1") + _append(runtime_root, event_kind="quota_should_run", run_id="turn-2") + calls = _capture_ack(monkeypatch) + + payload = quota_scheduler_followup.build_scheduler_followup_payload( + {}, + _followup_args(), + registry_path=tmp_path / "registry.json", + runtime_root=runtime_root, + turn_instance_id="turn-1", + scheduler_context=None, + operator_inbox_urgency_projector=lambda **_kwargs: {}, + ) + + # The Python route keeps no second freshness opinion: it binds the + # originating Turn and lets the native fence answer stale/missing. + assert payload == {"ok": True} + assert calls and calls[0]["turn_instance_id"] == "turn-1" + + +def test_scheduler_followup_builds_from_the_tolerant_settlement_view( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + captured: dict[str, object] = {} + + def _decision(*_args, **kwargs): + captured.update(kwargs) + return {"effective_action": "execute_todo"} + + monkeypatch.setattr( + quota_scheduler_followup, + "build_live_quota_should_run_decision", + _decision, + ) + quota_scheduler_followup._build_scheduler_followup_decision( + {}, + _followup_args(), + registry_path=tmp_path / "registry.json", + runtime_root=tmp_path / "runtime", + heartbeat_receipt=None, + turn_instance_id=None, + codex_app_current_rrule=None, + scheduler_context=None, + operator_inbox_urgency_projector=lambda **_kwargs: {}, + ) + assert captured["tolerate_malformed_settlement_history"] is True + + +def test_malformed_log_line_does_not_block_the_ack_route( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + runtime_root = tmp_path / "runtime" + _append(runtime_root, event_kind="quota_should_run", run_id="turn-1") + path = rollout_event_log_path(runtime_root, GOAL_ID) + with path.open("ab") as handle: + handle.write( + b'{"schema_version":"loopx_rollout_event_v0",' + b'"event_kind":"quota_should_run","goal_id":"receipt-fence",' + b'"agent_id":"agent-a","run_id":"turn-bad","summary":"' + b"\xff" + b'"}\n' + ) + calls = _capture_ack(monkeypatch) + + payload = quota_scheduler_followup.build_scheduler_followup_payload( + {}, + _followup_args(), + registry_path=tmp_path / "registry.json", + runtime_root=runtime_root, + turn_instance_id="turn-1", + scheduler_context=None, + operator_inbox_urgency_projector=lambda **_kwargs: {}, + ) + + assert payload == {"ok": True} + assert calls and calls[0]["turn_instance_id"] == "turn-1" + + +def test_fresh_receipt_still_reaches_the_ack_with_its_turn_fence( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + runtime_root = tmp_path / "runtime" + _append(runtime_root, event_kind="quota_should_run", run_id="turn-1") + calls: list[dict[str, object]] = [] + + def _ack(_status, **kwargs): + calls.append(kwargs) + return {"ok": True} + + monkeypatch.setattr( + quota_scheduler_followup, "record_quota_scheduler_ack", _ack + ) + monkeypatch.setattr( + quota_scheduler_followup, + "_build_scheduler_followup_decision", + lambda *_args, **_kwargs: {"effective_action": "execute_todo"}, + ) + args = Namespace( + quota_command="scheduler-ack-current", + goal_id=GOAL_ID, + agent_id=AGENT_ID, + codex_app_current_rrule=None, + applied_rrule="FREQ=MINUTELY;INTERVAL=3", + host_match_observed=False, + execute=True, + surface="trae_app", + state_key=None, + available_capabilities=[], + reset_token=None, + identity_signature=None, + reason_summary=None, + use_current_hint=False, + ) + payload = quota_scheduler_followup.build_scheduler_followup_payload( + {}, + args, + registry_path=tmp_path / "registry.json", + runtime_root=runtime_root, + turn_instance_id="turn-1", + scheduler_context=None, + operator_inbox_urgency_projector=lambda **_kwargs: {}, + ) + + assert payload == {"ok": True} + assert calls and calls[0]["turn_instance_id"] == "turn-1" + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize( + ("command", "include_turn"), + [ + ("scheduler-ack-current", True), + ("scheduler-fail-current", True), + ("scheduler-ack", True), + ("scheduler-ack", False), + ], +) +def test_real_compatibility_cli_survives_a_damaged_audit_row( + tmp_path: Path, provider: str, command: str, include_turn: bool +) -> None: + registry, runtime_root, project = write_cli_fixture( + tmp_path / provider, scoped_agents=True, provider=provider, + ) + goal_id = "needs-operator" + turn_id = "turn-damaged-history" + code, guard = run_json_cli_result( + "quota", + "should-run", + "--goal-id", + goal_id, + "--agent-id", + SCOPED_AGENT_ID, + "--trae_app", + "--app-automation-current-rrule", + "FREQ=MINUTELY;INTERVAL=3", + "--turn-instance-id", + turn_id, + registry_path=registry, + runtime_root=runtime_root, + cwd=project, + ) + assert code == 0, guard + + log_path = rollout_event_log_path(runtime_root, goal_id) + with log_path.open("ab") as handle: + handle.write( + b'{"schema_version":"loopx_rollout_event_v0",' + b'"event_kind":"quota_should_run","goal_id":"needs-operator",' + b'"agent_id":"' + SCOPED_AGENT_ID.encode() + b'",' + b'"run_id":"later-turn","summary":"' + b"\xff" + b'"}\n' + ) + + # The ACK derives its applied RRULE from the current hint; the failure + # return records the hint target. Both are ordinary scheduler writebacks. + operation_args = ("--use-current-hint",) if command == "scheduler-ack" else () + turn_args = ("--turn-instance-id", turn_id) if include_turn else () + code, result = run_json_cli_result( + "quota", + command, + "--goal-id", + goal_id, + "--agent-id", + SCOPED_AGENT_ID, + "--trae_app", + *turn_args, + "--execute", + *operation_args, + registry_path=registry, + runtime_root=runtime_root, + cwd=project, + ) + assert code == 0, result + assert result["ok"] is True, result + assert result["scheduler_state_mutated"] is True + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_scheduler_followup_cache_preview_cannot_poison_strict_status( + tmp_path: Path, provider: str +) -> None: + registry, runtime_root, project = write_cli_fixture( + tmp_path / provider, scoped_agents=True, provider=provider, + ) + goal_id = "needs-operator" + turn_id = "turn-cache-preview" + code, guard = run_json_cli_result( + "quota", + "should-run", + "--goal-id", + goal_id, + "--agent-id", + SCOPED_AGENT_ID, + "--trae_app", + "--app-automation-current-rrule", + "FREQ=MINUTELY;INTERVAL=3", + "--turn-instance-id", + turn_id, + registry_path=registry, + runtime_root=runtime_root, + cwd=project, + ) + assert code == 0, guard + + log_path = rollout_event_log_path(runtime_root, goal_id) + with log_path.open("ab") as handle: + handle.write( + b'{"schema_version":"loopx_rollout_event_v0",' + b'"event_kind":"quota_should_run","goal_id":"needs-operator",' + b'"agent_id":"' + SCOPED_AGENT_ID.encode() + b'",' + b'"run_id":"later-turn","summary":"' + b"\xff" + b'"}\n' + ) + + before_code, before_status = run_json_cli_result( + "quota", + "status", + "--goal-id", + goal_id, + registry_path=registry, + runtime_root=runtime_root, + cwd=project, + ) + code, preview = run_json_cli_result( + "quota", + "scheduler-ack-current", + "--goal-id", + goal_id, + "--agent-id", + SCOPED_AGENT_ID, + "--trae_app", + "--turn-instance-id", + turn_id, + "--execute", + "--write-projection-cache", + registry_path=registry, + runtime_root=runtime_root, + cwd=project, + ) + assert code == 0, preview + cache = preview["status_projection_cache"] + assert cache["written"] is False + assert cache["skipped_reason"] == "tolerant_scheduler_followup_view" + assert not Path(cache["path"]).exists() + + after_code, after_status = run_json_cli_result( + "quota", + "status", + "--goal-id", + goal_id, + "--use-projection-cache", + registry_path=registry, + runtime_root=runtime_root, + cwd=project, + ) + assert after_code == before_code + assert after_status.get("status_projection_cache", {}).get("hit") is not True diff --git a/tests/control_plane_ts/quota_settlement_readback.test.ts b/tests/control_plane_ts/quota_settlement_readback.test.ts index bcea7eb4c7..4ac27c75ce 100644 --- a/tests/control_plane_ts/quota_settlement_readback.test.ts +++ b/tests/control_plane_ts/quota_settlement_readback.test.ts @@ -558,6 +558,31 @@ test("scoped supersede settles only its exact Turn and never terminal acceptance }); }); +test("only the tolerant readback route survives a damaged audit row", async () => { + const root = await fixture({writeback: true, spend: true}); + try { + const clean = await readQuotaSettlement(request(root)); + const path = join(root, "goals", goalId, "rollout-event-log.jsonl"); + await appendFile(path, Buffer.concat([ + Buffer.from('{"schema_version":"loopx_rollout_event_v0","event_kind":"quota_should_run","goal_id":"'), + Buffer.from(goalId), + Buffer.from('","agent_id":"'), + Buffer.from(agentId), + Buffer.from('","run_id":"later-turn","summary":"'), + Buffer.from([0xff]), + Buffer.from('"}\n'), + ])); + await assert.rejects(readQuotaSettlement(request(root)), { + code: "malformed_settlement_state", + }); + const tolerant = await readQuotaSettlement(request(root, { + tolerate_malformed_history: true, + })); + assert.deepEqual(tolerant, clean, + "a damaged row must not change the settlement facts a valid return can read"); + } finally { await rm(root, {recursive: true, force: true}); } +}); + test("settlement progress requires both effects and exact receipts", async t => { const cases = [ { guard: false, state: "identity_required", next: "validation" }, diff --git a/tests/control_plane_ts/receipt_log_snapshot.test.ts b/tests/control_plane_ts/receipt_log_snapshot.test.ts index a5119950d2..ccd792882c 100644 --- a/tests/control_plane_ts/receipt_log_snapshot.test.ts +++ b/tests/control_plane_ts/receipt_log_snapshot.test.ts @@ -95,6 +95,42 @@ test("schema, malformed line numbers and tolerant valid rows survive reuse", asy }); }); +test("invalid UTF-8 rows are malformed, never receipts", async () => { + await fixture(async (path) => { + const valid = Buffer.from('{"run_id":"turn-1"}\n'); + const bad = Buffer.concat([ + Buffer.from('{"run_id":"turn-2","summary":"'), + Buffer.from([0xff]), + Buffer.from('"}\n'), + ]); + await writeFile(path, Buffer.concat([valid, bad])); + const snapshot = (await read(path))!; + assert.deepEqual(snapshot.records.map((r) => r.run_id), ["turn-1"]); + assert.equal(snapshot.firstErrorLine, 2); + // Cached prefix reuse must not resurrect or forget the bad row. + const reused = (await read(path))!; + assert.equal(reused.firstErrorLine, 2); + assert.deepEqual(reused.records.map((r) => r.run_id), ["turn-1"]); + // An unterminated tail with invalid bytes is malformed, never a record. + await writeFile(path, Buffer.concat([ + valid, + Buffer.from('{"run_id":"turn-3","summary":"'), + Buffer.from([0xff]), + ])); + const tail = (await read(path))!; + assert.deepEqual(tail.records.map((r) => r.run_id), ["turn-1"]); + assert.equal(tail.firstErrorLine, 2); + // Terminating the damaged tail keeps it malformed but never hides a later row. + await appendFile(path, Buffer.from('\n{"run_id":"turn-4"}\n')); + const completed = (await read(path))!; + assert.deepEqual( + completed.records.map((r) => r.run_id), + ["turn-1", "turn-4"], + ); + assert.equal(completed.firstErrorLine, 2); + }); +}); + test("eviction and concurrent cold readers affect cost, never evidence", async () => { await fixture(async (path) => { const row = JSON.stringify({payload: "x".repeat(2048)}) + "\n"; diff --git a/tests/control_plane_ts/scheduler_heartbeat_followup.test.ts b/tests/control_plane_ts/scheduler_heartbeat_followup.test.ts index a636ce3df3..19208ad53d 100644 --- a/tests/control_plane_ts/scheduler_heartbeat_followup.test.ts +++ b/tests/control_plane_ts/scheduler_heartbeat_followup.test.ts @@ -1,5 +1,6 @@ import assert from "node:assert/strict"; import { + appendFile, mkdir, mkdtemp, readFile, @@ -68,7 +69,11 @@ function receiptPath(runtimeRoot: string): string { return join(runtimeRoot, "goals", scope.goal_id, "rollout-event-log.jsonl"); } -async function appendReceipt(runtimeRoot: string, turnId: string): Promise { +async function appendReceipt( + runtimeRoot: string, + turnId: string, + goalRef?: Record, +): Promise { const path = receiptPath(runtimeRoot); await mkdir(dirname(path), { recursive: true }); await writeFile(path, `${JSON.stringify({ @@ -77,6 +82,7 @@ async function appendReceipt(runtimeRoot: string, turnId: string): Promise goal_id: scope.goal_id, agent_id: scope.agent_id, run_id: turnId, + ...(goalRef === undefined ? {} : { goal_ref: goalRef }), })}\n`, { encoding: "utf8", flag: "a" }); } @@ -226,6 +232,77 @@ test("missing or superseded heartbeat receipt rejects before scheduler mutation" assert.equal(JSON.parse(await readFile(legacyPath, "utf8")).goal_id, scope.goal_id); }); +async function appendMalformedReceipt( + runtimeRoot: string, + turnId: string, +): Promise { + const path = receiptPath(runtimeRoot); + await mkdir(dirname(path), { recursive: true }); + // Structurally a newer receipt, but the summary carries one raw 0xff byte. + const head = Buffer.from( + `{"schema_version":"loopx_rollout_event_v0","event_kind":"quota_should_run",` + + `"goal_id":"${scope.goal_id}","agent_id":"${scope.agent_id}",` + + `"run_id":"${turnId}","summary":"`, + ); + await appendFile(path, Buffer.concat([head, Buffer.from([0xff]), Buffer.from('"}\n')])); +} + +test("a malformed newer log row cannot block a still-valid ACK or failure", async (t) => { + const runtimeRoot = await tempRuntime(t); + await appendReceipt(runtimeRoot, "turn-followup-1"); + const base = request(runtimeRoot); + const statePath = schedulerStatePath(runtimeRoot, { + goalId: scope.goal_id, + agentId: scope.agent_id, + surface: scope.surface, + stateKey: scope.state_key, + }); + + await appendMalformedReceipt(runtimeRoot, "turn-bad-bytes"); + const snapshot = await readGoalRolloutEventSnapshot(runtimeRoot, scope.goal_id); + assert.equal(snapshot!.firstStrictErrorLine, 2); + assert.deepEqual( + goalHeartbeatReceiptsFromSnapshot(snapshot, scope.goal_id)!.map((r) => r.run_id), + ["turn-followup-1"], + ); + // Strict settlement readback still reports the damaged row. + assert.throws(() => strictGoalRolloutEvents(snapshot), { + code: "malformed_settlement_state", + }); + + const ack = await evaluateSchedulerHeartbeatFollowup(base); + assert.equal(ack.ok, true); + assert.equal((ack.scheduler_commit as Record).status, "written"); + + const failure = await evaluateSchedulerHeartbeatFollowup({ + ...base, + host_facts: { + ...(base.host_facts as Record), + operation: "host_failure", + applied_rrule: "FREQ=MINUTELY;INTERVAL=3", + observed_host_rrule: "FREQ=MINUTELY;INTERVAL=3", + failure_kind: "timeout", + source: "quota_scheduler_host_update_failure", + }, + }); + assert.equal(failure.ok, true); + assert.equal(failure.mode, "scheduler-fail-current"); + + // A genuinely newer receipt still fences both writebacks. + await appendReceipt(runtimeRoot, "turn-followup-2"); + const staleAck = await evaluateSchedulerHeartbeatFollowup({ + ...base, + host_facts: { + ...(base.host_facts as Record), + generated_at: "2026-08-27T06:35:00Z", + }, + }); + assert.equal(staleAck.ok, false); + assert.equal(staleAck.status, "heartbeat_receipt_stale"); + assert.equal(staleAck.scheduler_state_mutated, false); + assert.equal(JSON.parse(await readFile(statePath, "utf8")).goal_id, scope.goal_id); +}); + test("receipt-bound ACK rechecks freshness after waiting for the scheduler lock", async (t) => { const runtimeRoot = await tempRuntime(t); await appendReceipt(runtimeRoot, "turn-followup-1"); @@ -253,6 +330,87 @@ test("receipt-bound ACK rechecks freshness after waiting for the scheduler lock" await assert.rejects(readFile(path, "utf8"), { code: "ENOENT" }); }); +test("another Goal instance cannot authorize or supersede an exact-instance receipt", async (t) => { + const runtimeRoot = await tempRuntime(t); + const goalRefA = { + goal_id: scope.goal_id, + goal_instance_id: `ginst_${"a".repeat(32)}`, + }; + const goalRefB = { + goal_id: scope.goal_id, + goal_instance_id: `ginst_${"b".repeat(32)}`, + }; + const statePath = schedulerStatePath(runtimeRoot, { + goalId: scope.goal_id, + agentId: scope.agent_id, + surface: scope.surface, + stateKey: scope.state_key, + }); + await appendReceipt(runtimeRoot, "turn-followup-1", goalRefB); + + const foreign = await evaluateSchedulerHeartbeatFollowup(request(runtimeRoot, { + goal_ref: goalRefA, + })); + assert.equal(foreign.ok, false); + assert.equal(foreign.status, "heartbeat_receipt_missing"); + assert.equal(foreign.scheduler_state_mutated, false); + await assert.rejects(readFile(statePath, "utf8"), { code: "ENOENT" }); + + await appendReceipt(runtimeRoot, "turn-followup-1", goalRefA); + const fresh = await evaluateSchedulerHeartbeatFollowup(request(runtimeRoot, { + goal_ref: goalRefA, + })); + assert.equal(fresh.ok, true); + assert.equal((fresh.scheduler_commit as Record).status, "written"); + + // A newer receipt for another Goal lifetime is outside this transaction's + // freshness scope and cannot turn the exact instance's receipt stale. + await appendReceipt(runtimeRoot, "turn-followup-2", goalRefB); + const stillFresh = await evaluateSchedulerHeartbeatFollowup(request(runtimeRoot, { + goal_ref: goalRefA, + host_facts: { + ...((request(runtimeRoot).host_facts) as Record), + generated_at: "2026-08-27T06:35:00Z", + }, + })); + assert.equal(stillFresh.ok, true); +}); + +test("the scheduler lock rechecks freshness inside the exact Goal instance", async (t) => { + const runtimeRoot = await tempRuntime(t); + const goalRefA = { + goal_id: scope.goal_id, + goal_instance_id: `ginst_${"c".repeat(32)}`, + }; + const goalRefB = { + goal_id: scope.goal_id, + goal_instance_id: `ginst_${"d".repeat(32)}`, + }; + const exactRequest = request(runtimeRoot, { goal_ref: goalRefA }); + await appendReceipt(runtimeRoot, "turn-followup-1", goalRefA); + const path = schedulerStatePath(runtimeRoot, { + goalId: scope.goal_id, + agentId: scope.agent_id, + surface: scope.surface, + stateKey: scope.state_key, + }); + const held = await acquireFileMutationLock(path); + let released = false; + t.after(async () => { + if (!released) await releaseFileMutationLock(held.targetPath, held.token); + }); + + const inFlight = evaluateSchedulerHeartbeatFollowup(exactRequest); + await delay(250); + await appendReceipt(runtimeRoot, "turn-followup-2", goalRefB); + released = await releaseFileMutationLock(held.targetPath, held.token); + + const written = await inFlight; + assert.equal(written.ok, true); + assert.equal((written.scheduler_commit as Record).status, "written"); + assert.equal(JSON.parse(await readFile(path, "utf8")).goal_id, scope.goal_id); +}); + test("decision-free compatibility requests can omit a heartbeat receipt", async (t) => { const runtimeRoot = await tempRuntime(t); const result = await evaluateSchedulerHeartbeatFollowup(request(runtimeRoot, {