diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 9fd40340b6..e926acbc42 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/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], diff --git a/loopx/control_plane/quota/heartbeat_receipt.py b/loopx/control_plane/quota/heartbeat_receipt.py index f902b35fd9..e9ab9d32e3 100644 --- a/loopx/control_plane/quota/heartbeat_receipt.py +++ b/loopx/control_plane/quota/heartbeat_receipt.py @@ -154,51 +154,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, diff --git a/loopx/control_plane/quota/heartbeat_receipt_identity.ts b/loopx/control_plane/quota/heartbeat_receipt_identity.ts new file mode 100644 index 0000000000..dc4cef2a72 --- /dev/null +++ b/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( + goalId: string, + agentId: string, + entries: readonly { fact: HeartbeatReceiptFact; value: Value }[], +): Value | null { + if (entries.length === 0) return null; + const identities = new Map(); + 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; +} diff --git a/loopx/control_plane/quota/settlement_readback.ts b/loopx/control_plane/quota/settlement_readback.ts index d7259bb2c9..b9e74b3510 100644 --- a/loopx/control_plane/quota/settlement_readback.ts +++ b/loopx/control_plane/quota/settlement_readback.ts @@ -1,6 +1,10 @@ import { readFile } from "node:fs/promises"; import { isAbsolute, join } from "node:path"; +import { + ROLLOUT_EVENT_SCHEMA_VERSION, + goalRolloutEventLogPath, +} from "../rollout_receipt_log.ts"; import { effectIdsMatch, settlementBindReduce, @@ -35,6 +39,15 @@ import { refreshRecovery, type RefreshRetryRequest, } from "./refresh_recovery.ts"; +import { + heartbeatReceiptDetails as details, + heartbeatReceiptFactFromEvent, + normalizeHeartbeatReplanObligationId as normalizeReplanObligationId, + normalizeHeartbeatTodoId as normalizeTodoId, + optionalHeartbeatString as optionalString, + selectEffectiveHeartbeatReceipt, + type HeartbeatReceiptFact, +} from "./heartbeat_receipt_identity.ts"; import { refreshExternalDelivery } from "./refresh_external_delivery.ts"; @@ -44,11 +57,8 @@ export const QUOTA_SETTLEMENT_READBACK_RESULT_SCHEMA = "loopx_quota_settlement_readback_result_v0"; export const SEMANTIC_REPLAN_GUARD_SCHEMA = "semantic_replan_guard_v0"; -const ROLLOUT_EVENT_SCHEMA_VERSION = "loopx_rollout_event_v0"; const TURN_INSTANCE_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/; -const TODO_ID_PATTERN = /^todo_[a-z0-9_-]{3,64}$/; const AGENT_ID_PATTERN = /^[a-z][a-z0-9_.:@-]{0,79}$/; -const REPLAN_OBLIGATION_ID_PATTERN = /^replan-[a-f0-9]{16}$/; interface ReadbackRequest { runtime_root: string; @@ -67,11 +77,6 @@ interface ResultBundle extends JsonObject { payload: JsonObject; } -function optionalString(value: unknown): string | null { - if (value === null || value === undefined) return null; - return String(value).trim() || null; -} - function optionalRequestString(value: unknown, label: string): string | null { if (value === null || value === undefined) return null; if (typeof value !== "string") { @@ -85,18 +90,6 @@ function normalizeAgentId(value: unknown): string | null { return candidate && AGENT_ID_PATTERN.test(candidate) ? candidate : null; } -function normalizeTodoId(value: unknown): string | null { - const candidate = String(value ?? "").trim().toLowerCase(); - return candidate && TODO_ID_PATTERN.test(candidate) ? candidate : null; -} - -function normalizeReplanObligationId(value: unknown): string | null { - const candidate = String(value ?? "").trim(); - return candidate && REPLAN_OBLIGATION_ID_PATTERN.test(candidate) - ? candidate - : null; -} - function decodeRequest(value: unknown): ReadbackRequest { const request = requireJsonObject(value, "quota settlement readback request"); if (request.schema_version !== QUOTA_SETTLEMENT_READBACK_REQUEST_SCHEMA) { @@ -209,10 +202,6 @@ function runEffectMatches( quotaSpendMetadataMatches(run.quota_spend_commit, effectRef, expectedEffectRef)); } -function details(event: JsonObject | null): JsonObject { - return jsonObject(event?.details) ?? {}; -} - export function projectSemanticReplanGuard( receiptDetails: JsonObject, ): JsonObject { @@ -245,63 +234,6 @@ export function projectSemanticReplanGuard( }; } -function receiptIdentity( - event: JsonObject, -): { key: string; event: JsonObject } | null { - const eventDetails = details(event); - const todoId = normalizeTodoId(eventDetails.todo_id); - const replanObligationId = normalizeReplanObligationId( - eventDetails.replan_obligation_id, - ); - const effectId = optionalString(eventDetails.settlement_effect_id); - if (todoId && replanObligationId) { - throw new EffectRuntimeRequestError( - "heartbeat receipt has conflicting Todo and autonomous replan bindings", - ); - } - if (effectId && !todoId && !replanObligationId) { - throw new EffectRuntimeRequestError( - "heartbeat receipt has an effect identity without a Todo or autonomous replan binding; refuse to infer or upgrade it", - ); - } - if (!todoId && !replanObligationId) return null; - const identity = settlementIdentity({ - goal_id: optionalString(event.goal_id) ?? "", - agent_id: optionalString(event.agent_id) ?? "", - todo_id: todoId, - turn_instance_id: optionalString(event.run_id) ?? "", - replan_obligation_id: replanObligationId, - }); - return { - key: `${identity.binding_kind}\u0000${identity.binding_id}\u0000${effectId ?? identity.effect_id}`, - event, - }; -} - -function effectiveHeartbeatReceipt( - events: readonly JsonObject[], - identity: Pick, -): JsonObject | null { - const matching = events.filter((event) => - event.event_kind === "quota_should_run" && - optionalString(event.goal_id) === identity.goal_id && - optionalString(event.agent_id) === identity.agent_id && - optionalString(event.run_id) === identity.turn_instance_id - ); - if (matching.length === 0) return null; - const identities = new Map(); - for (const event of matching) { - const resolved = receiptIdentity(event); - if (resolved) identities.set(resolved.key, resolved.event); - } - if (identities.size > 1) { - throw new EffectRuntimeRequestError( - "heartbeat receipt has conflicting settlement identities for the same goal, agent, and turn", - ); - } - return identities.size === 1 ? [...identities.values()][0] : matching.at(-1)!; -} - function runMatchesBinding(run: JsonObject, identity: SettlementIdentity): boolean { return normalizeTodoId(run.todo_id) === identity.todo_id && normalizeReplanObligationId(run.replan_obligation_id) === @@ -360,6 +292,29 @@ function findStepEvent( ) ?? null; } +/** + * Resolve the Turn's effective receipt, or null when it persisted no guard. + * + * The identity rule lives in one module, so the readback resolves the effective + * receipt through the same reduction the closeout selector uses. + */ +function effectiveHeartbeatReceipt( + events: readonly JsonObject[], + identity: Pick, +): JsonObject | null { + const entries: { fact: HeartbeatReceiptFact; value: JsonObject }[] = []; + for (const event of events) { + if (event.event_kind !== "quota_should_run") continue; + if (optionalString(event.goal_id) !== identity.goal_id) continue; + if (optionalString(event.agent_id) !== identity.agent_id) continue; + if (optionalString(event.run_id) !== identity.turn_instance_id) continue; + entries.push({ fact: heartbeatReceiptFactFromEvent(event), value: event }); + } + return entries.length === 0 + ? null + : selectEffectiveHeartbeatReceipt(identity.goal_id, identity.agent_id, entries); +} + function writebackResult( identity: SettlementIdentity, run: JsonObject | null, @@ -699,7 +654,10 @@ export async function readQuotaSettlement(value: unknown): Promise { const request = decodeRequest(value); const goalRoot = join(request.runtime_root, "goals", request.goal_id); const [events, runs] = await Promise.all([ - readJsonLines(join(goalRoot, "rollout-event-log.jsonl"), ROLLOUT_EVENT_SCHEMA_VERSION), + readJsonLines( + goalRolloutEventLogPath(request.runtime_root, request.goal_id), + ROLLOUT_EVENT_SCHEMA_VERSION, + ), readJsonLines(join(goalRoot, "runs", "index.jsonl")), ]); const identityResult = resolveIdentity(request, events, runs); diff --git a/loopx/control_plane/quota/unsettled_host_turn.py b/loopx/control_plane/quota/unsettled_host_turn.py index ef12753ae8..fa7e3e18e2 100644 --- a/loopx/control_plane/quota/unsettled_host_turn.py +++ b/loopx/control_plane/quota/unsettled_host_turn.py @@ -1,3 +1,13 @@ +"""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 +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 +what the recovery obligation is all come from the TypeScript owner. +""" + from __future__ import annotations from .effective_action import EffectiveAction @@ -5,23 +15,36 @@ from pathlib import Path from typing import Any +from ..effect_runtime import EffectRuntimeRejected, 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, build_protocol_action_packet, ) -from ..todos.contract import TODO_TASK_CLASS_MONITOR -from ..todos.todo_semantics import todo_item_task_class -from .heartbeat_receipt import ( - heartbeat_receipt_settlement_replan_obligation_id, - heartbeat_receipt_settlement_todo_id, - prior_closeout_required_heartbeat_receipts, -) -from .settlement import read_heartbeat_settlement +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" +PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_METHOD = ( + "quota.prior_host_turn_closeout.preflight" +) +PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_REQUEST_SCHEMA = ( + "loopx_prior_host_turn_closeout_preflight_request_v0" +) +PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_RESULT_SCHEMA = ( + "loopx_prior_host_turn_closeout_preflight_result_v0" +) +UNSETTLED_HOST_TURN_RECOVERY_METHOD = "quota.unsettled_host_turn_recovery.reduce" +UNSETTLED_HOST_TURN_RECOVERY_REQUEST_SCHEMA = ( + "loopx_prior_host_turn_recovery_request_v0" +) +UNSETTLED_HOST_TURN_RECOVERY_RESULT_SCHEMA = ( + "loopx_prior_host_turn_recovery_result_v0" +) + def _bound_todo_item( *, @@ -49,23 +72,7 @@ def _bound_todo_item( return dict(item) -def _typed_lifecycle_closeout(item: Mapping[str, Any] | None) -> str | None: - if item is None: - return None - status = str(item.get("status") or "") - if ( - status == "open" - and item.get("resume_when") - and isinstance(item.get("successor_todo_ids"), list) - and bool(item.get("successor_todo_ids")) - ): - return "typed_external_wait" - if status in {"done", "blocked", "deferred"}: - return f"todo_{status}" - return None - - -def _committed_monitor_poll_closeout( +def _committed_monitor_poll_fact( *, runtime_root: Path, goal_id: str, @@ -73,15 +80,17 @@ def _committed_monitor_poll_closeout( todo_id: str | None, prior_turn_instance_id: str, todo_item: Mapping[str, Any] | None, -) -> dict[str, Any] | None: - """Read an exact committed no-spend closeout for a monitor-bound Turn.""" +) -> 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 None + return {} receipt = find_quota_monitor_poll_turn( runtime_root, goal_id=goal_id, @@ -90,17 +99,77 @@ def _committed_monitor_poll_closeout( turn_instance_id=prior_turn_instance_id, ) if receipt is None: - return 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 - effect_id = str(commit_metadata.get("effect_id") or "").strip() - effect_base = ( - f"quota-monitor-poll:{goal_id}:{agent_id}:{prior_turn_instance_id}" - ) - if effect_id not in {effect_base, f"{effect_base}:todo:{todo_id}"}: + return { + "task_class": todo_item_task_class(dict(item)), + # The verdict reads the persisted status verbatim; trimming here would + # accept a value the legacy projection never accepted. + "status": str(item.get("status") or ""), + "has_resume_when": bool(item.get("resume_when")), + "has_successor_todo_ids": ( + isinstance(item.get("successor_todo_ids"), list) + and bool(item.get("successor_todo_ids")) + ), + "target_key": str(item.get("target_key") or "").strip() or None, + "cadence": str(item.get("cadence") or "").strip() or None, + } + + +def _prior_closeout_preflight( + *, + runtime_root: Path, + goal_id: str, + agent_id: str, + current_turn_instance_id: str | None, +) -> tuple[dict[str, Any], list[str]] | 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 + settlement itself, so this side ships a runtime path and an identity rather + than a megabyte log, and a settled prior Turn never causes a bound-fact read. + """ + + try: + result = effect_runtime_result( + PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_METHOD, + { + "schema_version": PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_REQUEST_SCHEMA, + "runtime_root": str(runtime_root.expanduser()), + "goal_id": goal_id, + "agent_id": agent_id, + "exclude_turn_instance_id": current_turn_instance_id, + }, + ) + except EffectRuntimeRejected as exc: + # Keep the public diagnostic the identity rule has always published, + # even though the rule now lives in the typed owner. + if exc.diagnostic_code == "heartbeat_receipt_identity_conflict": + raise HeartbeatReceiptIdentityConflictError(str(exc)) from None + raise + if not isinstance(result, Mapping) or ( + result.get("schema_version") + != PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_RESULT_SCHEMA + ): + raise RuntimeError("TypeScript closeout preflight result shape mismatch") + status = result.get("status") + if status == "none": return None - return receipt + if status != "candidate": + raise RuntimeError("TypeScript closeout preflight result shape mismatch") + candidate = result.get("candidate") + 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] def _unsettled_host_turn_recovery( @@ -113,86 +182,70 @@ def _unsettled_host_turn_recovery( ) -> dict[str, Any] | None: if not agent_id or not current_turn_instance_id: return None - for receipt in prior_closeout_required_heartbeat_receipts( - runtime_root, + preflight = _prior_closeout_preflight( + runtime_root=runtime_root, goal_id=goal_id, agent_id=agent_id, - exclude_turn_instance_id=current_turn_instance_id, - ): - todo_id = heartbeat_receipt_settlement_todo_id(receipt) - replan_obligation_id = heartbeat_receipt_settlement_replan_obligation_id( - receipt - ) - prior_turn_id = str(receipt.get("run_id") or "").strip() - if not prior_turn_id or not (todo_id or replan_obligation_id): - continue - readback = read_heartbeat_settlement( - runtime_root, - goal_id=goal_id, - agent_id=agent_id, - todo_id=todo_id, - turn_instance_id=prior_turn_id, - replan_obligation_id=replan_obligation_id, - ) - if readback is not None and readback.settlement.failure is None: - return None - todo_item = _bound_todo_item( - registry_path=registry_path, - runtime_root=runtime_root, - goal_id=goal_id, - todo_id=todo_id, - ) - if _committed_monitor_poll_closeout( + current_turn_instance_id=current_turn_instance_id, + ) + if preflight is None: + return None + selected, missing_receipts = 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 = ( + str(selected.get("binding_id") or "") + 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. + todo_item = _bound_todo_item( + registry_path=registry_path, + runtime_root=runtime_root, + goal_id=goal_id, + todo_id=todo_id, + ) + 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, - ) is not None: - return None - lifecycle_closeout = _typed_lifecycle_closeout(todo_item) - if lifecycle_closeout is not None: - return None - details_value = receipt.get("details") - details = details_value if isinstance(details_value, Mapping) else {} - effect_id = str(details.get("settlement_effect_id") or "").strip() - missing_receipts: list[str] = [] - if readback is None or readback.writeback.failure is not None: - missing_receipts.append("durable_writeback_receipt") - if readback is None or readback.spend.failure is not None: - missing_receipts.append("quota_spend_receipt") - binding_kind = "todo" if todo_id else "autonomous_replan" - binding_id = todo_id or replan_obligation_id - recovery = { - "schema_version": UNSETTLED_HOST_TURN_RECOVERY_SCHEMA_VERSION, - "state": "recovery_required", - "reason_code": "required_closeout_receipt_missing", - "prior_turn_instance_id": prior_turn_id, - "prior_event_id": receipt.get("event_id"), - "binding_kind": binding_kind, - "binding_id": binding_id, - "settlement_effect_id": effect_id, + ), + } + verdict = effect_runtime_result( + UNSETTLED_HOST_TURN_RECOVERY_METHOD, + { + "schema_version": UNSETTLED_HOST_TURN_RECOVERY_REQUEST_SCHEMA, + "goal_id": goal_id, + "agent_id": agent_id, + "candidate": selected, "missing_receipts": missing_receipts, - "accepted_closeouts": [ - "validated_writeback_and_quota_spend", - "exact_committed_quota_monitor_poll", - "typed_external_wait_with_runnable_successor", - "typed_blocker_or_lifecycle_transition", - ], - "external_state_policy": "typed_host_observation_only", - "quota_policy": "no_spend_for_recovery_transition", - } - if todo_item is not None: - recovery["binding_task_class"] = todo_item_task_class(todo_item) - recovery["binding_target_key"] = ( - str(todo_item.get("target_key") or "").strip() or None - ) - recovery["binding_cadence"] = ( - str(todo_item.get("cadence") or "").strip() or None - ) - return recovery - return None + "binding_facts": binding_facts, + }, + ) + if not isinstance(verdict, Mapping) or ( + verdict.get("schema_version") != UNSETTLED_HOST_TURN_RECOVERY_RESULT_SCHEMA + ): + raise RuntimeError("TypeScript recovery result shape mismatch") + status = verdict.get("status") + if status == "none": + return None + if status != "recovery_required": + raise RuntimeError("TypeScript recovery result shape mismatch") + recovery = verdict.get("recovery") + obligation = verdict.get("obligation") + if not isinstance(recovery, Mapping) or not isinstance(obligation, Mapping): + raise RuntimeError("TypeScript recovery result shape mismatch") + if recovery.get("schema_version") != UNSETTLED_HOST_TURN_RECOVERY_SCHEMA_VERSION: + raise RuntimeError("TypeScript recovery result shape mismatch") + return {"recovery": dict(recovery), "obligation": dict(obligation)} def apply_unsettled_host_turn_recovery_if_required( @@ -210,16 +263,17 @@ def apply_unsettled_host_turn_recovery_if_required( ) -> bool: """Preempt ordinary selection when the preceding host Turn lacks closeout.""" - recovery = _unsettled_host_turn_recovery( + verdict = _unsettled_host_turn_recovery( registry_path=registry_path, runtime_root=runtime_root, goal_id=goal_id, agent_id=agent_id, current_turn_instance_id=current_turn_instance_id, ) - if recovery is None: + if verdict is None: return False - binding_id = str(recovery.get("binding_id") or "prior binding") + recovery = verdict["recovery"] + obligation = verdict["obligation"] payload.pop("selected_todo", None) payload.pop("todo_id", None) payload.pop("action_portfolio", None) @@ -233,36 +287,33 @@ def apply_unsettled_host_turn_recovery_if_required( "normal_delivery_allowed": False, "recovery_delivery_allowed": False, "self_repair_allowed": False, - "reason": "a prior must-attempt heartbeat has no legal closeout receipt", - "recommended_action": ( - f"Recover prior unsettled host Turn for {binding_id}; use a typed " - "lifecycle observation, then rerun quota and continue independent work" - ), + "reason": obligation["reason"], + "recommended_action": obligation["recommended_action"], "unsettled_host_turn_recovery": recovery, "heartbeat_recommendation": { "source": "unsettled_host_turn_recovery", "recommended_mode": "unsettled_host_turn_recovery", - "notify": "DONT_NOTIFY", - "spend_policy": "no spend for the recovery transition", - "reason": "prior must-attempt host Turn is missing a legal closeout", + "notify": obligation["notify"], + "spend_policy": obligation["spend_policy"], + "reason": obligation["recommendation_reason"], "agent_must_attempt": True, }, "execution_obligation": { "must_attempt_work": True, "kind": "unsettled_host_turn_recovery", - "contract": "repair_prior_turn_closeout", - "contract_obligation": "author_typed_closeout_then_continue_successor", - "delivery_allowed": False, + "contract": obligation["contract"], + "contract_obligation": obligation["contract_obligation"], + "delivery_allowed": obligation["delivery_allowed"], "notify_is_execution_gate": False, - "reason": "prior must-attempt host Turn is missing a legal closeout", + "reason": obligation["recommendation_reason"], }, "work_lane_contract": { "schema_version": "work_lane_contract_v1", - "lane": "control_plane_recovery", - "next_lane": "advancement_task", - "obligation": "author_typed_closeout_then_continue_successor", - "must_attempt_work": True, - "reason_codes": ["unsettled_host_turn"], + "lane": obligation["lane"], + "next_lane": obligation["next_lane"], + "obligation": obligation["obligation"], + "must_attempt_work": obligation["must_attempt_work"], + "reason_codes": [obligation["reason_code"]], "monitor_policy": "typed_observation_only", "action": "repair the prior Turn closeout without spending quota", }, @@ -271,8 +322,8 @@ def apply_unsettled_host_turn_recovery_if_required( "keep_active": True, "pause_allowed": False, "automation_action": "execute_bounded_recovery", - "reason": "prior must-attempt host Turn remains unsettled", - "spend_policy": "no spend for the recovery transition", + "reason": obligation["unsettled_reason"], + "spend_policy": obligation["spend_policy"], }, } ) diff --git a/loopx/control_plane/quota/unsettled_host_turn_recovery.ts b/loopx/control_plane/quota/unsettled_host_turn_recovery.ts new file mode 100644 index 0000000000..bfbd463814 --- /dev/null +++ b/loopx/control_plane/quota/unsettled_host_turn_recovery.ts @@ -0,0 +1,505 @@ +/** + * Prior-host-Turn closeout recovery. + * + * One must-attempt heartbeat Turn can leave the goal without a legal closeout + * receipt. Whether that happened, which closeout is accepted, and what the + * recovery obligation is are domain decisions; they live here so the Python + * caller only reads its own persisted facts, transports them, and projects the + * typed result back into the existing public payload. + * + * The transaction is two requests around one real Python provider: a + * fail-closed preflight that reads the persisted receipts and the settlement + * readback and names the exact prior Turn whose bound facts the caller must + * read, then one final reduction over those checkpointed facts. A prior Turn + * whose settlement already validates is answered by the preflight, so the + * caller reads no bound facts for it. + */ +import type { JsonObject } from "../effect_program.ts"; +import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import { + jsonObject, + requireJsonObject, + requireNonEmptyString, +} from "../runtime_decode.ts"; +import { readGoalHeartbeatReceipts } from "../rollout_receipt_log.ts"; +import { + heartbeatReceiptBinding, + heartbeatReceiptFactFromEvent, + normalizeHeartbeatReplanObligationId, + normalizeHeartbeatTodoId, + optionalHeartbeatString, + selectEffectiveHeartbeatReceipt, + type HeartbeatReceiptBinding, + type HeartbeatReceiptFact, +} from "./heartbeat_receipt_identity.ts"; +import { + QUOTA_SETTLEMENT_READBACK_REQUEST_SCHEMA, + readQuotaSettlement, +} from "./settlement_readback.ts"; + +export const PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_REQUEST_SCHEMA = + "loopx_prior_host_turn_closeout_preflight_request_v0"; +export const PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_RESULT_SCHEMA = + "loopx_prior_host_turn_closeout_preflight_result_v0"; +export const UNSETTLED_HOST_TURN_RECOVERY_REQUEST_SCHEMA = + "loopx_prior_host_turn_recovery_request_v0"; +export const UNSETTLED_HOST_TURN_RECOVERY_RESULT_SCHEMA = + "loopx_prior_host_turn_recovery_result_v0"; +export const UNSETTLED_HOST_TURN_RECOVERY_SCHEMA_VERSION = + "unsettled_host_turn_recovery_v0"; + +const MONITOR_TASK_CLASS = "continuous_monitor"; +const SETTLED_LIFECYCLE_STATUSES = ["done", "blocked", "deferred"] as const; +const WRITEBACK_RECEIPT = "durable_writeback_receipt"; +const SPEND_RECEIPT = "quota_spend_receipt"; +const MISSING_RECEIPT_NAMES = [WRITEBACK_RECEIPT, SPEND_RECEIPT] as const; + +export const ACCEPTED_CLOSEOUTS = [ + "validated_writeback_and_quota_spend", + "exact_committed_quota_monitor_poll", + "typed_external_wait_with_runnable_successor", + "typed_blocker_or_lifecycle_transition", +] as const; +export type AcceptedCloseout = (typeof ACCEPTED_CLOSEOUTS)[number]; + +export interface PriorHostTurnCloseoutCandidate extends JsonObject { + prior_turn_instance_id: string; + event_id: string | null; + binding_kind: HeartbeatReceiptBinding["binding_kind"]; + /** The Todo id or autonomous replan obligation id, per `binding_kind`. */ + binding_id: string; + settlement_effect_id: string | null; +} + +interface PreflightRequest { + runtime_root: string; + goal_id: string; + agent_id: string; + exclude_turn_instance_id: string | null; +} + +function decodePreflightRequest(value: unknown): PreflightRequest { + const request = requireJsonObject(value, "prior host Turn closeout preflight request"); + if (request.schema_version !== PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_REQUEST_SCHEMA) { + throw new EffectRuntimeRequestError( + "Prior host Turn closeout preflight request schema mismatch", + ); + } + return { + runtime_root: requireNonEmptyString(request.runtime_root, "runtime_root"), + goal_id: requireNonEmptyString(request.goal_id, "goal_id"), + agent_id: requireNonEmptyString(request.agent_id, "agent_id"), + exclude_turn_instance_id: optionalHeartbeatString( + request.exclude_turn_instance_id, + ), + }; +} + +function candidate( + binding: HeartbeatReceiptBinding, + turnInstanceId: string, + fact: HeartbeatReceiptFact, +): PriorHostTurnCloseoutCandidate { + return { + prior_turn_instance_id: turnInstanceId, + event_id: fact.event_id, + binding_kind: binding.binding_kind, + binding_id: binding.binding_id, + settlement_effect_id: binding.settlement_effect_id, + }; +} + +/** + * Select the prior must-attempt Turns that still require a host closeout. + * + * Every prior Turn is identity-validated, so a conflicting receipt fails the + * read instead of being hidden behind a newer Turn. Results are newest first + * and carry at most one effective receipt per Turn, which is the same single + * binding the caller's recovery decision may act on. + */ +function selectCloseoutCandidates( + receipts: readonly JsonObject[], + goalId: string, + agentId: string, + excludeTurnInstanceId: string | null, +): { candidates: PriorHostTurnCloseoutCandidate[]; turnsValidated: number } { + const perTurn = new Map(); + const newestFirst: string[] = []; + for (const event of receipts) { + const fact = heartbeatReceiptFactFromEvent(event); + const turnInstanceId = fact.run_id; + if (!turnInstanceId || turnInstanceId === excludeTurnInstanceId) { + continue; + } + const existing = perTurn.get(turnInstanceId); + if (existing) existing.push(fact); + else perTurn.set(turnInstanceId, [fact]); + const position = newestFirst.indexOf(turnInstanceId); + if (position !== -1) newestFirst.splice(position, 1); + newestFirst.push(turnInstanceId); + } + + const candidates: PriorHostTurnCloseoutCandidate[] = []; + let validated = 0; + for (const turnInstanceId of [...newestFirst].reverse()) { + const entries = (perTurn.get(turnInstanceId) ?? []).map((fact) => ({ + fact, + value: fact, + })); + const effective = selectEffectiveHeartbeatReceipt(goalId, agentId, entries); + validated += 1; + if (effective === null) continue; + // Closeout is opt-in on the persisted receipt, so older receipts from + // before this contract cannot become new recovery obligations. + if (!effective.closeout_required) continue; + const binding = heartbeatReceiptBinding(goalId, agentId, effective); + if (binding === null) continue; + candidates.push(candidate(binding, turnInstanceId, effective)); + } + return {candidates, turnsValidated: validated}; +} + +function settlementReadbackRequest( + request: PreflightRequest, + candidate: PriorHostTurnCloseoutCandidate, +): JsonObject { + return { + schema_version: QUOTA_SETTLEMENT_READBACK_REQUEST_SCHEMA, + runtime_root: request.runtime_root, + goal_id: request.goal_id, + agent_id: request.agent_id, + todo_id: candidate.binding_kind === "todo" ? candidate.binding_id : null, + turn_instance_id: candidate.prior_turn_instance_id, + replan_obligation_id: candidate.binding_kind === "autonomous_replan" + ? candidate.binding_id + : null, + infer_turn_instance_id: false, + allow_unbound_binding: false, + }; +} + +function bundleFailed(readback: JsonObject, step: string): boolean { + const bundle = jsonObject(readback[step]); + const payload = bundle === null ? null : jsonObject(bundle.payload); + if (payload === null || typeof payload.ok !== "boolean") { + throw new EffectRuntimeRequestError( + `quota settlement readback has no ${step} result`, + "malformed_settlement_readback", + ); + } + return payload.ok !== true; +} + +/** + * Fail-closed preflight for the prior-host-Turn closeout transaction. + * + * It reads the persisted guards itself, so the caller ships a runtime path and + * an identity instead of a whole rollout log, and it validates the selected + * Turn's settlement so a Turn that already settled never makes the caller read + * bound facts. + */ +export async function preflightPriorHostTurnCloseout( + value: unknown, +): Promise { + const request = decodePreflightRequest(value); + const receipts = await readGoalHeartbeatReceipts( + request.runtime_root, + request.goal_id, + request.agent_id, + ); + const { candidates, turnsValidated } = selectCloseoutCandidates( + receipts ?? [], + request.goal_id, + request.agent_id, + request.exclude_turn_instance_id, + ); + if (candidates.length === 0) { + return { + schema_version: PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_RESULT_SCHEMA, + status: "none", + reason: "no_prior_turn_requires_closeout", + turns_validated: turnsValidated, + }; + } + const selected = candidates[0]!; + const readback = await readQuotaSettlement( + settlementReadbackRequest(request, selected), + ); + if (readback.found === true && !bundleFailed(readback, "settlement")) { + // The Turn has a validated settlement closeout, so it needs no recovery. + return { + schema_version: PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_RESULT_SCHEMA, + status: "none", + reason: "prior_turn_settlement_validated", + turns_validated: turnsValidated, + prior_turn_instance_id: selected.prior_turn_instance_id, + accepted_closeout: "validated_writeback_and_quota_spend", + }; + } + const missingReceipts: string[] = []; + if (readback.found !== true || bundleFailed(readback, "writeback")) { + missingReceipts.push(WRITEBACK_RECEIPT); + } + if (readback.found !== true || bundleFailed(readback, "spend")) { + missingReceipts.push(SPEND_RECEIPT); + } + return { + schema_version: PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_RESULT_SCHEMA, + status: "candidate", + turns_validated: turnsValidated, + candidate: selected, + missing_receipts: missingReceipts, + }; +} + +function decodeCandidate(value: unknown): PriorHostTurnCloseoutCandidate { + const raw = requireJsonObject(value, "prior host Turn recovery candidate"); + const bindingKind = raw.binding_kind; + if (bindingKind !== "todo" && bindingKind !== "autonomous_replan") { + throw new EffectRuntimeRequestError( + "prior host Turn recovery candidate binding_kind is unsupported", + ); + } + const bindingId = requireNonEmptyString(raw.binding_id, "candidate.binding_id"); + if (bindingKind === "todo" && normalizeHeartbeatTodoId(bindingId) !== bindingId) { + throw new EffectRuntimeRequestError( + "prior host Turn recovery candidate Todo binding is not a legal Todo id", + ); + } + if ( + bindingKind === "autonomous_replan" && + normalizeHeartbeatReplanObligationId(bindingId) !== bindingId + ) { + throw new EffectRuntimeRequestError( + "prior host Turn recovery candidate autonomous replan binding is malformed", + ); + } + return { + prior_turn_instance_id: requireNonEmptyString( + raw.prior_turn_instance_id, + "candidate.prior_turn_instance_id", + ), + event_id: optionalHeartbeatString(raw.event_id), + binding_kind: bindingKind, + binding_id: bindingId, + settlement_effect_id: optionalHeartbeatString(raw.settlement_effect_id), + }; +} + +interface CandidateTodoFacts { + task_class: string; + status: string; + has_resume_when: boolean; + has_successor_todo_ids: boolean; + target_key: string | null; + cadence: string | null; +} + +function decodeTodoFacts(value: unknown): CandidateTodoFacts | null { + if (value === null || value === undefined) return null; + const raw = requireJsonObject(value, "prior host Turn recovery bound Todo"); + if ( + typeof raw.has_resume_when !== "boolean" || + typeof raw.has_successor_todo_ids !== "boolean" + ) { + throw new EffectRuntimeRequestError( + "prior host Turn recovery Todo must declare its retained wait shape", + ); + } + return { + task_class: typeof raw.task_class === "string" ? raw.task_class : "", + // The lifecycle verdict reads the persisted status verbatim; trimming here + // would accept a value the legacy projection never accepted. + status: typeof raw.status === "string" ? raw.status : "", + has_resume_when: raw.has_resume_when, + has_successor_todo_ids: raw.has_successor_todo_ids, + target_key: optionalHeartbeatString(raw.target_key), + cadence: optionalHeartbeatString(raw.cadence), + }; +} + +interface BindingFacts { + read: boolean; + todo: CandidateTodoFacts | null; + committedMonitorPoll: JsonObject | null; +} + +function decodeBindingFacts(value: unknown): BindingFacts { + const raw = requireJsonObject(value, "prior host Turn recovery binding facts"); + if (raw.status === "not_read") { + return { read: false, todo: null, committedMonitorPoll: null }; + } + if (raw.status !== "read") { + throw new EffectRuntimeRequestError( + "prior host Turn recovery binding facts status is unsupported", + ); + } + return { + read: true, + todo: decodeTodoFacts(raw.todo ?? null), + committedMonitorPoll: jsonObject(raw.committed_monitor_poll) ?? null, + }; +} + +/** + * Decode the checkpointed missing-receipt list the preflight produced. + * + * The names are the closeout policy's own vocabulary, so an unknown or + * reordered list is a caller error rather than something to normalize silently. + */ +function decodeMissingReceipts(value: unknown): string[] { + if (!Array.isArray(value)) { + throw new EffectRuntimeRequestError("missing_receipts must be an array"); + } + const expected = MISSING_RECEIPT_NAMES.filter((name) => value.includes(name)); + if ( + expected.length === 0 || + expected.length !== value.length || + expected.some((name, index) => name !== value[index]) + ) { + throw new EffectRuntimeRequestError( + "missing_receipts must be the closeout receipts this transaction names, in order", + ); + } + return [...expected]; +} + +function committedMonitorPollEffectIds( + candidateBinding: PriorHostTurnCloseoutCandidate, + goalId: string, + agentId: string, +): string[] | null { + if (candidateBinding.binding_kind !== "todo") return null; + const base = `quota-monitor-poll:${goalId}:${agentId}:${candidateBinding.prior_turn_instance_id}`; + return [base, `${base}:todo:${candidateBinding.binding_id}`]; +} + +function acceptedLifecycleCloseout(todo: CandidateTodoFacts | null): AcceptedCloseout | null { + if (todo === null) return null; + if ( + todo.status === "open" && + todo.has_resume_when && + todo.has_successor_todo_ids + ) { + return "typed_external_wait_with_runnable_successor"; + } + const settled = (SETTLED_LIFECYCLE_STATUSES as readonly string[]).includes( + todo.status, + ); + return settled ? "typed_blocker_or_lifecycle_transition" : null; +} + +function recoveryObligation( + repair: "monitor_poll" | "lifecycle", + bindingId: string, +): JsonObject { + return { + lane: "control_plane_recovery", + next_lane: "advancement_task", + obligation: "author_typed_closeout_then_continue_successor", + contract: "repair_prior_turn_closeout", + contract_obligation: "author_typed_closeout_then_continue_successor", + must_attempt_work: true, + delivery_allowed: false, + notify: "DONT_NOTIFY", + spend_policy: "no spend for the recovery transition", + reason_code: "unsettled_host_turn", + reason: "a prior must-attempt heartbeat has no legal closeout receipt", + recommendation_reason: "prior must-attempt host Turn is missing a legal closeout", + unsettled_reason: "prior must-attempt host Turn remains unsettled", + recommended_action: + `Recover prior unsettled host Turn for ${bindingId}; use a typed ` + + "lifecycle observation, then rerun quota and continue independent work", + repair, + }; +} + +/** + * Reduce the selected prior Turn to its closeout verdict. + * + * The accepted closeout order is policy: a validated settlement wins, then an + * exactly committed monitor-poll closeout, then a typed lifecycle transition. + * Anything else is a recovery obligation with the exact missing receipts named. + */ +export function reduceUnsettledHostTurnRecovery(value: unknown): JsonObject { + const request = requireJsonObject(value, "prior host Turn recovery request"); + if (request.schema_version !== UNSETTLED_HOST_TURN_RECOVERY_REQUEST_SCHEMA) { + throw new EffectRuntimeRequestError( + "Prior host Turn recovery request schema mismatch", + ); + } + const goalId = requireNonEmptyString(request.goal_id, "goal_id"); + const agentId = requireNonEmptyString(request.agent_id, "agent_id"); + const selected = decodeCandidate(request.candidate); + const missingReceipts = decodeMissingReceipts(request.missing_receipts); + + const bindingFacts = decodeBindingFacts(request.binding_facts); + if (!bindingFacts.read) { + // A caller that skipped the bound-Todo read cannot be given a verdict: the + // monitor-poll and lifecycle closeouts are exactly what that read proves. + throw new EffectRuntimeRequestError( + "prior host Turn recovery requires its bound Todo facts once the preflight named it as a closeout candidate", + ); + } + const todo = bindingFacts.todo; + const acceptedEffectIds = committedMonitorPollEffectIds(selected, goalId, agentId); + const committedEffectId = bindingFacts.committedMonitorPoll === null + ? null + : optionalHeartbeatString(bindingFacts.committedMonitorPoll.effect_id); + if ( + acceptedEffectIds !== null && + todo?.task_class === MONITOR_TASK_CLASS && + committedEffectId !== null && + acceptedEffectIds.includes(committedEffectId) + ) { + return { + schema_version: UNSETTLED_HOST_TURN_RECOVERY_RESULT_SCHEMA, + status: "none", + prior_turn_instance_id: selected.prior_turn_instance_id, + accepted_closeout: "exact_committed_quota_monitor_poll", + }; + } + + const lifecycleCloseout = acceptedLifecycleCloseout(todo); + if (lifecycleCloseout !== null) { + return { + schema_version: UNSETTLED_HOST_TURN_RECOVERY_RESULT_SCHEMA, + status: "none", + prior_turn_instance_id: selected.prior_turn_instance_id, + accepted_closeout: lifecycleCloseout, + }; + } + + const recovery: JsonObject = { + schema_version: UNSETTLED_HOST_TURN_RECOVERY_SCHEMA_VERSION, + state: "recovery_required", + reason_code: "required_closeout_receipt_missing", + prior_turn_instance_id: selected.prior_turn_instance_id, + prior_event_id: selected.event_id, + binding_kind: selected.binding_kind, + binding_id: selected.binding_id, + settlement_effect_id: selected.settlement_effect_id, + missing_receipts: missingReceipts, + accepted_closeouts: [...ACCEPTED_CLOSEOUTS], + external_state_policy: "typed_host_observation_only", + quota_policy: "no_spend_for_recovery_transition", + }; + if (todo !== null) { + recovery.binding_task_class = todo.task_class; + recovery.binding_target_key = todo.target_key; + recovery.binding_cadence = todo.cadence; + } + const repair = todo?.task_class === MONITOR_TASK_CLASS + ? "monitor_poll" as const + : "lifecycle" as const; + // The typed repair lane lets the CLI renderer project its command list + // without re-deriving which closeout family the Turn belongs to. + recovery.repair = repair; + return { + schema_version: UNSETTLED_HOST_TURN_RECOVERY_RESULT_SCHEMA, + status: "recovery_required", + prior_turn_instance_id: selected.prior_turn_instance_id, + recovery, + obligation: recoveryObligation(repair, selected.binding_id), + }; +} diff --git a/loopx/control_plane/rollout_receipt_log.ts b/loopx/control_plane/rollout_receipt_log.ts new file mode 100644 index 0000000000..c2bb76e7af --- /dev/null +++ b/loopx/control_plane/rollout_receipt_log.ts @@ -0,0 +1,99 @@ +/** + * One owner for the goal rollout-event log location and its receipt reads. + * + * The log is a goal-level runtime artifact that more than one transaction + * reads: the settlement readback, the receipt-bound scheduler follow-up, and + * prior-host-Turn closeout resolution. Resolving the path and reading its + * 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"; + +export const ROLLOUT_EVENT_SCHEMA_VERSION = "loopx_rollout_event_v0"; +export const HEARTBEAT_RECEIPT_EVENT_KIND = "quota_should_run"; + +/** Reject a goal id that is not one path segment, before any log read. */ +export function goalPathSegment(value: unknown): string { + const label = "goal_id"; + const result = requireNonEmptyString(value, label).trim(); + if ( + result === "." || + result === ".." || + result.includes("/") || + result.includes("\\") + ) { + throw new EffectRuntimeRequestError( + `${label} must be a single path segment`, + "invalid_goal_id", + ); + } + return result; +} + +/** Resolve one goal's rollout-event log inside `runtime_root`. */ +export function goalRolloutEventLogPath( + runtimeRoot: string, + goalId: string, +): string { + const root = resolve(runtimeRoot); + const path = resolve( + root, + "goals", + goalPathSegment(goalId), + "rollout-event-log.jsonl", + ); + const child = relative(root, path); + if (child === "" || child === ".." || child.startsWith(`..${sep}`)) { + throw new EffectRuntimeRequestError( + "rollout event log path escapes runtime_root", + "invalid_rollout_event_log_path", + ); + } + return path; +} + +/** + * Read the heartbeat receipts this goal persisted for one Agent. + * + * The read mirrors the established non-strict reader: a malformed or + * differently versioned line is skipped rather than allowed to erase or + * manufacture a receipt, and an absent log is `null` because "no receipts yet" + * and "no log yet" are different facts for a caller that must report state. + */ +export async function readGoalHeartbeatReceipts( + runtimeRoot: string, + goalId: string, + agentId?: string | null, +): Promise { + 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 receipts: JsonObject[] = []; + for (const line of text.split(/\r?\n/)) { + if (!line.trim()) continue; + try { + const event = jsonObject(JSON.parse(line)); + if ( + event?.schema_version === ROLLOUT_EVENT_SCHEMA_VERSION && + event.event_kind === HEARTBEAT_RECEIPT_EVENT_KIND && + event.goal_id === goalId && + (agentId === undefined || event.agent_id === agentId) + ) { + receipts.push(event); + } + } catch { + // Match the established non-strict rollout-event reader: unrelated + // malformed lines do not manufacture or erase a valid receipt. + } + } + return receipts; +} diff --git a/loopx/control_plane/scheduler/heartbeat_followup.ts b/loopx/control_plane/scheduler/heartbeat_followup.ts index 179ab327d7..be177dcca8 100644 --- a/loopx/control_plane/scheduler/heartbeat_followup.ts +++ b/loopx/control_plane/scheduler/heartbeat_followup.ts @@ -1,8 +1,9 @@ -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 { + goalPathSegment, + readGoalHeartbeatReceipts, +} from "../rollout_receipt_log.ts"; import { jsonObject, requireBoolean, @@ -24,7 +25,6 @@ export const SCHEDULER_HEARTBEAT_FOLLOWUP_RESULT_SCHEMA = export const SCHEDULER_HEARTBEAT_FOLLOWUP_ERROR_SCHEMA = "loopx_scheduler_host_followup_error_v0"; -const ROLLOUT_EVENT_SCHEMA = "loopx_rollout_event_v0"; const ACK_CLASSIFICATION = "quota_scheduler_ack"; const FAILURE_CLASSIFICATION = "quota_scheduler_host_update_failure"; @@ -52,17 +52,6 @@ function optionalText(value: unknown): string | null { return requireNonEmptyString(value, "scheduler follow-up optional text").trim(); } -function pathSegment(value: unknown, label: string): string { - const result = requireNonEmptyString(value, label).trim(); - if (result === "." || result === ".." || result.includes("/") || result.includes("\\")) { - throw new EffectRuntimeRequestError( - `${label} must be a single path segment`, - `invalid_${label}`, - ); - } - return result; -} - function followupOperation(facts: JsonObject): FollowupOperation { const value = facts.operation ?? facts.outcome; if (value === "ack") return "ack"; @@ -89,7 +78,7 @@ function requestObject(value: unknown): SchedulerHeartbeatFollowupRequest { "scheduler_host_facts_schema_mismatch", ); } - pathSegment(hostFacts.goal_id, "goal_id"); + goalPathSegment(hostFacts.goal_id); requireNonEmptyString(hostFacts.agent_id, "agent_id"); followupOperation(hostFacts); const turnInstanceId = optionalText(input.turn_instance_id); @@ -138,18 +127,6 @@ function compactBefore(value: JsonObject): JsonObject { }; } -function receiptLogPath(runtimeRoot: string, goalId: string): string { - const root = resolve(runtimeRoot); - const path = resolve(root, "goals", pathSegment(goalId, "goal_id"), "rollout-event-log.jsonl"); - const child = relative(root, path); - if (child === "" || child === ".." || child.startsWith(`..${sep}`)) { - throw new EffectRuntimeRequestError( - "scheduler follow-up receipt path escapes runtime_root", - "invalid_scheduler_receipt_path", - ); - } - return path; -} async function heartbeatReceiptStatus( runtimeRoot: string, @@ -157,29 +134,8 @@ async function heartbeatReceiptStatus( agentId: string, turnInstanceId: string, ): Promise { - let text: string; - try { - text = await readFile(receiptLogPath(runtimeRoot, goalId), "utf8"); - } catch (error) { - if ((error as NodeJS.ErrnoException).code === "ENOENT") return "missing"; - throw error; - } - const receipts: JsonObject[] = []; - for (const line of text.split(/\r?\n/)) { - if (!line.trim()) continue; - try { - const event = jsonObject(JSON.parse(line)); - if ( - event?.schema_version === ROLLOUT_EVENT_SCHEMA && - event.event_kind === "quota_should_run" && - event.goal_id === goalId && - event.agent_id === agentId - ) receipts.push(event); - } catch { - // Match the established non-strict rollout-event reader: unrelated - // malformed lines do not manufacture or erase a valid receipt. - } - } + const receipts = await readGoalHeartbeatReceipts(runtimeRoot, goalId, agentId); + if (receipts === null) return "missing"; const firstMatch = receipts.findIndex((event) => event.run_id === turnInstanceId); if (firstMatch < 0) return "missing"; return receipts.slice(firstMatch + 1).some((event) => event.run_id !== turnInstanceId) diff --git a/loopx/control_plane/work_items/unsettled_host_turn_contract.py b/loopx/control_plane/work_items/unsettled_host_turn_contract.py index f551cbe598..4c2cd3e530 100644 --- a/loopx/control_plane/work_items/unsettled_host_turn_contract.py +++ b/loopx/control_plane/work_items/unsettled_host_turn_contract.py @@ -24,7 +24,9 @@ def recovery_cli_actions( if turn_instance_id else " --turn-instance-id " ) - if recovery.get("binding_task_class") == "continuous_monitor": + # The repair lane is a typed fact from the recovery transaction; this + # renderer only turns it into operator commands. + if recovery.get("repair") == "monitor_poll": prior_turn_id = str( recovery.get("prior_turn_instance_id") or "" ) diff --git a/tests/control_plane/test_effect_turn_live_quota_decision.py b/tests/control_plane/test_effect_turn_live_quota_decision.py index b4eaf6569a..d83c7a9853 100644 --- a/tests/control_plane/test_effect_turn_live_quota_decision.py +++ b/tests/control_plane/test_effect_turn_live_quota_decision.py @@ -3,6 +3,8 @@ import json from pathlib import Path +import pytest + from loopx.control_plane.effect_program import ( interpret_quota_should_run_packet, ) @@ -15,6 +17,9 @@ bind_action_selection_cli_routes, build_live_quota_should_run_decision, ) +from loopx.control_plane.quota.error_codes import ( + HeartbeatReceiptIdentityConflictError, +) from loopx.control_plane.testing.quota_fixtures import quota_status_payload from loopx.rollout_event_log import build_rollout_event @@ -818,3 +823,104 @@ def test_duplicate_required_inbox_routes_project_one_public_safe_read( assert len(packet["required_reads"]) == 1 assert packet["required_reads"][0]["command"] == command + + +def test_prior_closeout_identity_conflict_fails_closed( + tmp_path: Path, +) -> None: + """A conflicting prior receipt cannot be resolved into a recovery verdict.""" + + runtime_root = tmp_path / "runtime" + registry_path = tmp_path / "registry.json" + state_path = tmp_path / "ACTIVE_GOAL_STATE.md" + agent_id = "codex-fixture" + prior_turn_id = "managed-prior-turn" + state_path.write_text( + "# Goal\n\n## Agent Todo\n\n" + "- [ ] [P1] Keep advancing the selected task.\n" + " \n", + encoding="utf-8", + ) + registry_path.write_text( + json.dumps( + { + "common_runtime_root": str(runtime_root), + "goals": [ + { + "id": GOAL_ID, + "repo": str(tmp_path), + "state_file": str(state_path), + } + ], + } + ), + encoding="utf-8", + ) + goal_runtime = runtime_root / "goals" / GOAL_ID + goal_runtime.mkdir(parents=True) + conflicting = [ + {"todo_id": "todo_ordinary_work"}, + {"todo_id": "todo_other_work"}, + ] + (goal_runtime / "rollout-event-log.jsonl").write_text( + "".join( + json.dumps( + build_rollout_event( + goal_id=GOAL_ID, + event_kind="quota_should_run", + agent_id=agent_id, + todo_id=str(details["todo_id"]), + run_id=prior_turn_id, + status="normal_run", + summary="managed heartbeat guard requires closeout", + details={**details, "closeout_required": True}, + ) + ) + + "\n" + for details in conflicting + ), + encoding="utf-8", + ) + + todo_text = "[P1] Keep advancing the selected task." + status = quota_status_payload( + goal_id=GOAL_ID, + status="active", + agent_todo_items=[ + { + "todo_id": "todo_ordinary_work", + "index": 1, + "text": todo_text, + "role": "agent", + "status": "open", + "priority": "P1", + "task_class": "advancement_task", + } + ], + recommended_action=todo_text, + next_action=todo_text, + coordination={ + "registered_agents": [agent_id], + "agent_model": "peer_v1", + }, + claim_scope_agent_id=agent_id, + ) + with pytest.raises(HeartbeatReceiptIdentityConflictError): + build_live_quota_should_run_decision( + status, + goal_id=GOAL_ID, + agent_id=agent_id, + available_capabilities=["shell"], + include_scheduler_detail=False, + codex_app_current_rrule=None, + registry_path=registry_path, + runtime_root=runtime_root, + route_source="loopx_turn_plan", + turn_instance_id="managed-current-turn", + scheduler_execution_context={ + "host_surface": "generic_cli", + "scheduler_owner": "agent_cli_loop", + "execution_mode": "interactive", + }, + ) diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index 2994ef7973..9ec52e4c6c 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -1346,6 +1346,89 @@ def test_recovery_guard_accepts_the_todo_it_must_settle( assert recovery.get("error_code") != "quota_unexpected_collection_error" +def test_prior_turn_with_several_receipts_recovers_once(tmp_path: Path) -> None: + project, runtime, registry_path = _write_fixture(tmp_path) + prior_turn_id = "turn-multi-receipt-prior" + prior_rc, prior = _run_cli( + registry_path, + runtime, + "quota", + "should-run", + "--codex-app", + "--goal-id", + GOAL_ID, + "--agent-id", + AGENT_ID, + "--turn-instance-id", + prior_turn_id, + "--todo-id", + TODO_ID, + "--scan-path", + str(project), + ) + assert prior_rc == 0, prior + assert prior["heartbeat_receipt"]["closeout_required"] is True + + # A real goal persists more than one guard per Turn: an identity upgrade + # append and an older guard that never carried a binding. The newest bound + # receipt decides the Turn, and the unbound one can neither outrank it nor + # turn the Turn into an identity conflict. + log_path = runtime / "goals" / GOAL_ID / "rollout-event-log.jsonl" + with log_path.open("a", encoding="utf-8") as stream: + for event_id, details in ( + ( + "multi-receipt-identity-upgrade", + { + "todo_id": TODO_ID, + "settlement_effect_id": ( + f"{GOAL_ID}:{AGENT_ID}:{TODO_ID}:{prior_turn_id}" + ), + "closeout_required": True, + }, + ), + ("multi-receipt-stale-unbound", {"stall_observation": "not_applicable"}), + ): + stream.write( + json.dumps( + { + "schema_version": "loopx_rollout_event_v0", + "event_id": event_id, + "event_kind": "quota_should_run", + "goal_id": GOAL_ID, + "agent_id": AGENT_ID, + "run_id": prior_turn_id, + "status": "turn_run_once", + "summary": "prior Turn guard", + "details": details, + }, + sort_keys=True, + ) + + "\n" + ) + + recovery_rc, recovery = _run_cli( + registry_path, + runtime, + "quota", + "should-run", + "--codex-app", + "--goal-id", + GOAL_ID, + "--agent-id", + AGENT_ID, + "--begin-turn", + "--scan-path", + str(project), + ) + assert recovery_rc == 0, recovery + assert recovery["effective_action"] == "unsettled_host_turn_recovery" + packet = recovery["unsettled_host_turn_recovery"] + assert packet["prior_turn_instance_id"] == prior_turn_id + assert packet["binding_kind"] == "todo" + assert packet["binding_id"] == TODO_ID + assert packet["prior_event_id"] == "multi-receipt-identity-upgrade" + + @pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"]) @pytest.mark.parametrize("hidden_count", [0, 6]) def test_prior_host_closeout_survives_hidden_todo_lifecycle( diff --git a/tests/control_plane_ts/unsettled_host_turn_recovery.test.ts b/tests/control_plane_ts/unsettled_host_turn_recovery.test.ts new file mode 100644 index 0000000000..c75644ac9f --- /dev/null +++ b/tests/control_plane_ts/unsettled_host_turn_recovery.test.ts @@ -0,0 +1,569 @@ +import assert from "node:assert/strict"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; + +import type { JsonObject } from "../../loopx/control_plane/effect_program.ts"; +import { + ACCEPTED_CLOSEOUTS, + PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_REQUEST_SCHEMA, + UNSETTLED_HOST_TURN_RECOVERY_REQUEST_SCHEMA, + preflightPriorHostTurnCloseout, + reduceUnsettledHostTurnRecovery, +} from "../../loopx/control_plane/quota/unsettled_host_turn_recovery.ts"; + +const GOAL = "goal-a"; +const AGENT = "agent-a"; + +interface Runtime { + root: string; + close(): Promise; +} + +/** A disposable runtime whose only persisted state is the receipts under test. */ +async function runtimeWith(events: JsonObject[]): Promise { + const root = await mkdtemp(join(tmpdir(), "loopx-closeout-")); + const goalRoot = join(root, "goals", GOAL); + await mkdir(goalRoot, {recursive: true}); + await writeFile( + join(goalRoot, "rollout-event-log.jsonl"), + events.map((event) => JSON.stringify(event)).join("\n") + "\n", + "utf8", + ); + return {root, close: () => rm(root, {recursive: true, force: true})}; +} + +function receipt( + turn: string, + details: JsonObject, + overrides: JsonObject = {}, +): JsonObject { + return { + schema_version: "loopx_rollout_event_v0", + event_id: `event-${turn}`, + event_kind: "quota_should_run", + goal_id: GOAL, + agent_id: AGENT, + run_id: turn, + details, + ...overrides, + }; +} + +function closeoutRequired( + turn: string, + todoId: string | null, + replanId: string | null = null, +): JsonObject { + return { + ...(todoId ? {todo_id: todoId} : {}), + ...(replanId ? {replan_obligation_id: replanId} : {}), + ...(todoId ? {settlement_effect_id: `${GOAL}:${AGENT}:${todoId}:${turn}`} : {}), + closeout_required: true, + }; +} + +function preflight(runtimeRoot: string, exclude: string | null = "current-turn") { + return preflightPriorHostTurnCloseout({ + schema_version: PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_REQUEST_SCHEMA, + runtime_root: runtimeRoot, + goal_id: GOAL, + agent_id: AGENT, + exclude_turn_instance_id: exclude, + }); +} + +function candidateFrom(result: JsonObject): JsonObject { + assert.equal(result.status, "candidate"); + return result.candidate as JsonObject; +} + +function reduce( + candidate: JsonObject, + missingReceipts: string[] = [ + "durable_writeback_receipt", + "quota_spend_receipt", + ], + bindingFacts: JsonObject = {status: "not_read"}, +) { + return reduceUnsettledHostTurnRecovery({ + schema_version: UNSETTLED_HOST_TURN_RECOVERY_REQUEST_SCHEMA, + goal_id: GOAL, + agent_id: AGENT, + candidate, + missing_receipts: missingReceipts, + binding_facts: bindingFacts, + }); +} + +const READ_TODO = (todo: JsonObject | null, committed: JsonObject | null = null) => ({ + status: "read", + todo, + committed_monitor_poll: committed, +}); + +const ADVANCEMENT_OPEN = { + task_class: "advancement_task", + status: "open", + has_resume_when: false, + has_successor_todo_ids: false, + target_key: null, + cadence: null, +}; + +test("the preflight reads the persisted receipts and names the newest required closeout", async () => { + const runtime = await runtimeWith([ + receipt("turn-old", closeoutRequired("turn-old", "todo_older")), + receipt("turn-new", closeoutRequired("turn-new", "todo_newer")), + // A Turn that repeats later is the more recent one, exactly as the + // persisted log's append order defines "newest". + receipt("turn-old", closeoutRequired("turn-old", "todo_older")), + ]); + try { + const result = await preflight(runtime.root); + assert.equal(result.schema_version, "loopx_prior_host_turn_closeout_preflight_result_v0"); + assert.equal(result.turns_validated, 2); + const candidate = candidateFrom(result); + assert.equal(candidate.prior_turn_instance_id, "turn-old"); + assert.equal(candidate.binding_kind, "todo"); + assert.equal(candidate.binding_id, "todo_older"); + assert.equal(candidate.event_id, "event-turn-old"); + // The receipt exists but its writeback and spend receipts do not. + assert.deepEqual(result.missing_receipts, [ + "durable_writeback_receipt", + "quota_spend_receipt", + ]); + } finally { + await runtime.close(); + } +}); + +test("a receipt that does not opt in never becomes a recovery obligation", async () => { + const runtime = await runtimeWith([ + receipt("turn-a", {todo_id: "todo_alpha", settlement_effect_id: "e"}), + receipt("turn-b", { + todo_id: "todo_beta", + settlement_effect_id: `${GOAL}:${AGENT}:todo_beta:turn-b`, + closeout_required: false, + }), + ]); + try { + const result = await preflight(runtime.root); + assert.equal(result.status, "none"); + assert.equal(result.reason, "no_prior_turn_requires_closeout"); + assert.equal(result.turns_validated, 2); + } finally { + await runtime.close(); + } +}); + +test("the current Turn, other goals, other agents, and other event kinds are ignored", async () => { + const runtime = await runtimeWith([ + receipt("current-turn", closeoutRequired("current-turn", "todo_current")), + receipt("turn-other-goal", closeoutRequired("turn-other-goal", "todo_other"), { + goal_id: "goal-b", + }), + receipt("turn-other-agent", closeoutRequired("turn-other-agent", "todo_other"), { + agent_id: "agent-b", + }), + receipt("turn-other-kind", closeoutRequired("turn-other-kind", "todo_other"), { + event_kind: "quota_settled", + }), + receipt("", closeoutRequired("", "todo_unbound_turn")), + ]); + try { + const result = await preflight(runtime.root); + assert.equal(result.status, "none"); + assert.equal(result.turns_validated, 0); + } finally { + await runtime.close(); + } +}); + +test("one receipt may bind only one settlement identity", async () => { + const runtime = await runtimeWith([ + receipt("turn-conflict", { + todo_id: "todo_alpha", + replan_obligation_id: `replan-${"a".repeat(16)}`, + closeout_required: true, + }), + ]); + try { + await assert.rejects( + () => preflight(runtime.root), + /conflicting Todo and autonomous replan bindings/, + ); + } finally { + await runtime.close(); + } +}); + +test("a receipt cannot claim an effect identity it does not bind", async () => { + const runtime = await runtimeWith([ + receipt("turn-a", { + settlement_effect_id: "effect-without-binding", + closeout_required: true, + }), + ]); + try { + await assert.rejects( + () => preflight(runtime.root), + /refuse to infer or upgrade it/, + ); + } finally { + await runtime.close(); + } +}); + +test("a Todo binding the settlement authority cannot address fails the read", async () => { + const runtime = await runtimeWith([ + receipt("turn-a", { + todo_id: "TODO_ALPHA", + closeout_required: true, + }), + ]); + try { + await assert.rejects( + () => preflight(runtime.root), + /not a legal Todo id/, + ); + } finally { + await runtime.close(); + } +}); + +test("receipts of one Turn that disagree on the binding are a conflict, not a preference", async () => { + const runtime = await runtimeWith([ + receipt("turn-a", closeoutRequired("turn-a", "todo_alpha")), + receipt("turn-a", closeoutRequired("turn-a", "todo_beta")), + ]); + try { + await assert.rejects( + () => preflight(runtime.root), + /conflicting settlement identities/, + ); + } finally { + await runtime.close(); + } +}); + +test("a Turn with many receipts resolves the one that binds its settlement", async () => { + const crowded: JsonObject[] = []; + for (let index = 0; index < 40; index += 1) { + crowded.push( + receipt(`turn-crowded`, { + stall_observation: "not_applicable", + revision: index, + }), + ); + } + crowded.push(receipt("turn-crowded", closeoutRequired("turn-crowded", "todo_crowded"))); + crowded.push(receipt("turn-newer", closeoutRequired("turn-newer", "todo_newer"))); + const runtime = await runtimeWith(crowded); + try { + const result = await preflight(runtime.root); + assert.equal(result.turns_validated, 2); + assert.equal(candidateFrom(result).binding_id, "todo_newer"); + } finally { + await runtime.close(); + } +}); + +test("a prior Turn that already validates its settlement needs no bound facts", async () => { + // The receipt alone is not a settlement: the readback resolves the identity + // and reports the missing writeback and spend receipts instead... + const unsettled = await runtimeWith([ + receipt("turn-a", closeoutRequired("turn-a", "todo_alpha")), + ]); + try { + assert.equal((await preflight(unsettled.root)).status, "candidate"); + } finally { + await unsettled.close(); + } + // ...so a caller can never observe "validated" without the persisted + // writeback and spend receipts, which the Python CLI fixtures cover + // end-to-end through the real entrypoint. +}); + +test("a malformed or foreign log line fails the read instead of erasing a closeout", async () => { + const runtime = await runtimeWith([ + receipt("turn-a", closeoutRequired("turn-a", "todo_alpha")), + ]); + try { + await writeFile( + join(runtime.root, "goals", GOAL, "rollout-event-log.jsonl"), + `${JSON.stringify(receipt("turn-a", closeoutRequired("turn-a", "todo_alpha")))}\nnot json\n`, + "utf8", + ); + 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")), + ]); + try { + const candidate = candidateFrom(await preflight(runtime.root)); + assert.throws( + () => reduce(candidate), + /requires its bound Todo facts/, + ); + } finally { + await runtime.close(); + } +}); + +test("the missing-receipt checkpoint is the closeout policy's own vocabulary", async () => { + const runtime = await runtimeWith([ + receipt("turn-a", closeoutRequired("turn-a", "todo_alpha")), + ]); + try { + const candidate = candidateFrom(await preflight(runtime.root)); + assert.throws( + () => reduce( + candidate, + ["quota_spend_receipt", "durable_writeback_receipt"], + READ_TODO(ADVANCEMENT_OPEN), + ), + /must be the closeout receipts/, + ); + assert.throws( + () => reduce(candidate, ["unknown_receipt"], READ_TODO(ADVANCEMENT_OPEN)), + /must be the closeout receipts/, + ); + assert.throws( + () => reduce(candidate, [], READ_TODO(ADVANCEMENT_OPEN)), + /must be the closeout receipts/, + ); + } finally { + await runtime.close(); + } +}); + +test("an exactly committed monitor-poll closeout is accepted", async () => { + const todoId = "todo_monitor"; + const runtime = await runtimeWith([ + receipt("turn-a", closeoutRequired("turn-a", todoId)), + ]); + try { + const candidate = candidateFrom(await preflight(runtime.root)); + const verdict = reduce( + candidate, + ["durable_writeback_receipt", "quota_spend_receipt"], + READ_TODO( + {task_class: "continuous_monitor", status: "open", has_resume_when: false, + has_successor_todo_ids: false, target_key: "target", cadence: "1h"}, + {effect_id: `quota-monitor-poll:${GOAL}:${AGENT}:turn-a:todo:${todoId}`}, + ), + ); + assert.equal(verdict.status, "none"); + assert.equal(verdict.accepted_closeout, "exact_committed_quota_monitor_poll"); + } finally { + await runtime.close(); + } +}); + +test("a monitor-poll receipt for another Turn or effect is not an accepted closeout", async () => { + const todoId = "todo_monitor"; + const runtime = await runtimeWith([ + receipt("turn-a", closeoutRequired("turn-a", todoId)), + ]); + try { + const candidate = candidateFrom(await preflight(runtime.root)); + for (const effectId of [ + `quota-monitor-poll:${GOAL}:${AGENT}:turn-other:todo:${todoId}`, + `quota-monitor-poll:${GOAL}:${AGENT}:turn-a:todo:todo_other`, + "", + ]) { + const verdict = reduce( + candidate, + ["durable_writeback_receipt", "quota_spend_receipt"], + READ_TODO( + {task_class: "continuous_monitor", status: "open", has_resume_when: false, + has_successor_todo_ids: false, target_key: null, cadence: null}, + {effect_id: effectId}, + ), + ); + assert.equal(verdict.status, "recovery_required"); + } + } finally { + await runtime.close(); + } +}); + +test("a committed monitor-poll closeout is not accepted for a non-monitor Turn", async () => { + const runtime = await runtimeWith([ + receipt("turn-a", closeoutRequired("turn-a", "todo_alpha")), + ]); + try { + const candidate = candidateFrom(await preflight(runtime.root)); + const verdict = reduce( + candidate, + ["durable_writeback_receipt", "quota_spend_receipt"], + READ_TODO( + {...ADVANCEMENT_OPEN, target_key: null, cadence: null}, + {effect_id: `quota-monitor-poll:${GOAL}:${AGENT}:turn-a:todo:todo_alpha`}, + ), + ); + assert.equal(verdict.status, "recovery_required"); + assert.equal((verdict.obligation as JsonObject).repair, "lifecycle"); + } finally { + await runtime.close(); + } +}); + +test("typed lifecycle closeouts are accepted and other open Todos are not", async () => { + const runtime = await runtimeWith([ + receipt("turn-a", closeoutRequired("turn-a", "todo_alpha")), + ]); + try { + const candidate = candidateFrom(await preflight(runtime.root)); + for (const [todo, accepted] of [ + [{...ADVANCEMENT_OPEN, has_resume_when: true, has_successor_todo_ids: true}, + "typed_external_wait_with_runnable_successor"], + [{...ADVANCEMENT_OPEN, status: "done"}, + "typed_blocker_or_lifecycle_transition"], + [{...ADVANCEMENT_OPEN, status: "blocked"}, + "typed_blocker_or_lifecycle_transition"], + [{...ADVANCEMENT_OPEN, status: "deferred"}, + "typed_blocker_or_lifecycle_transition"], + // A retained wait without a runnable successor is not a closeout. + [{...ADVANCEMENT_OPEN, has_resume_when: true, has_successor_todo_ids: false}, null], + // The persisted status is read verbatim; padding is not a transition. + [{...ADVANCEMENT_OPEN, status: "done "}, null], + ] as const) { + const verdict = reduce( + candidate, + ["durable_writeback_receipt", "quota_spend_receipt"], + READ_TODO(todo), + ); + if (accepted === null) { + assert.equal(verdict.status, "recovery_required"); + } else { + assert.equal(verdict.status, "none"); + assert.equal(verdict.accepted_closeout, accepted); + } + } + } finally { + await runtime.close(); + } +}); + +test("an unsettled Turn keeps the exact public recovery payload", async () => { + const runtime = await runtimeWith([ + receipt("turn-a", closeoutRequired("turn-a", "todo_alpha")), + ]); + try { + const candidate = candidateFrom(await preflight(runtime.root)); + const verdict = reduce( + candidate, + ["durable_writeback_receipt", "quota_spend_receipt"], + READ_TODO({...ADVANCEMENT_OPEN, target_key: "key-1", cadence: "1h"}), + ); + assert.equal(verdict.status, "recovery_required"); + assert.deepEqual(verdict.recovery, { + schema_version: "unsettled_host_turn_recovery_v0", + state: "recovery_required", + reason_code: "required_closeout_receipt_missing", + prior_turn_instance_id: "turn-a", + prior_event_id: "event-turn-a", + binding_kind: "todo", + binding_id: "todo_alpha", + settlement_effect_id: `${GOAL}:${AGENT}:todo_alpha:turn-a`, + missing_receipts: ["durable_writeback_receipt", "quota_spend_receipt"], + accepted_closeouts: [...ACCEPTED_CLOSEOUTS], + external_state_policy: "typed_host_observation_only", + quota_policy: "no_spend_for_recovery_transition", + binding_task_class: "advancement_task", + binding_target_key: "key-1", + binding_cadence: "1h", + repair: "lifecycle", + }); + const obligation = verdict.obligation as JsonObject; + assert.equal(obligation.lane, "control_plane_recovery"); + assert.equal(obligation.must_attempt_work, true); + assert.equal(obligation.delivery_allowed, false); + assert.equal(obligation.notify, "DONT_NOTIFY"); + assert.equal(obligation.repair, "lifecycle"); + } finally { + await runtime.close(); + } +}); + +test("a monitor-bound recovery names only the receipts that are actually missing", async () => { + const runtime = await runtimeWith([ + receipt("turn-a", closeoutRequired("turn-a", "todo_monitor")), + ]); + try { + const candidate = candidateFrom(await preflight(runtime.root)); + const verdict = reduce( + candidate, + ["quota_spend_receipt"], + READ_TODO({ + task_class: "continuous_monitor", status: "open", has_resume_when: false, + has_successor_todo_ids: false, target_key: "target", cadence: "30m", + }), + ); + const recovery = verdict.recovery as JsonObject; + assert.equal(verdict.status, "recovery_required"); + assert.equal(recovery.repair, "monitor_poll"); + assert.equal(recovery.binding_target_key, "target"); + assert.deepEqual(recovery.missing_receipts, ["quota_spend_receipt"]); + assert.equal((verdict.obligation as JsonObject).repair, "monitor_poll"); + } finally { + await runtime.close(); + } +}); + +test("an autonomous replan binding recovers without a Todo binding", async () => { + const replanId = `replan-${"b".repeat(16)}`; + const runtime = await runtimeWith([ + receipt("turn-a", closeoutRequired("turn-a", null, replanId)), + ]); + try { + const candidate = candidateFrom(await preflight(runtime.root)); + assert.equal(candidate.binding_kind, "autonomous_replan"); + const verdict = reduce( + candidate, + ["durable_writeback_receipt", "quota_spend_receipt"], + READ_TODO(null), + ); + const recovery = verdict.recovery as JsonObject; + assert.equal(recovery.binding_kind, "autonomous_replan"); + assert.equal(recovery.binding_id, replanId); + assert.equal((verdict.obligation as JsonObject).repair, "lifecycle"); + assert.equal("binding_task_class" in recovery, false); + } finally { + await runtime.close(); + } +}); + +test("candidate bindings are revalidated before any verdict", async () => { + assert.throws( + () => reduce({ + prior_turn_instance_id: "turn-a", + event_id: "event-turn-a", + binding_kind: "todo", + binding_id: "TODO_ALPHA", + settlement_effect_id: null, + }, ["durable_writeback_receipt", "quota_spend_receipt"], READ_TODO(null)), + /not a legal Todo id/, + ); + assert.throws( + () => reduce({ + prior_turn_instance_id: "turn-a", + event_id: null, + binding_kind: "autonomous_replan", + binding_id: "replan-not-an-obligation", + settlement_effect_id: null, + }, ["durable_writeback_receipt", "quota_spend_receipt"], READ_TODO(null)), + /autonomous replan binding is malformed/, + ); +}); diff --git a/tsconfig.control-plane.json b/tsconfig.control-plane.json index db61112b93..bcf19da66a 100644 --- a/tsconfig.control-plane.json +++ b/tsconfig.control-plane.json @@ -19,6 +19,7 @@ "loopx/control_plane/effect_runtime_io.ts", "loopx/control_plane/effect_runtime_server.ts", "loopx/control_plane/runtime_decode.ts", + "loopx/control_plane/rollout_receipt_log.ts", "loopx/control_plane/capability_hooks.ts", "loopx/control_plane/post_writeback_hook_transaction.ts", "loopx/control_plane/presentation/action_review_plan.ts", @@ -47,6 +48,8 @@ "loopx/control_plane/goals/goal_amendment_proposal.ts", "loopx/control_plane/quota/settlement_workspace_causality.ts", "loopx/control_plane/quota/settlement_readback.ts", + "loopx/control_plane/quota/heartbeat_receipt_identity.ts", + "loopx/control_plane/quota/unsettled_host_turn_recovery.ts", "loopx/control_plane/quota/monitor_poll_commit.ts", "loopx/control_plane/quota/accounting_artifact_transaction.ts", "loopx/control_plane/quota/spend_commit.ts",