diff --git a/loopx/control_plane/quota/settlement_readback.ts b/loopx/control_plane/quota/settlement_readback.ts index 5c9f8d7feb..55a6ea80c1 100644 --- a/loopx/control_plane/quota/settlement_readback.ts +++ b/loopx/control_plane/quota/settlement_readback.ts @@ -125,6 +125,14 @@ function settlementScope( return {runtimeRoot, goalId}; } +/** Validate the canonical scope before callers perform optional receipt IO. */ +export function validateQuotaSettlementScope( + runtimeRootValue: unknown, + goalIdValue: unknown, +): { runtimeRoot: string; goalId: string } { + return settlementScope(runtimeRootValue, goalIdValue); +} + /** Receipt verification owns progress; a durable debit alone is not settlement. */ function settlementProgress( identity: SettlementResult, writeback: SettlementResult, spend: SettlementResult, diff --git a/loopx/control_plane/work_items/replan_history_settlement.ts b/loopx/control_plane/work_items/replan_history_settlement.ts index 0282ce3997..eb5d489ff4 100644 --- a/loopx/control_plane/work_items/replan_history_settlement.ts +++ b/loopx/control_plane/work_items/replan_history_settlement.ts @@ -1,8 +1,13 @@ /** Replan history IO: qualify work through the existing settlement owner. */ import type { JsonObject } from "../effect_program.ts"; -import { jsonObject, requireJsonObject, requireNonEmptyString } from "../runtime_decode.ts"; -import { readQuotaSettlementSnapshot, readQuotaSettlementForAdmittedOwnerFromSnapshot, - QUOTA_SETTLEMENT_READBACK_REQUEST_SCHEMA } from "../quota/settlement_readback.ts"; +import { jsonObject, requireJsonObject } from "../runtime_decode.ts"; +import { + readQuotaSettlementSnapshot, + readQuotaSettlementForAdmittedOwnerFromSnapshot, + QUOTA_SETTLEMENT_READBACK_REQUEST_SCHEMA, + validateQuotaSettlementScope, +} from "../quota/settlement_readback.ts"; +import { readGoalRolloutEventSnapshot, strictGoalRolloutEvents } from "../rollout_receipt_log.ts"; import { parseQuotaAccountingOwner, withQuotaAccountingOwner, withBorrowedQuotaAccountingOwner, quotaOwnerOwnsProjection } from "../quota/source_admission.ts"; import { projectReplanHistory } from "./replan_history.ts"; @@ -13,23 +18,43 @@ export async function projectSettledReplanHistory(value: unknown): Promise { - const snapshot = await readQuotaSettlementSnapshot(runtimeRoot, goalId); // Scope before ACKs too: a retired instance cannot reset the current lane. const runs = (Array.isArray(request.runs) ? request.runs : []).filter( raw => quotaOwnerOwnsProjection(owner, jsonObject(raw)?.goal_ref)); + const rolloutSnapshot = await readGoalRolloutEventSnapshot(runtimeRoot, goalId); + const events = strictGoalRolloutEvents(rolloutSnapshot); + const receiptKeys = new Set(events.filter(event => + event.event_kind === "quota_should_run" && event.goal_id === goalId && + typeof event.agent_id === "string" && typeof event.run_id === "string" && + quotaOwnerOwnsProjection(owner, event.goal_ref) + ).map(event => JSON.stringify([event.agent_id, event.run_id]))); + const settlementRuns = runs.filter(raw => { + const row = jsonObject(raw); + return row !== null && typeof row.agent_id === "string" && row.agent_id !== "" && + typeof row.turn_id === "string" && row.turn_id !== "" && + receiptKeys.has(JSON.stringify([row.agent_id, row.turn_id])); + }); + // History rows without a matching should-run receipt cannot be qualified + // by settlement. Avoid parsing the strict quota run ledger unless at least + // one exact current-owner Turn can contribute to effective-cadence counts. + if (settlementRuns.length === 0) { + return projectReplanHistory({...request, runs}, new Set()); + } + const snapshot = await readQuotaSettlementSnapshot(runtimeRoot, goalId, rolloutSnapshot); const qualified = new Set(); const seen = new Set(); - for (const raw of runs) { + for (const raw of settlementRuns) { const row = jsonObject(raw); - if (!row || typeof row.agent_id !== "string" || !row.agent_id || - typeof row.turn_id !== "string" || !row.turn_id) continue; + if (!row) continue; const key = JSON.stringify([row.agent_id, row.turn_id]); if (seen.has(key)) continue; seen.add(key); diff --git a/tests/control_plane_ts/quota_settlement_readback.test.ts b/tests/control_plane_ts/quota_settlement_readback.test.ts index 7f2a3cd9e6..940ad14804 100644 --- a/tests/control_plane_ts/quota_settlement_readback.test.ts +++ b/tests/control_plane_ts/quota_settlement_readback.test.ts @@ -13,6 +13,10 @@ import { dirname, join } from "node:path"; import test from "node:test"; import { settlementIdentity } from "../../loopx/control_plane/effect_program.ts"; +import { + createEffectRuntimeHandlers, + dispatchEffectRuntimeMethod, +} from "../../loopx/control_plane/effect_runtime_handlers.ts"; import { acquireFileMutationLock, releaseFileMutationLock, @@ -1661,3 +1665,79 @@ test("effective cadence scopes settlement and ACKs to its admitted Goal instance await rm(root, { recursive: true, force: true }); } }); + +test("effective cadence skips settlement read without a matching should-run receipt", async () => { + const { projectSettledReplanHistory } = await import( + "../../loopx/control_plane/work_items/replan_history_settlement.ts"); + const root = await fixture({guard: false}); + await appendFile( + join(root, "goals", goalId, "runs", "index.jsonl"), + "not-json\n", + ); + const request = { + schema_version: "replan_history_request_v0", + operation: "periodic", + agent_id: agentId, + monitor_agent_id: agentId, + neutral_classifications: [], + stall_threshold: 2, + periodic_threshold: 1, + monitor_threshold: 6, + streak_threshold: 5, + monitor_schema: "dead_monitor_repeat_v0", + todos: {monitors: [], advancements: [], resume: null}, + settlement_source: {runtime_root: root, goal_id: goalId}, + runs: [{ + agent_id: agentId, + public_agent_id: agentId, + monitor_agent_id: agentId, + classification: "progress", + generated_at: "2026-09-24T10:00:00Z", + observed_at: 1, + turn_id: "turn-history-only", + accepted_ack: false, + progress: null, + monitor: {}, + }], + }; + try { + const result = await projectSettledReplanHistory(request); + assert.equal(result.trigger, null); + } finally { + await rm(root, {recursive: true, force: true}); + } +}); + +test( + "replan history RPC rejects a relative runtime root before empty-receipt early return", + async () => { + const handlers = createEffectRuntimeHandlers({ + fingerprint: "replan-history-runtime-root-test", + requestShutdown() {}, + }); + const request = { + schema_version: "replan_history_request_v0", + operation: "periodic", + agent_id: agentId, + monitor_agent_id: agentId, + neutral_classifications: [], + stall_threshold: 2, + periodic_threshold: 1, + monitor_threshold: 6, + streak_threshold: 5, + monitor_schema: "dead_monitor_repeat_v0", + todos: {monitors: [], advancements: [], resume: null}, + settlement_source: {runtime_root: "relative", goal_id: goalId}, + runs: [], + }; + + await assert.rejects( + dispatchEffectRuntimeMethod( + handlers, + "work_item.replan_history.project", + request, + ), + /runtime_root must be absolute/, + ); + }, +);