Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions loopx/control_plane/effect_runtime_handlers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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],
Expand Down
45 changes: 0 additions & 45 deletions loopx/control_plane/quota/heartbeat_receipt.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
182 changes: 182 additions & 0 deletions loopx/control_plane/quota/heartbeat_receipt_identity.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,182 @@
/**
* One owner for the persisted heartbeat-receipt settlement identity rule.
*
* A Turn can persist several `quota_should_run` events. The effective receipt
* is the one that binds a settlement identity; a Turn whose events disagree on
* that binding is an identity conflict and must fail closed instead of letting
* a caller infer, upgrade, or silently prefer one binding.
*
* Both the settlement readback and the prior-host-Turn closeout selection read
* this rule, so it lives here rather than in either caller.
*/
import type { JsonObject } from "../effect_program.ts";
import {settlementIdentity, type SettlementIdentity} from "../effect_program.ts";
import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
import { jsonObject } from "../runtime_decode.ts";

const TODO_ID_PATTERN = /^todo_[a-z0-9_-]{3,64}$/;
const REPLAN_OBLIGATION_ID_PATTERN = /^replan-[a-f0-9]{16}$/;
/** Public diagnostic preserved from the retired Python identity rule. */
const IDENTITY_CONFLICT_CODE = "heartbeat_receipt_identity_conflict";

function receiptIdentityConflict(message: string): EffectRuntimeRequestError {
return new EffectRuntimeRequestError(message, IDENTITY_CONFLICT_CODE);
}

export function heartbeatReceiptDetails(event: JsonObject | null): JsonObject {
return jsonObject(event?.details) ?? {};
}

export function optionalHeartbeatString(value: unknown): string | null {
if (value === null || value === undefined) return null;
return String(value).trim() || null;
}

export function normalizeHeartbeatTodoId(value: unknown): string | null {
const candidate = String(value ?? "").trim().toLowerCase();
return candidate && TODO_ID_PATTERN.test(candidate) ? candidate : null;
}

export function normalizeHeartbeatReplanObligationId(
value: unknown,
): string | null {
const candidate = String(value ?? "").trim();
return candidate && REPLAN_OBLIGATION_ID_PATTERN.test(candidate)
? candidate
: null;
}

export interface HeartbeatReceiptBinding extends JsonObject {
binding_kind: "todo" | "autonomous_replan";
/** The Todo id or autonomous replan obligation id, per `binding_kind`. */
binding_id: string;
/**
* The effect id the receipt declares, kept verbatim for projection.
*
* It is intentionally not repaired from the identity's derived effect id: a
* receipt that names an effect the settlement authority could not derive
* stays visible in the projected payload instead of being silently rewritten.
*/
settlement_effect_id: string | null;
/** The conflict-detection key: one Turn may declare only one of these. */
identity_key: string;
}

/**
* The only receipt fields the settlement binding rule can read.
*
* The persisted rollout event and the settlement readback's own projection both
* decode into this shape, so one rule serves both readers. The goal and Agent
* a receipt belongs to are not receipt facts: they are the scope the reader
* already selected.
*/
export interface HeartbeatReceiptFact extends JsonObject {
event_id: string | null;
run_id: string | null;
todo_id: string | null;
replan_obligation_id: string | null;
settlement_effect_id: string | null;
closeout_required: boolean;
}

export function heartbeatReceiptFactFromEvent(
event: JsonObject,
): HeartbeatReceiptFact {
const eventDetails = heartbeatReceiptDetails(event);
return {
event_id: optionalHeartbeatString(event.event_id),
run_id: optionalHeartbeatString(event.run_id),
todo_id: optionalHeartbeatString(eventDetails.todo_id),
replan_obligation_id: optionalHeartbeatString(
eventDetails.replan_obligation_id,
),
settlement_effect_id: optionalHeartbeatString(
eventDetails.settlement_effect_id,
),
closeout_required: eventDetails.closeout_required === true,
};
}

/**
* Resolve the settlement binding a persisted receipt declares.
*
* A receipt without a binding is not a settlement identity; it is reported as
* `null` so the caller can decide whether an unbound receipt is admissible.
*/
export function heartbeatReceiptBinding(
goalId: string,
agentId: string,
fact: HeartbeatReceiptFact,
): HeartbeatReceiptBinding | null {
const declaredTodoId = optionalHeartbeatString(fact.todo_id);
const replanObligationId = normalizeHeartbeatReplanObligationId(
fact.replan_obligation_id,
);
const declaredEffectId = optionalHeartbeatString(fact.settlement_effect_id);
if (declaredTodoId && replanObligationId) {
throw receiptIdentityConflict(
"heartbeat receipt has conflicting Todo and autonomous replan bindings",
);
}
if (declaredEffectId && !declaredTodoId && !replanObligationId) {
throw receiptIdentityConflict(
"heartbeat receipt has an effect identity without a Todo or autonomous replan binding; refuse to infer or upgrade it",
);
}
if (!declaredTodoId && !replanObligationId) return null;
// A Todo-binding the settlement authority cannot address is not a binding we
// may act on: acting on the raw string would create a recovery obligation
// whose identity no other reader can reproduce, and dropping it would let a
// required closeout disappear. Refuse the read instead.
const todoId = declaredTodoId;
if (todoId !== null && normalizeHeartbeatTodoId(todoId) !== todoId) {
throw receiptIdentityConflict(
"heartbeat receipt declares a Todo binding that is not a legal Todo id",
);
}
const identity = settlementIdentity({
goal_id: goalId,
agent_id: agentId,
todo_id: todoId,
turn_instance_id: fact.run_id ?? "",
replan_obligation_id: replanObligationId,
});
return {
binding_kind: identity.binding_kind === "todo"
? "todo"
: "autonomous_replan",
binding_id: identity.binding_id,
settlement_effect_id: declaredEffectId,
identity_key: `${identity.binding_kind}\u0000${identity.binding_id}\u0000${
declaredEffectId ?? identity.effect_id
}`,
};
}

/**
* Reduce one Turn's receipts to its effective receipt.
*
* Receipts without a settlement binding are admissible only while the Turn
* declares no binding at all; they cannot outrank a bound receipt, and they
* cannot silently turn a conflicting Turn into a valid one.
*/
export function selectEffectiveHeartbeatReceipt<Value>(
goalId: string,
agentId: string,
entries: readonly { fact: HeartbeatReceiptFact; value: Value }[],
): Value | null {
if (entries.length === 0) return null;
const identities = new Map<string, Value>();
for (const entry of entries) {
const binding = heartbeatReceiptBinding(goalId, agentId, entry.fact);
if (binding) identities.set(binding.identity_key, entry.value);
}
if (identities.size > 1) {
throw receiptIdentityConflict(
"heartbeat receipt has conflicting settlement identities for the same goal, agent, and turn",
);
}
return identities.size === 1
? [...identities.values()][0]!
: entries.at(-1)!.value;
}
Loading
Loading