From 7c57a60f7598b9df9cacdfe1367f96952c813ed2 Mon Sep 17 00:00:00 2001 From: JUN Date: Mon, 14 Sep 2026 22:00:14 +0900 Subject: [PATCH 1/2] feat(lib): reserve tokens and output before dispatch, and keep the ledger across restart (#4546) Part of the stacked delivery closing the remaining OCX-4546 cost-guard scope. Verification posture: local suite, typecheck, install and build NOT run by explicit instruction. Hosted CI at the exact final head is the only proof. Pushed with --no-verify. --- scripts/test-layout/layout.json | 2 + src/lib/spend-reservation-ledger.ts | 368 +++++++++++++++++++++ src/lib/workflow-budget.ts | 109 +++++- tests/fixtures/test-layout-expected.json | 2 + tests/lib/spend-reservation-ledger.test.ts | 155 +++++++++ tests/lib/workflow-budget.test.ts | 164 +++++++++ 6 files changed, 789 insertions(+), 11 deletions(-) create mode 100644 src/lib/spend-reservation-ledger.ts create mode 100644 tests/lib/spend-reservation-ledger.test.ts create mode 100644 tests/lib/workflow-budget.test.ts diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 38fcac80cd..6b18f66c9c 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1271,6 +1271,7 @@ "skill-ocx.test.ts": "ci-workflows", "slug-codec.test.ts": "codex-integration", "sponsor-presets.test.ts": "providers", + "spend-reservation-ledger.test.ts": "lib", "sse-client-frame-bounds.test.ts": "responses", "sse-decoder.test.ts": "responses", "sse-failed-tail.test.ts": "responses", @@ -1414,6 +1415,7 @@ "windows-user-principal.test.ts": "windows", "winsw-stop-hardening.test.ts": "windows", "winsw.test.ts": "service", + "workflow-budget.test.ts": "lib", "ws-endpoint.test.ts": "responses", "ws-failure-stage.test.ts": "responses", "ws-upstream-reuse.test.ts": "responses", diff --git a/src/lib/spend-reservation-ledger.ts b/src/lib/spend-reservation-ledger.ts new file mode 100644 index 0000000000..0286e65e07 --- /dev/null +++ b/src/lib/spend-reservation-ledger.ts @@ -0,0 +1,368 @@ +/** + * Durable token spend reservation, above the send-count workflow guard (#4546). + * + * The count cap treats a 1k-token send and a 150k-token send as the same unit, and the + * in-memory ledger forgets everything on restart: an exhausted root came back with a fresh + * allowance after every relaunch, and a second process never saw the first one's spend at + * all. This ledger reserves TOKENS before dispatch and rebuilds its state from a journal + * under the opencodex home directory, so an exhausted scope is still exhausted after a + * restart. + * + * A reservation is always the request's whole input plus its ENFORCEABLE output ceiling -- + * the caller's max_output_tokens, or the model's documented cap when the caller sent none. + * Never an optimistic estimate, and never shrunk by a cache-hit expectation: a prefix that + * misses is billed in full, so the safety figure reserves as if it misses. Cache + * expectations may inform efficiency reporting; they do not move this number. + * + * Admission requires, at every scope that applies at once -- root workflow, authenticated + * identity, and account pool: + * + * settled spend + in-flight reservations + unresolved spend + this reservation <= limit + * + * Unresolved spend is the conservative residue of a send whose usage frame was lost: the + * tokens may have been billed, so the reservation is moved to unresolved rather than + * released. Minting a new root id mints no new budget because the identity and pool scopes + * still hold the spend. + * + * SUPPORTED TOPOLOGY: this guarantees a single proxy process against its own journal. The + * file is append-friendly, but nothing here serializes two live processes writing it, so a + * second proxy sharing the same OPENCODEX_HOME is explicitly outside the guarantee -- that + * needs a shared store with cross-process atomicity and is declared out of scope rather + * than implied. + */ + +import { appendFileSync, existsSync, mkdirSync, readFileSync } from "node:fs"; +import { dirname, join } from "node:path"; +// Definition-site import, not the ../config barrel -- same reasoning as +// src/quota/reset-seen-store.ts: the barrel pulls ~154 modules into a hot path. +import { getConfigDir } from "../config/paths"; +import { assertNotRealHomeUnderTest } from "./test-home-guard"; + +export const SPEND_LEDGER_JOURNAL_FILENAME = "spend-ledger.jsonl"; + +export type SpendScope = "root" | "identity" | "pool"; + +export interface SpendScopeLimit { + /** + * Approved token ceiling for the scope. Undefined means OBSERVE ONLY: spend is still + * accounted and reported, but nothing is refused. That is the unconfigured default -- + * an install that never opted in keeps the count caps and is not newly refused. + */ + readonly maxTokens?: number; +} + +export interface SpendReservationPolicy { + readonly root: SpendScopeLimit; + readonly identity: SpendScopeLimit; + readonly pool: SpendScopeLimit; + /** + * How long a dormant scope's accounting is retained. A scope may be dropped only when it + * is BOTH inactive (no open reservation) AND not exhausted inside this window; dropping + * an exhausted scope would hand it a fresh allowance on next use. + */ + readonly retentionMs: number; +} + +/** + * Unconfigured default: every limit undefined, so token accounting runs in observe-only + * mode and the count caps remain the only enforcement. Real numbers belong behind + * explicit operator configuration. + */ +export const DEFAULT_SPEND_RESERVATION_POLICY: SpendReservationPolicy = { + root: {}, + identity: {}, + pool: {}, + retentionMs: 7 * 24 * 60 * 60_000, +}; + +export interface SpendScopes { + readonly rootId?: string; + readonly identityId?: string; + readonly poolId?: string; +} + +export interface SpendUsage { + readonly inputTokens: number; + readonly outputTokens: number; +} + +export interface SpendReservationRequest { + /** Stable id of the physical send. Settlement is idempotent on this key. */ + readonly sendId: string; + readonly scopes: SpendScopes; + readonly inputTokens: number; + /** Enforceable output ceiling -- max_output_tokens or the model's documented cap. */ + readonly outputCeilingTokens: number; + readonly at?: number; +} + +export type SpendDenial = { + readonly reason: "spend-limit-exceeded"; + readonly scope: SpendScope; + readonly scopeId: string; + readonly limit: number; + readonly projected: number; +}; + +export type SpendReservationDecision = + | { readonly reserved: true; readonly sendId: string; readonly tokens: number } + | { readonly reserved: false; readonly denial: SpendDenial }; + +interface ScopeState { + settled: number; + reserved: number; + unresolved: number; + lastSeenAt: number; +} + +interface Reservation { + readonly scopes: SpendScopes; + readonly tokens: number; + status: "open" | "settled" | "lost"; + readonly at: number; +} + +type JournalRecord = + | { v: 1; kind: "reserve"; sendId: string; scopes: SpendScopes; tokens: number; at: number } + | { v: 1; kind: "settle"; sendId: string; tokens: number; at: number } + | { v: 1; kind: "lost"; sendId: string; at: number }; + +/** + * Append-only persistence. `read` returns raw lines so replay tolerates a torn tail write: + * an unparseable final line is skipped, which loses at most the record that never made it + * to disk intact. + */ +export interface SpendJournal { + read(): string[]; + append(line: string): void; +} + +export function createFileSpendJournal(path: string): SpendJournal { + return { + read(): string[] { + if (!existsSync(path)) return []; + return readFileSync(path, "utf8").split("\n").filter((line) => line.length > 0); + }, + append(line: string): void { + const dir = dirname(path); + // The guard runs before any mutation so a rejected write leaves nothing behind. + assertNotRealHomeUnderTest(dir); + mkdirSync(dir, { recursive: true, mode: 0o700 }); + appendFileSync(path, line + "\n", { encoding: "utf8", mode: 0o600 }); + }, + }; +} + +export interface ScopeSpendSnapshot { + readonly settled: number; + readonly reserved: number; + readonly unresolved: number; + readonly exhausted: boolean; +} + +export interface SpendReservationLedger { + reserve(request: SpendReservationRequest): SpendReservationDecision; + /** + * Settle with real usage. Returns false when the send is unknown or already resolved -- + * double settlement is as wrong as none, so a repeat call changes nothing. + */ + settle(sendId: string, usage: SpendUsage): boolean; + /** + * Usage never arrived. The reservation moves to unresolved spend -- it may have been + * billed -- rather than being released. Idempotent on the same key as settle. + */ + markLost(sendId: string): boolean; + snapshot(scope: SpendScope, scopeId: string): ScopeSpendSnapshot | undefined; + exhausted(scope: SpendScope, scopeId: string): boolean; + /** Drop dormant scopes per the retention rule in SpendReservationPolicy. */ + prune(now?: number): void; + /** Journal writes that failed; a nonzero count means durability is degraded. */ + readonly persistFailures: number; +} + +const scopeKey = (scope: SpendScope, id: string): string => scope + "\0" + id; + +const sanitizeTokens = (value: number): number => + Number.isFinite(value) ? Math.max(0, Math.trunc(value)) : 0; + +export function createSpendReservationLedger(options: { + readonly journal?: SpendJournal; + readonly policy?: SpendReservationPolicy; + readonly now?: () => number; +} = {}): SpendReservationLedger { + const policy = options.policy ?? DEFAULT_SPEND_RESERVATION_POLICY; + const journal = options.journal; + const now = options.now ?? (() => Date.now()); + const scopes = new Map(); + const reservations = new Map(); + let persistFailures = 0; + + const scopeState = (scope: SpendScope, id: string): ScopeState => { + const key = scopeKey(scope, id); + let state = scopes.get(key); + if (!state) { + state = { settled: 0, reserved: 0, unresolved: 0, lastSeenAt: 0 }; + scopes.set(key, state); + } + return state; + }; + + const limitFor = (scope: SpendScope): number | undefined => policy[scope].maxTokens; + + const isExhausted = (scope: SpendScope, state: ScopeState): boolean => { + const limit = limitFor(scope); + return limit !== undefined && state.settled + state.reserved + state.unresolved >= limit; + }; + + const eachScope = (targets: SpendScopes, fn: (scope: SpendScope, id: string, state: ScopeState) => void): void => { + if (targets.rootId !== undefined) fn("root", targets.rootId, scopeState("root", targets.rootId)); + if (targets.identityId !== undefined) fn("identity", targets.identityId, scopeState("identity", targets.identityId)); + if (targets.poolId !== undefined) fn("pool", targets.poolId, scopeState("pool", targets.poolId)); + }; + + const append = (record: JournalRecord): void => { + if (!journal) return; + try { + journal.append(JSON.stringify(record)); + } catch { + // In-memory state still bounds this process; the counter is how a caller learns the + // restart guarantee degraded instead of discovering it after the fact. + persistFailures += 1; + } + }; + + const applyReserve = (sendId: string, targets: SpendScopes, tokens: number, at: number): void => { + if (reservations.has(sendId)) return; + reservations.set(sendId, { scopes: targets, tokens, status: "open", at }); + eachScope(targets, (_scope, _id, state) => { + state.reserved += tokens; + state.lastSeenAt = Math.max(state.lastSeenAt, at); + }); + }; + + const applySettle = (sendId: string, tokens: number, at: number, lost: boolean): void => { + const reservation = reservations.get(sendId); + if (!reservation || reservation.status !== "open") return; + reservation.status = lost ? "lost" : "settled"; + eachScope(reservation.scopes, (_scope, _id, state) => { + state.reserved = Math.max(0, state.reserved - reservation.tokens); + // A lost send keeps its whole reservation as unresolved spend; a settled one books + // the real figure, which may be lower OR higher than the ceiling that was reserved. + if (lost) state.unresolved += reservation.tokens; + else state.settled += tokens; + state.lastSeenAt = Math.max(state.lastSeenAt, at); + }); + }; + + // Rebuild from the journal before serving: an exhausted scope must still be exhausted + // after a restart, which is the whole reason this store exists. + if (journal) { + for (const line of journal.read()) { + let record: JournalRecord; + try { + record = JSON.parse(line) as JournalRecord; + } catch { + continue; + } + if (record.v !== 1) continue; + if (record.kind === "reserve") applyReserve(record.sendId, record.scopes, sanitizeTokens(record.tokens), record.at); + else if (record.kind === "settle") applySettle(record.sendId, sanitizeTokens(record.tokens), record.at, false); + else if (record.kind === "lost") applySettle(record.sendId, 0, record.at, true); + } + } + + return { + get persistFailures() { return persistFailures; }, + + reserve(request: SpendReservationRequest): SpendReservationDecision { + const tokens = sanitizeTokens(request.inputTokens) + sanitizeTokens(request.outputCeilingTokens); + const at = request.at ?? now(); + // Check every scope before mutating any: a refusal must not leave a partial + // reservation booked on the scopes that would have passed. + const checks: { scope: SpendScope; id: string; state: ScopeState }[] = []; + eachScope(request.scopes, (scope, id, state) => checks.push({ scope, id, state })); + for (const { scope, id, state } of checks) { + const limit = limitFor(scope); + if (limit === undefined) continue; + const projected = state.settled + state.reserved + state.unresolved + tokens; + if (projected > limit) { + return { reserved: false, denial: { reason: "spend-limit-exceeded", scope, scopeId: id, limit, projected } }; + } + } + applyReserve(request.sendId, request.scopes, tokens, at); + append({ v: 1, kind: "reserve", sendId: request.sendId, scopes: request.scopes, tokens, at }); + return { reserved: true, sendId: request.sendId, tokens }; + }, + + settle(sendId: string, usage: SpendUsage): boolean { + const reservation = reservations.get(sendId); + if (!reservation || reservation.status !== "open") return false; + const tokens = sanitizeTokens(usage.inputTokens) + sanitizeTokens(usage.outputTokens); + applySettle(sendId, tokens, now(), false); + append({ v: 1, kind: "settle", sendId, tokens, at: now() }); + return true; + }, + + markLost(sendId: string): boolean { + const reservation = reservations.get(sendId); + if (!reservation || reservation.status !== "open") return false; + applySettle(sendId, 0, now(), true); + append({ v: 1, kind: "lost", sendId, at: now() }); + return true; + }, + + snapshot(scope: SpendScope, scopeId: string): ScopeSpendSnapshot | undefined { + const state = scopes.get(scopeKey(scope, scopeId)); + if (!state) return undefined; + return { + settled: state.settled, + reserved: state.reserved, + unresolved: state.unresolved, + exhausted: isExhausted(scope, state), + }; + }, + + exhausted(scope: SpendScope, scopeId: string): boolean { + const state = scopes.get(scopeKey(scope, scopeId)); + return state !== undefined && isExhausted(scope, state); + }, + + prune(at: number = now()): void { + const cutoff = at - policy.retentionMs; + for (const [key, state] of scopes) { + const scope = key.slice(0, key.indexOf("\0")) as SpendScope; + // Removal requires BOTH inactive and not exhausted inside the window. An + // exhausted-but-idle scope that was dropped would be recreated fresh under the + // same id -- the exact laundering the ceiling exists to stop. + if (state.reserved > 0 || state.lastSeenAt >= cutoff) continue; + if (isExhausted(scope, state)) continue; + scopes.delete(key); + } + for (const [sendId, reservation] of reservations) { + if (reservation.status === "open" || reservation.at >= cutoff) continue; + reservations.delete(sendId); + } + }, + }; +} + +let sharedLedger: SpendReservationLedger | undefined; + +/** + * Process-wide ledger backed by the journal under OPENCODEX_HOME. Created lazily so + * importing the module -- or running a request path that never reserves -- touches no + * disk. + */ +export function sharedSpendLedger(): SpendReservationLedger { + if (!sharedLedger) { + sharedLedger = createSpendReservationLedger({ + journal: createFileSpendJournal(join(getConfigDir(), SPEND_LEDGER_JOURNAL_FILENAME)), + }); + } + return sharedLedger; +} + +/** Test seam. Production never discards the ledger: that would reset a spent budget. */ +export function resetSharedSpendLedgerForTest(): void { + sharedLedger = undefined; +} diff --git a/src/lib/workflow-budget.ts b/src/lib/workflow-budget.ts index 5cbe8e665f..0aa0cb158b 100644 --- a/src/lib/workflow-budget.ts +++ b/src/lib/workflow-budget.ts @@ -10,11 +10,22 @@ * header when the client supplies one. A retry is not a new user task and gets no new * allowance; a genuinely new top-level request does. * - * This ledger is process-local and in-memory. It bounds a single proxy process honestly and - * says nothing about a second process sharing the same account pool; that needs a shared - * durable store and is declared out of scope rather than implied. + * Two caps intersect here. The COUNT caps (concurrency, distinct children, physical sends) + * are process-local and in-memory. The TOKEN cap is the durable spend-reservation ledger in + * spend-reservation-ledger.ts: when the caller supplies a spend request, admission also + * reserves input + enforceable output ceiling against the root, identity and pool scopes, + * and that accounting survives a restart. The count caps alone remain the guarantee for a + * second process sharing the pool; the durable ledger's single-process topology is stated + * in that module's header and applies here unchanged. */ +import { + sharedSpendLedger, + type SpendReservationLedger, + type SpendScope, + type SpendUsage, +} from "./spend-reservation-ledger"; + export interface WorkflowBudgetPolicy { /** Children admitted concurrently under one root. */ readonly maxConcurrentChildren: number; @@ -42,7 +53,8 @@ export const DEFAULT_WORKFLOW_BUDGET_POLICY: WorkflowBudgetPolicy = { export type WorkflowDenial = | "workflow-concurrency-exhausted" | "workflow-sends-exhausted" - | "workflow-children-exhausted"; + | "workflow-children-exhausted" + | "workflow-spend-exhausted"; export type WorkflowLane = "interactive" | "worker"; @@ -53,7 +65,30 @@ export interface WorkflowAdmission { export type WorkflowDecision = | { admitted: true; lease: WorkflowAdmission } - | { admitted: false; reason: WorkflowDenial; rootId: string }; + | { + admitted: false; + reason: WorkflowDenial; + rootId: string; + /** Which spend scope refused, when the denial came from the token ledger. */ + spendScope?: SpendScope; + }; + +/** + * Token reservation attached to an admission. `outputCeilingTokens` is the ENFORCEABLE + * ceiling -- the caller's max_output_tokens or the model's documented cap, never an + * optimistic estimate and never shrunk by a cache-hit expectation. Omitting `spend` + * entirely keeps the historical count-only admission, which is also what an unconfigured + * install gets: token accounting is observed by default and refuses nothing until an + * operator sets real limits. + */ +export interface WorkflowSpendRequest { + /** Stable id of the physical send; settlement is idempotent on this key. */ + readonly sendId: string; + readonly identityId?: string; + readonly poolId?: string; + readonly inputTokens: number; + readonly outputCeilingTokens: number; +} interface WorkflowState { active: number; @@ -64,13 +99,17 @@ interface WorkflowState { const roots = new Map(); -function pruneOldestRoot(): void { +function pruneOldestRoot(policy: WorkflowBudgetPolicy, spendLedger?: SpendReservationLedger): void { let oldestKey: string | undefined; let oldestAt = Number.POSITIVE_INFINITY; for (const [key, state] of roots) { // An active root is never evicted: dropping it would hand its fan-out a fresh allowance, - // which is the exact laundering this ledger exists to prevent. + // which is the exact laundering this ledger exists to prevent. The same holds for an + // EXHAUSTED-but-idle root -- count-exhausted or spend-exhausted -- because recreating it + // fresh under the same id resets the very ceiling that already fired. if (state.active > 0) continue; + if (state.sends >= policy.maxPhysicalSends) continue; + if (spendLedger?.exhausted("root", key) === true) continue; if (state.lastSeenMs < oldestAt) { oldestAt = state.lastSeenMs; oldestKey = key; } } if (oldestKey !== undefined) roots.delete(oldestKey); @@ -81,6 +120,12 @@ function pruneOldestRoot(): void { * * `childId` distinguishes the members of a fan-out; omit it for the root's own turns. * An interactive lane may use the reserved slots a worker lane may not. + * + * When `spend` is given, admission also reserves its tokens on the spend ledger -- at the + * root, identity and pool scopes at once -- before a concurrency slot is taken. A turn + * released without settlement moves its reservation to unresolved spend, because a send + * whose usage never arrived may still have been billed; releasing it would understate the + * scope. */ export function admitWorkflowTurn( rootId: string | undefined, @@ -88,11 +133,16 @@ export function admitWorkflowTurn( policy: WorkflowBudgetPolicy = DEFAULT_WORKFLOW_BUDGET_POLICY, childId?: string, now: number = Date.now(), + spend?: WorkflowSpendRequest, + spendLedger?: SpendReservationLedger, ): WorkflowDecision | undefined { if (!rootId) return undefined; + // An explicit ledger is consulted even without a spend request, so root eviction can + // still see spend-exhausted entries. With neither, no token tracking is in play. + const ledger = spendLedger ?? (spend ? sharedSpendLedger() : undefined); let state = roots.get(rootId); if (!state) { - if (roots.size >= policy.maxTrackedRoots) pruneOldestRoot(); + if (roots.size >= policy.maxTrackedRoots) pruneOldestRoot(policy, ledger); state = { active: 0, sends: 0, children: new Set(), lastSeenMs: now }; roots.set(rootId, state); } @@ -112,6 +162,24 @@ export function admitWorkflowTurn( return { admitted: false, reason: "workflow-concurrency-exhausted", rootId }; } + if (spend && ledger) { + const decision = ledger.reserve({ + sendId: spend.sendId, + scopes: { rootId, identityId: spend.identityId, poolId: spend.poolId }, + inputTokens: spend.inputTokens, + outputCeilingTokens: spend.outputCeilingTokens, + at: now, + }); + if (!decision.reserved) { + return { + admitted: false, + reason: "workflow-spend-exhausted", + rootId, + spendScope: decision.denial.scope, + }; + } + } + state.active += 1; if (childId !== undefined) state.children.add(childId); let released = false; @@ -123,9 +191,14 @@ export function admitWorkflowTurn( if (released) return; released = true; const current = roots.get(rootId); - if (!current) return; - current.active = Math.max(0, current.active - 1); - current.lastSeenMs = Date.now(); + if (current) { + current.active = Math.max(0, current.active - 1); + current.lastSeenMs = Date.now(); + } + // A turn that ends without a settlement keeps its cost as unresolved spend rather + // than being released: the send may have been billed even though its usage frame + // never arrived. markLost is a no-op once settleWorkflowSpend already ran. + if (spend && ledger) ledger.markLost(spend.sendId); }, }, }; @@ -143,6 +216,20 @@ export function chargeWorkflowSends(rootId: string | undefined, sends: number): state.lastSeenMs = Date.now(); } +/** + * Settle a send's reservation with the usage the response actually reported. Idempotent + * per send id -- a second call returns false and books nothing. When the usage frame was + * lost, call this never and let the lease's release move the reservation to unresolved + * spend, or call the ledger's markLost directly. + */ +export function settleWorkflowSpend( + sendId: string, + usage: SpendUsage, + spendLedger?: SpendReservationLedger, +): boolean { + return (spendLedger ?? sharedSpendLedger()).settle(sendId, usage); +} + /** * Whether this root has already spent its whole physical-send ceiling. * diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 2fcc00f1ed..2b2722d86c 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1099,6 +1099,7 @@ "skill-ocx.test.ts": "ci-workflows", "slug-codec.test.ts": "codex-integration", "sponsor-presets.test.ts": "providers", + "spend-reservation-ledger.test.ts": "lib", "sse-client-frame-bounds.test.ts": "responses", "sse-decoder.test.ts": "responses", "sse-failed-tail.test.ts": "responses", @@ -1242,6 +1243,7 @@ "windows-user-principal.test.ts": "windows", "winsw-stop-hardening.test.ts": "windows", "winsw.test.ts": "service", + "workflow-budget.test.ts": "lib", "ws-endpoint.test.ts": "responses", "ws-failure-stage.test.ts": "responses", "ws-upstream-reuse.test.ts": "responses", diff --git a/tests/lib/spend-reservation-ledger.test.ts b/tests/lib/spend-reservation-ledger.test.ts new file mode 100644 index 0000000000..b2c613dbba --- /dev/null +++ b/tests/lib/spend-reservation-ledger.test.ts @@ -0,0 +1,155 @@ +import { describe, expect, test } from "bun:test"; +import { + createSpendReservationLedger, + type SpendJournal, + type SpendReservationPolicy, +} from "../../src/lib/spend-reservation-ledger"; + +/** In-memory journal: same replay contract as the file store, without touching disk. */ +const memoryJournal = (): SpendJournal & { lines: string[] } => { + const lines: string[] = []; + return { lines, read: () => [...lines], append: (line) => { lines.push(line); } }; +}; + +const policy = (maxTokens: number | undefined, retentionMs = 60_000): SpendReservationPolicy => ({ + root: { maxTokens }, + identity: { maxTokens }, + pool: { maxTokens }, + retentionMs, +}); + +describe("spend reservation ledger", () => { + test("reserves input plus the enforceable output ceiling and refuses at the boundary", () => { + const ledger = createSpendReservationLedger({ policy: policy(100), now: () => 1_000 }); + // 60 input + 40 ceiling = 100 exactly: the boundary admits. + expect(ledger.reserve({ + sendId: "s1", + scopes: { rootId: "r1" }, + inputTokens: 60, + outputCeilingTokens: 40, + }).reserved).toBe(true); + // One more token projects past the limit and is refused, naming the scope. + const denied = ledger.reserve({ + sendId: "s2", + scopes: { rootId: "r1" }, + inputTokens: 1, + outputCeilingTokens: 0, + }); + expect(denied.reserved).toBe(false); + if (!denied.reserved) { + expect(denied.denial.scope).toBe("root"); + expect(denied.denial.limit).toBe(100); + expect(denied.denial.projected).toBe(101); + } + // The refused reservation booked nothing: settling its send id is a no-op. + expect(ledger.settle("s2", { inputTokens: 1, outputTokens: 0 })).toBe(false); + }); + + test("enforces root, identity and pool scopes at once, so a fresh root id mints no budget", () => { + const ledger = createSpendReservationLedger({ policy: policy(100), now: () => 1_000 }); + const req = (sendId: string, rootId: string) => ({ + sendId, + scopes: { rootId, identityId: "user-1", poolId: "pool-1" }, + inputTokens: 60, + outputCeilingTokens: 40, + }); + expect(ledger.reserve(req("s1", "root-a")).reserved).toBe(true); + // A brand-new root still carries the identity and pool spend: all three scopes are + // checked, so laundering through a fresh root id fails on the identity scope. + const denied = ledger.reserve(req("s2", "root-b")); + expect(denied.reserved).toBe(false); + if (!denied.reserved) expect(denied.denial.scope).toBe("identity"); + // A different identity under the same pool is still stopped at the pool scope. + const poolDenied = ledger.reserve({ + sendId: "s3", + scopes: { rootId: "root-c", identityId: "user-2", poolId: "pool-1" }, + inputTokens: 60, + outputCeilingTokens: 40, + }); + expect(poolDenied.reserved).toBe(false); + if (!poolDenied.reserved) expect(poolDenied.denial.scope).toBe("pool"); + }); + + test("settlement is idempotent per send id", () => { + const ledger = createSpendReservationLedger({ policy: policy(1_000), now: () => 1_000 }); + ledger.reserve({ sendId: "s1", scopes: { rootId: "r1" }, inputTokens: 100, outputCeilingTokens: 100 }); + expect(ledger.settle("s1", { inputTokens: 90, outputTokens: 10 })).toBe(true); + // The double settlement books nothing: reserved stays released exactly once. + expect(ledger.settle("s1", { inputTokens: 90, outputTokens: 10 })).toBe(false); + const snap = ledger.snapshot("root", "r1"); + expect(snap?.settled).toBe(100); + expect(snap?.reserved).toBe(0); + // markLost after a settlement is likewise a no-op. + expect(ledger.markLost("s1")).toBe(false); + }); + + test("lost usage becomes unresolved spend instead of being released", () => { + const ledger = createSpendReservationLedger({ policy: policy(150), now: () => 1_000 }); + ledger.reserve({ sendId: "s1", scopes: { rootId: "r1" }, inputTokens: 100, outputCeilingTokens: 50 }); + expect(ledger.markLost("s1")).toBe(true); + const snap = ledger.snapshot("root", "r1"); + expect(snap?.reserved).toBe(0); + expect(snap?.unresolved).toBe(150); + // Unresolved spend still counts: the full reservation may have been billed. + expect(ledger.exhausted("root", "r1")).toBe(true); + expect(ledger.reserve({ + sendId: "s2", scopes: { rootId: "r1" }, inputTokens: 1, outputCeilingTokens: 0, + }).reserved).toBe(false); + }); + + test("an exhausted root stays exhausted across a simulated restart", () => { + const journal = memoryJournal(); + const first = createSpendReservationLedger({ journal, policy: policy(100), now: () => 1_000 }); + first.reserve({ sendId: "s1", scopes: { rootId: "r1" }, inputTokens: 60, outputCeilingTokens: 40 }); + first.settle("s1", { inputTokens: 60, outputTokens: 40 }); + expect(first.exhausted("root", "r1")).toBe(true); + + // Restart: a new ledger replays the same journal and refuses the same root. + const second = createSpendReservationLedger({ journal, policy: policy(100), now: () => 2_000 }); + expect(second.exhausted("root", "r1")).toBe(true); + expect(second.reserve({ + sendId: "s2", scopes: { rootId: "r1" }, inputTokens: 1, outputCeilingTokens: 0, + }).reserved).toBe(false); + // And the replayed settlement is still idempotent after the rebuild. + expect(second.settle("s1", { inputTokens: 60, outputTokens: 40 })).toBe(false); + }); + + test("the unconfigured default observes spend but refuses nothing", () => { + const ledger = createSpendReservationLedger({ now: () => 1_000 }); + for (let i = 0; i < 10; i += 1) { + expect(ledger.reserve({ + sendId: `s${i}`, scopes: { rootId: "r1" }, inputTokens: 1_000_000, outputCeilingTokens: 1_000_000, + }).reserved).toBe(true); + } + const snap = ledger.snapshot("root", "r1"); + expect(snap?.reserved).toBe(20_000_000); + expect(snap?.exhausted).toBe(false); + }); + + test("prune removes a dormant under-limit scope but never an exhausted one", () => { + const journal = memoryJournal(); + const ledger = createSpendReservationLedger({ journal, policy: policy(100, 1_000), now: () => 0 }); + ledger.reserve({ sendId: "s1", scopes: { rootId: "spent" }, inputTokens: 60, outputCeilingTokens: 40 }); + ledger.settle("s1", { inputTokens: 60, outputTokens: 40 }); + ledger.reserve({ sendId: "s2", scopes: { rootId: "light" }, inputTokens: 10, outputCeilingTokens: 0 }); + ledger.settle("s2", { inputTokens: 10, outputTokens: 0 }); + + ledger.prune(10_000); + // Both are idle and past the retention window, but only the under-limit one may go. + expect(ledger.snapshot("root", "light")).toBeUndefined(); + const spent = ledger.snapshot("root", "spent"); + expect(spent?.exhausted).toBe(true); + expect(ledger.reserve({ + sendId: "s3", scopes: { rootId: "spent" }, inputTokens: 1, outputCeilingTokens: 0, + }).reserved).toBe(false); + }); + + test("a torn tail line in the journal is skipped on replay", () => { + const journal = memoryJournal(); + const first = createSpendReservationLedger({ journal, policy: policy(100), now: () => 1_000 }); + first.reserve({ sendId: "s1", scopes: { rootId: "r1" }, inputTokens: 60, outputCeilingTokens: 40 }); + journal.lines.push("{not-json"); + const second = createSpendReservationLedger({ journal, policy: policy(100), now: () => 2_000 }); + expect(second.exhausted("root", "r1")).toBe(true); + }); +}); diff --git a/tests/lib/workflow-budget.test.ts b/tests/lib/workflow-budget.test.ts new file mode 100644 index 0000000000..16f711ff70 --- /dev/null +++ b/tests/lib/workflow-budget.test.ts @@ -0,0 +1,164 @@ +import { beforeEach, describe, expect, test } from "bun:test"; +import { + createSpendReservationLedger, + type SpendJournal, + type SpendReservationPolicy, +} from "../../src/lib/spend-reservation-ledger"; +import { + admitWorkflowTurn, + chargeWorkflowSends, + DEFAULT_WORKFLOW_BUDGET_POLICY, + resetWorkflowBudgetsForTest, + settleWorkflowSpend, + workflowBudgetSnapshot, + workflowSendCeilingReached, + type WorkflowBudgetPolicy, +} from "../../src/lib/workflow-budget"; + +const memoryJournal = (): SpendJournal & { lines: string[] } => { + const lines: string[] = []; + return { lines, read: () => [...lines], append: (line) => { lines.push(line); } }; +}; + +const spendPolicy = (maxTokens: number | undefined): SpendReservationPolicy => ({ + root: { maxTokens }, + identity: { maxTokens }, + pool: { maxTokens }, + retentionMs: 60_000, +}); + +const smallPolicy: WorkflowBudgetPolicy = { + maxConcurrentChildren: 2, + maxPhysicalSends: 3, + maxDistinctChildren: 2, + interactiveReserve: 1, + maxTrackedRoots: 2, +}; + +beforeEach(() => { + resetWorkflowBudgetsForTest(); +}); + +describe("workflow count caps", () => { + test("the physical-send ceiling refuses before dispatch", () => { + admitWorkflowTurn("r1", "interactive", smallPolicy); + chargeWorkflowSends("r1", 3); + expect(workflowSendCeilingReached("r1", smallPolicy)).toBe(true); + const decision = admitWorkflowTurn("r1", "interactive", smallPolicy); + expect(decision?.admitted).toBe(false); + if (decision && !decision.admitted) expect(decision.reason).toBe("workflow-sends-exhausted"); + }); + + test("a worker lane may not take the interactive reserve", () => { + const workerCeiling = smallPolicy.maxConcurrentChildren - smallPolicy.interactiveReserve; + for (let i = 0; i < workerCeiling; i += 1) { + expect(admitWorkflowTurn("r1", "worker", smallPolicy, `c${i}`)?.admitted).toBe(true); + } + const denied = admitWorkflowTurn("r1", "worker", smallPolicy, "c-extra"); + expect(denied?.admitted).toBe(false); + if (denied && !denied.admitted) expect(denied.reason).toBe("workflow-concurrency-exhausted"); + // The interactive turn that owns the fan-out still gets in. + expect(admitWorkflowTurn("r1", "interactive", smallPolicy)?.admitted).toBe(true); + }); + + test("distinct children are capped", () => { + // Concurrency is deliberately not the binding constraint here. + const policy: WorkflowBudgetPolicy = { + maxConcurrentChildren: 10, + maxPhysicalSends: 100, + maxDistinctChildren: 2, + interactiveReserve: 0, + maxTrackedRoots: 10, + }; + admitWorkflowTurn("r1", "worker", policy, "c1"); + admitWorkflowTurn("r1", "worker", policy, "c2"); + const denied = admitWorkflowTurn("r1", "worker", policy, "c3"); + expect(denied?.admitted).toBe(false); + if (denied && !denied.admitted) expect(denied.reason).toBe("workflow-children-exhausted"); + }); + + test("an exhausted-but-idle root is never evicted to make room", () => { + // Fill the root to its send ceiling, then let it go idle: only the new + // exhausted-but-idle rule can still protect it from eviction. + const filled = admitWorkflowTurn("full", "interactive", smallPolicy); + chargeWorkflowSends("full", 3); + if (filled?.admitted) filled.lease.release(); + // Two more roots arrive, forcing eviction pressure at maxTrackedRoots = 2. + admitWorkflowTurn("n1", "interactive", smallPolicy); + admitWorkflowTurn("n2", "interactive", smallPolicy); + // The exhausted root survived the prune: recreating it must not reset its allowance. + const decision = admitWorkflowTurn("full", "interactive", smallPolicy); + expect(decision?.admitted).toBe(false); + if (decision && !decision.admitted) expect(decision.reason).toBe("workflow-sends-exhausted"); + }); +}); + +describe("workflow spend reservation", () => { + test("the token cap intersects the count caps", () => { + const ledger = createSpendReservationLedger({ journal: memoryJournal(), policy: spendPolicy(100), now: () => 1_000 }); + const spend = (sendId: string) => ({ + sendId, inputTokens: 60, outputCeilingTokens: 40, + }); + expect(admitWorkflowTurn("r1", "interactive", DEFAULT_WORKFLOW_BUDGET_POLICY, + undefined, 1_000, spend("s1"), ledger)?.admitted).toBe(true); + const denied = admitWorkflowTurn("r1", "interactive", DEFAULT_WORKFLOW_BUDGET_POLICY, + undefined, 1_000, spend("s2"), ledger); + expect(denied?.admitted).toBe(false); + if (denied && !denied.admitted) { + expect(denied.reason).toBe("workflow-spend-exhausted"); + expect(denied.spendScope).toBe("root"); + } + }); + + test("identity and pool scopes hold spend across fresh root ids", () => { + const ledger = createSpendReservationLedger({ journal: memoryJournal(), policy: spendPolicy(100), now: () => 1_000 }); + const spend = (sendId: string) => ({ + sendId, identityId: "user-1", poolId: "pool-1", inputTokens: 60, outputCeilingTokens: 40, + }); + expect(admitWorkflowTurn("root-a", "interactive", DEFAULT_WORKFLOW_BUDGET_POLICY, + undefined, 1_000, spend("s1"), ledger)?.admitted).toBe(true); + const denied = admitWorkflowTurn("root-b", "interactive", DEFAULT_WORKFLOW_BUDGET_POLICY, + undefined, 1_000, spend("s2"), ledger); + expect(denied?.admitted).toBe(false); + if (denied && !denied.admitted) expect(denied.spendScope).toBe("identity"); + }); + + test("settlement is idempotent and a release without it becomes unresolved spend", () => { + const ledger = createSpendReservationLedger({ journal: memoryJournal(), policy: spendPolicy(1_000), now: () => 1_000 }); + const admitted = admitWorkflowTurn("r1", "interactive", DEFAULT_WORKFLOW_BUDGET_POLICY, + undefined, 1_000, { sendId: "s1", inputTokens: 100, outputCeilingTokens: 50 }, ledger); + expect(admitted?.admitted).toBe(true); + expect(settleWorkflowSpend("s1", { inputTokens: 90, outputTokens: 10 }, ledger)).toBe(true); + // Double settlement books nothing. + expect(settleWorkflowSpend("s1", { inputTokens: 90, outputTokens: 10 }, ledger)).toBe(false); + if (admitted?.admitted) admitted.lease.release(); + const settled = ledger.snapshot("root", "r1"); + expect(settled?.settled).toBe(100); + expect(settled?.unresolved).toBe(0); + + // A turn released without settlement keeps its cost as unresolved spend. + const lost = admitWorkflowTurn("r1", "interactive", DEFAULT_WORKFLOW_BUDGET_POLICY, + undefined, 2_000, { sendId: "s2", inputTokens: 30, outputCeilingTokens: 20 }, ledger); + if (lost?.admitted) lost.lease.release(); + const after = ledger.snapshot("root", "r1"); + expect(after?.unresolved).toBe(50); + // And a late settle for the lost send is correctly refused. + expect(settleWorkflowSpend("s2", { inputTokens: 30, outputTokens: 20 }, ledger)).toBe(false); + }); + + test("a spend-exhausted idle root survives eviction pressure", () => { + const ledger = createSpendReservationLedger({ journal: memoryJournal(), policy: spendPolicy(100), now: () => 1_000 }); + const exhausted = admitWorkflowTurn("full", "interactive", smallPolicy, + undefined, 1_000, { sendId: "s1", inputTokens: 60, outputCeilingTokens: 40 }, ledger); + // The lease is released so the root is idle, but its spend is exhausted. + if (exhausted?.admitted) exhausted.lease.release(); + admitWorkflowTurn("n1", "interactive", smallPolicy, undefined, 2_000, undefined, ledger); + admitWorkflowTurn("n2", "interactive", smallPolicy, undefined, 3_000, undefined, ledger); + const snap = workflowBudgetSnapshot("full"); + expect(snap).toBeDefined(); + const denied = admitWorkflowTurn("full", "interactive", smallPolicy, + undefined, 4_000, { sendId: "s2", inputTokens: 1, outputCeilingTokens: 0 }, ledger); + expect(denied?.admitted).toBe(false); + if (denied && !denied.admitted) expect(denied.reason).toBe("workflow-spend-exhausted"); + }); +}); From 1d5d299e75e8db88a9cc69967ff5c4d05f25bee9 Mon Sep 17 00:00:00 2001 From: JUN Date: Mon, 14 Sep 2026 23:14:20 +0900 Subject: [PATCH 2/2] fix(lib): refuse a duplicate send id, fail closed on a lost journal write, and bound retention (#4546) Review findings on the reservation ledger: a reused send id authorised a free dispatch, a failed journal append still admitted the request, replay parsed unvalidated JSON, retention was unbounded, an undispatched reservation booked phantom debt, and raw account identifiers reached disk. Verification posture: local suite, typecheck, install and build NOT run by explicit instruction. Hosted CI at the exact final head is the only proof. Pushed with --no-verify. --- scripts/test-layout/layout.json | 1 + src/lib/spend-reservation-ledger.ts | 752 +++++++++++++++++--- src/lib/workflow-budget.ts | 104 ++- tests/fixtures/test-layout-expected.json | 1 + tests/lib/spend-ledger-file-journal.test.ts | 65 ++ tests/lib/spend-reservation-ledger.test.ts | 246 ++++++- tests/lib/workflow-budget.test.ts | 79 +- 7 files changed, 1126 insertions(+), 122 deletions(-) create mode 100644 tests/lib/spend-ledger-file-journal.test.ts diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 6b18f66c9c..f12d79f775 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1271,6 +1271,7 @@ "skill-ocx.test.ts": "ci-workflows", "slug-codec.test.ts": "codex-integration", "sponsor-presets.test.ts": "providers", + "spend-ledger-file-journal.test.ts": "lib", "spend-reservation-ledger.test.ts": "lib", "sse-client-frame-bounds.test.ts": "responses", "sse-decoder.test.ts": "responses", diff --git a/src/lib/spend-reservation-ledger.ts b/src/lib/spend-reservation-ledger.ts index 0286e65e07..21cdd78c72 100644 --- a/src/lib/spend-reservation-ledger.ts +++ b/src/lib/spend-reservation-ledger.ts @@ -29,16 +29,49 @@ * second proxy sharing the same OPENCODEX_HOME is explicitly outside the guarantee -- that * needs a shared store with cross-process atomicity and is declared out of scope rather * than implied. + * + * Five properties this file owes its callers. Each one was absent in the first draft, and a + * budget that can be bypassed is worse than no budget because it looks like protection: + * + * 1. IDENTITY OF A SEND. A send id is either KNOWN -- and then reserving it again is refused + * rather than waved through booking nothing -- or FULLY forgotten, and then it books a + * fresh reservation. There is no third state where the ledger recognises an id and + * charges nothing for it, which is what let one id authorise unlimited physical sends. + * 2. DURABILITY BEFORE ADMISSION. Under a configured limit the reserve record must be on + * disk before the request is admitted. Failing open on a disk-full or permission error + * forgets the request across a restart, which is the exact case durability exists for. + * Observe-only mode still admits, and says so through `durable: false`. + * 3. REPLAY VALIDATES. Every journal record is checked field by field before it moves a + * counter. A corrupt record in the MIDDLE of the file would silently undercount, so it + * fails accounting closed instead; only an unparseable FINAL line -- a torn tail write -- + * is dropped quietly. + * 4. BOUNDED RETENTION. Cleanup runs automatically, writes durable tombstones so replay + * cannot resurrect what it removed, and compacts the journal to a checkpoint. When + * nothing can be evicted safely, admission is refused rather than made room for by + * forgetting an exhausted scope -- forgetting one is the laundering this layer prevents. + * 5. NOTHING IDENTIFYING ON DISK. Root ids come from a client header and identity ids are + * credential ids, so the journal stores salted aliases only, under owner-only permissions + * that are re-applied to an EXISTING file rather than trusted from its creation. */ -import { appendFileSync, existsSync, mkdirSync, readFileSync } from "node:fs"; +import { appendFileSync, chmodSync, existsSync, mkdirSync, readFileSync, renameSync, statSync, writeFileSync } from "node:fs"; +import { createHash, randomBytes } from "node:crypto"; import { dirname, join } from "node:path"; // Definition-site import, not the ../config barrel -- same reasoning as // src/quota/reset-seen-store.ts: the barrel pulls ~154 modules into a hot path. import { getConfigDir } from "../config/paths"; import { assertNotRealHomeUnderTest } from "./test-home-guard"; +// Windows chmod does not remove inherited ACEs; this is the repository's icacls path. +import { hardenSecretPath } from "./windows-secret-acl"; export const SPEND_LEDGER_JOURNAL_FILENAME = "spend-ledger.jsonl"; +/** + * Per-install alias salt, beside the journal. Losing it is exactly as bad as losing the + * journal -- both reset accounting, both live in the same 0700 directory -- so it is not a + * new weakness, and keeping it out of the journal stops a copied or attached journal from + * being reversible by dictionary attack on guessable pool and identity ids. + */ +export const SPEND_LEDGER_SALT_FILENAME = "spend-ledger.salt"; export type SpendScope = "root" | "identity" | "pool"; @@ -61,8 +94,28 @@ export interface SpendReservationPolicy { * an exhausted scope would hand it a fresh allowance on next use. */ readonly retentionMs: number; + /** + * Hard ceiling on tracked scopes. Retention alone bounds nothing: a caller minting a fresh + * root id per request fills the map long before the window elapses. At the ceiling the + * ledger evicts the oldest scope that is safe to forget -- idle, under its limit, past + * retention -- and if there is none it REFUSES the new scope. Refusing is the only answer + * left: the alternative is evicting an exhausted scope, which hands it a fresh allowance. + */ + readonly maxTrackedScopes?: number; + /** Hard ceiling on remembered send ids, with the same evict-or-refuse rule. */ + readonly maxTrackedSends?: number; + /** + * Journal records after which the file is compacted into a single checkpoint. Without + * this the file grows forever even while the in-memory maps stay bounded, and replay + * resurrects every entry cleanup removed. + */ + readonly compactAfterRecords?: number; } +const DEFAULT_MAX_TRACKED_SCOPES = 4_096; +const DEFAULT_MAX_TRACKED_SENDS = 16_384; +const DEFAULT_COMPACT_AFTER_RECORDS = 8_192; + /** * Unconfigured default: every limit undefined, so token accounting runs in observe-only * mode and the count caps remain the only enforcement. Real numbers belong behind @@ -96,16 +149,40 @@ export interface SpendReservationRequest { readonly at?: number; } -export type SpendDenial = { - readonly reason: "spend-limit-exceeded"; - readonly scope: SpendScope; - readonly scopeId: string; - readonly limit: number; - readonly projected: number; -}; +/** + * Why a reservation was refused. Every member refuses a DISPATCH: none of them is an + * "already fine, carry on" answer, because that is precisely how a duplicate send id used + * to buy an unlimited number of physical sends while the scope totals never moved. + */ +export type SpendDenial = + | { + readonly reason: "spend-limit-exceeded"; + readonly scope: SpendScope; + readonly scopeId: string; + readonly limit: number; + readonly projected: number; + } + /** This send id is already known -- open, settled, lost or abandoned. */ + | { readonly reason: "duplicate-send-id"; readonly sendId: string } + /** The reserve record could not be written, and a configured limit needs it to survive. */ + | { readonly reason: "reserve-not-durable"; readonly sendId: string } + /** Replay rejected records mid-file, so no scope total can be proven complete. */ + | { readonly reason: "journal-corrupt"; readonly corruptRecords: number } + /** Tracking is full and nothing may be forgotten safely. */ + | { readonly reason: "tracking-capacity-exhausted"; readonly scope?: SpendScope }; export type SpendReservationDecision = - | { readonly reserved: true; readonly sendId: string; readonly tokens: number } + | { + readonly reserved: true; + readonly sendId: string; + readonly tokens: number; + /** + * False only in observe-only mode, where the reservation was admitted although its + * journal record did not reach disk. A restart will not remember this spend; the flag + * is how a caller learns that instead of discovering it after the fact. + */ + readonly durable: boolean; + } | { readonly reserved: false; readonly denial: SpendDenial }; interface ScopeState { @@ -115,44 +192,241 @@ interface ScopeState { lastSeenAt: number; } +/** + * `open` means admitted but not yet handed to a transport: it may still be abandoned for + * free. `dispatched` means bytes left for upstream, so from there a missing usage frame is + * unresolved SPEND rather than a release -- it may have been billed. Only a dispatched send + * can become `lost`; only an undispatched one can become `abandoned`. + */ +type ReservationStatus = "open" | "dispatched" | "settled" | "lost" | "abandoned"; + +interface ScopeRef { + readonly scope: SpendScope; + readonly alias: string; +} + interface Reservation { - readonly scopes: SpendScopes; + readonly targets: readonly ScopeRef[]; readonly tokens: number; - status: "open" | "settled" | "lost"; + status: ReservationStatus; readonly at: number; + /** When the status last changed; drives eviction of resolved entries. */ + resolvedAt: number; } +/** + * Journal shape. Every id on disk is a salted alias, never a root header value, credential + * id or pool name. `forget` and `drop` are the tombstones that make bounded cleanup + * durable -- without them replay rebuilds exactly what cleanup removed -- and `checkpoint` + * is a whole-state snapshot that lets the file be compacted instead of growing forever. + */ type JournalRecord = - | { v: 1; kind: "reserve"; sendId: string; scopes: SpendScopes; tokens: number; at: number } - | { v: 1; kind: "settle"; sendId: string; tokens: number; at: number } - | { v: 1; kind: "lost"; sendId: string; at: number }; + | { v: 1; kind: "reserve"; send: string; targets: ScopeRef[]; tokens: number; at: number } + | { v: 1; kind: "dispatch"; send: string; at: number } + | { v: 1; kind: "settle"; send: string; tokens: number; at: number } + | { v: 1; kind: "lost"; send: string; at: number } + | { v: 1; kind: "abandon"; send: string; at: number } + | { v: 1; kind: "forget"; send: string; at: number } + | { v: 1; kind: "drop"; scope: SpendScope; alias: string; at: number } + | { + v: 1; + kind: "checkpoint"; + at: number; + scopes: { scope: SpendScope; alias: string; settled: number; unresolved: number; seenAt: number }[]; + sends: { send: string; status: ReservationStatus; targets: ScopeRef[]; tokens: number; at: number; resolvedAt: number }[]; + }; + +const isCountable = (value: unknown): value is number => + typeof value === "number" && Number.isFinite(value) && value >= 0; + +const isAlias = (value: unknown): value is string => + typeof value === "string" && value.length > 0 && value.length <= 256; + +const isScopeName = (value: unknown): value is SpendScope => + value === "root" || value === "identity" || value === "pool"; + +const isStatus = (value: unknown): value is ReservationStatus => + value === "open" || value === "dispatched" || value === "settled" + || value === "lost" || value === "abandoned"; + +const parseTargets = (value: unknown): ScopeRef[] | undefined => { + if (!Array.isArray(value) || value.length > 3) return undefined; + const targets: ScopeRef[] = []; + for (const entry of value) { + if (typeof entry !== "object" || entry === null) return undefined; + const { scope, alias } = entry as { scope?: unknown; alias?: unknown }; + if (!isScopeName(scope) || !isAlias(alias)) return undefined; + targets.push({ scope, alias }); + } + return targets; +}; + +/** + * Validate one journal line into a record, or reject it. + * + * Exported because this is the boundary where a hostile or damaged file meets the accounting: + * `JSON.parse(line) as JournalRecord` type-asserts a lie, and a bare `null` line or a + * `{"v":1,"kind":"reserve"}` with no fields crashed the rebuild rather than being rejected. + * Every field is checked, including that numbers are finite and non-negative. + */ +export function parseSpendJournalRecord(line: string): JournalRecord | undefined { + let raw: unknown; + try { + raw = JSON.parse(line); + } catch { + return undefined; + } + if (typeof raw !== "object" || raw === null || Array.isArray(raw)) return undefined; + const record = raw as Record; + if (record.v !== 1) return undefined; + if (!isCountable(record.at)) return undefined; + const at = record.at; + switch (record.kind) { + case "reserve": { + const targets = parseTargets(record.targets); + if (!isAlias(record.send) || targets === undefined || !isCountable(record.tokens)) return undefined; + return { v: 1, kind: "reserve", send: record.send, targets, tokens: record.tokens, at }; + } + case "settle": + if (!isAlias(record.send) || !isCountable(record.tokens)) return undefined; + return { v: 1, kind: "settle", send: record.send, tokens: record.tokens, at }; + case "dispatch": + if (!isAlias(record.send)) return undefined; + return { v: 1, kind: "dispatch", send: record.send, at }; + case "lost": + if (!isAlias(record.send)) return undefined; + return { v: 1, kind: "lost", send: record.send, at }; + case "abandon": + if (!isAlias(record.send)) return undefined; + return { v: 1, kind: "abandon", send: record.send, at }; + case "forget": + if (!isAlias(record.send)) return undefined; + return { v: 1, kind: "forget", send: record.send, at }; + case "drop": + if (!isScopeName(record.scope) || !isAlias(record.alias)) return undefined; + return { v: 1, kind: "drop", scope: record.scope, alias: record.alias, at }; + case "checkpoint": { + if (!Array.isArray(record.scopes) || !Array.isArray(record.sends)) return undefined; + const scopes: { scope: SpendScope; alias: string; settled: number; unresolved: number; seenAt: number }[] = []; + for (const entry of record.scopes) { + if (typeof entry !== "object" || entry === null) return undefined; + const e = entry as Record; + if (!isScopeName(e.scope) || !isAlias(e.alias)) return undefined; + if (!isCountable(e.settled) || !isCountable(e.unresolved) || !isCountable(e.seenAt)) return undefined; + scopes.push({ scope: e.scope, alias: e.alias, settled: e.settled, unresolved: e.unresolved, seenAt: e.seenAt }); + } + const sends: { send: string; status: ReservationStatus; targets: ScopeRef[]; tokens: number; at: number; resolvedAt: number }[] = []; + for (const entry of record.sends) { + if (typeof entry !== "object" || entry === null) return undefined; + const e = entry as Record; + const targets = parseTargets(e.targets); + if (!isAlias(e.send) || !isStatus(e.status) || targets === undefined) return undefined; + if (!isCountable(e.tokens) || !isCountable(e.at) || !isCountable(e.resolvedAt)) return undefined; + sends.push({ send: e.send, status: e.status, targets, tokens: e.tokens, at: e.at, resolvedAt: e.resolvedAt }); + } + return { v: 1, kind: "checkpoint", at, scopes, sends }; + } + default: + return undefined; + } +} /** - * Append-only persistence. `read` returns raw lines so replay tolerates a torn tail write: - * an unparseable final line is skipped, which loses at most the record that never made it - * to disk intact. + * Append-mostly persistence. `read` returns raw lines so replay can tell a torn TAIL write + * from corruption earlier in the file; only the former is safe to drop quietly. `append` + * THROWS when the record did not reach storage -- that signal is what lets admission refuse + * rather than admit a request a restart would forget. `rewrite` is optional: a store that + * cannot replace its contents atomically simply never compacts. */ export interface SpendJournal { read(): string[]; append(line: string): void; + rewrite?(lines: string[]): void; +} + +/** + * Re-apply owner-only permissions to a file that already exists. + * + * `mode` in a write option is honoured only when the file is CREATED, so a journal that was + * created loose -- by an older build, a restored backup, or a lax umask -- would keep its + * mode forever. Best-effort by design: a non-owner cannot chmod, and failing every append + * over it would be worse than the loose mode it is fixing. + * + * `force` marks the points where the WINDOWS ACL can actually be wrong: creation, compaction, + * and each process's replay. Windows chmod cannot drop inherited ACEs, so icacls is the real + * boundary there, and its memo keys on the file's ctime -- which every append changes. Running + * it per reservation would therefore spawn a process per send while protecting nothing an + * append can alter. On POSIX the mode is checked on every write and repaired the moment it + * drifts, which costs one stat. + */ +function hardenLedgerFile(path: string, options: { readonly force?: boolean } = {}): void { + if (process.platform === "win32") { + if (options.force) hardenSecretPath(path, { required: false }); + return; + } + try { + if ((statSync(path).mode & 0o777) === 0o600) return; + chmodSync(path, 0o600); + } catch { /* best-effort: a non-owner cannot chmod */ } } export function createFileSpendJournal(path: string): SpendJournal { + const ensureDir = (): string => { + const dir = dirname(path); + // The guard runs before any mutation so a rejected write leaves nothing behind. + assertNotRealHomeUnderTest(dir); + mkdirSync(dir, { recursive: true, mode: 0o700 }); + return dir; + }; return { read(): string[] { if (!existsSync(path)) return []; + // Replay is once per process and is the moment a journal inherited from an older build + // or a restored backup first passes through here. + hardenLedgerFile(path, { force: true }); return readFileSync(path, "utf8").split("\n").filter((line) => line.length > 0); }, append(line: string): void { - const dir = dirname(path); - // The guard runs before any mutation so a rejected write leaves nothing behind. - assertNotRealHomeUnderTest(dir); - mkdirSync(dir, { recursive: true, mode: 0o700 }); + ensureDir(); + const created = !existsSync(path); appendFileSync(path, line + "\n", { encoding: "utf8", mode: 0o600 }); + hardenLedgerFile(path, { force: created }); + }, + rewrite(lines: string[]): void { + ensureDir(); + // Same directory, so the rename is atomic on the same filesystem: a crash mid-compaction + // leaves either the old journal or the new one, never a half-written ledger. + const temp = `${path}.compact-${process.pid}`; + writeFileSync(temp, lines.map((line) => line + "\n").join(""), { encoding: "utf8", mode: 0o600 }); + hardenLedgerFile(temp, { force: true }); + renameSync(temp, path); + hardenLedgerFile(path, { force: true }); }, }; } +/** + * Load the per-install alias salt, minting it on first use. + * + * The salt must be STABLE across restarts or replay cannot match a live request to its own + * recorded spend, which would hand every scope a fresh allowance -- so it is a file, not a + * per-process value. + */ +export function loadOrCreateSpendLedgerSalt(path: string): string { + if (existsSync(path)) { + hardenLedgerFile(path, { force: true }); + const existing = readFileSync(path, "utf8").trim(); + if (/^[0-9a-f]{32,}$/.test(existing)) return existing; + } + const dir = dirname(path); + assertNotRealHomeUnderTest(dir); + mkdirSync(dir, { recursive: true, mode: 0o700 }); + const salt = randomBytes(32).toString("hex"); + writeFileSync(path, salt + "\n", { encoding: "utf8", mode: 0o600 }); + hardenLedgerFile(path, { force: true }); + return salt; +} + export interface ScopeSpendSnapshot { readonly settled: number; readonly reserved: number; @@ -162,6 +436,19 @@ export interface ScopeSpendSnapshot { export interface SpendReservationLedger { reserve(request: SpendReservationRequest): SpendReservationDecision; + /** + * The send left for upstream. Until this is called the reservation may be abandoned for + * free; after it, a missing usage frame becomes unresolved spend. Returns false when the + * send is unknown or no longer open. + */ + markDispatched(sendId: string): boolean; + /** + * The send never happened -- local validation, routing, or a refusal before any byte left + * this process. The reservation is RELEASED and books nothing, because inventing debt the + * account never incurred is its own way of breaking the budget. Refused once the send is + * dispatched: from there only settle or markLost is honest. + */ + abandon(sendId: string): boolean; /** * Settle with real usage. Returns false when the send is unknown or already resolved -- * double settlement is as wrong as none, so a repeat call changes nothing. @@ -174,13 +461,25 @@ export interface SpendReservationLedger { markLost(sendId: string): boolean; snapshot(scope: SpendScope, scopeId: string): ScopeSpendSnapshot | undefined; exhausted(scope: SpendScope, scopeId: string): boolean; - /** Drop dormant scopes per the retention rule in SpendReservationPolicy. */ + /** + * Drop dormant scopes per the retention rule in SpendReservationPolicy. Cleanup also runs + * automatically on every reservation, so nothing depends on a caller remembering this. + */ prune(now?: number): void; + /** Whether this send id is already known, and therefore refused. */ + knows(sendId: string): boolean; /** Journal writes that failed; a nonzero count means durability is degraded. */ readonly persistFailures: number; + /** + * Records replay rejected in the MIDDLE of the journal. Nonzero means no scope total can + * be proven complete, so configured limits refuse rather than undercount. + */ + readonly corruptRecords: number; + /** True when durability is degraded in either direction: failed writes or a corrupt file. */ + readonly degraded: boolean; } -const scopeKey = (scope: SpendScope, id: string): string => scope + "\0" + id; +const scopeKey = (scope: SpendScope, alias: string): string => scope + "\0" + alias; const sanitizeTokens = (value: number): number => Number.isFinite(value) ? Math.max(0, Math.trunc(value)) : 0; @@ -189,16 +488,37 @@ export function createSpendReservationLedger(options: { readonly journal?: SpendJournal; readonly policy?: SpendReservationPolicy; readonly now?: () => number; + /** + * Per-install alias salt. Production passes the file-backed value from + * `loadOrCreateSpendLedgerSalt`; an empty default is for in-memory journals, which have + * no file anyone could correlate. + */ + readonly salt?: string; } = {}): SpendReservationLedger { const policy = options.policy ?? DEFAULT_SPEND_RESERVATION_POLICY; const journal = options.journal; const now = options.now ?? (() => Date.now()); + const salt = options.salt ?? ""; + const maxTrackedScopes = policy.maxTrackedScopes ?? DEFAULT_MAX_TRACKED_SCOPES; + const maxTrackedSends = policy.maxTrackedSends ?? DEFAULT_MAX_TRACKED_SENDS; + const compactAfterRecords = policy.compactAfterRecords ?? DEFAULT_COMPACT_AFTER_RECORDS; const scopes = new Map(); const reservations = new Map(); let persistFailures = 0; + let corruptRecords = 0; + let recordsOnDisk = 0; + + /** + * Salted alias for one identifier. The raw value -- a client-supplied root header, a + * credential id, a pool name -- never leaves this function, so nothing identifying is + * written to disk or held in a map key. + */ + const aliasFor = (kind: SpendScope | "send", id: string): string => + createHash("sha256").update(salt).update("\u0000").update(kind).update("\u0000").update(id) + .digest("hex").slice(0, 32); - const scopeState = (scope: SpendScope, id: string): ScopeState => { - const key = scopeKey(scope, id); + const scopeState = (scope: SpendScope, alias: string): ScopeState => { + const key = scopeKey(scope, alias); let state = scopes.get(key); if (!state) { state = { settled: 0, reserved: 0, unresolved: 0, lastSeenAt: 0 }; @@ -214,105 +534,364 @@ export function createSpendReservationLedger(options: { return limit !== undefined && state.settled + state.reserved + state.unresolved >= limit; }; - const eachScope = (targets: SpendScopes, fn: (scope: SpendScope, id: string, state: ScopeState) => void): void => { - if (targets.rootId !== undefined) fn("root", targets.rootId, scopeState("root", targets.rootId)); - if (targets.identityId !== undefined) fn("identity", targets.identityId, scopeState("identity", targets.identityId)); - if (targets.poolId !== undefined) fn("pool", targets.poolId, scopeState("pool", targets.poolId)); + /** The scopes a request touches, as aliases. Creates no state: a refusal must leave none. */ + const refsFor = (targets: SpendScopes): ScopeRef[] => { + const refs: ScopeRef[] = []; + if (targets.rootId !== undefined) refs.push({ scope: "root", alias: aliasFor("root", targets.rootId) }); + if (targets.identityId !== undefined) refs.push({ scope: "identity", alias: aliasFor("identity", targets.identityId) }); + if (targets.poolId !== undefined) refs.push({ scope: "pool", alias: aliasFor("pool", targets.poolId) }); + return refs; }; - const append = (record: JournalRecord): void => { - if (!journal) return; + /** + * Returns whether the record reached storage. With no journal there is nothing to fail, + * and the caller's durability question is vacuously satisfied. + */ + const append = (record: JournalRecord): boolean => { + if (!journal) return true; try { journal.append(JSON.stringify(record)); + recordsOnDisk += 1; + return true; } catch { // In-memory state still bounds this process; the counter is how a caller learns the // restart guarantee degraded instead of discovering it after the fact. persistFailures += 1; + return false; } }; - const applyReserve = (sendId: string, targets: SpendScopes, tokens: number, at: number): void => { - if (reservations.has(sendId)) return; - reservations.set(sendId, { scopes: targets, tokens, status: "open", at }); - eachScope(targets, (_scope, _id, state) => { + const applyReserve = (send: string, targets: readonly ScopeRef[], tokens: number, at: number): void => { + if (reservations.has(send)) return; + reservations.set(send, { targets, tokens, status: "open", at, resolvedAt: at }); + for (const ref of targets) { + const state = scopeState(ref.scope, ref.alias); state.reserved += tokens; state.lastSeenAt = Math.max(state.lastSeenAt, at); - }); + } }; - const applySettle = (sendId: string, tokens: number, at: number, lost: boolean): void => { - const reservation = reservations.get(sendId); - if (!reservation || reservation.status !== "open") return; - reservation.status = lost ? "lost" : "settled"; - eachScope(reservation.scopes, (_scope, _id, state) => { + const isLive = (status: ReservationStatus): boolean => status === "open" || status === "dispatched"; + + /** + * Resolve a live reservation. `settled` books the real figure, `lost` keeps the whole + * reservation as unresolved spend because it may have been billed, and `abandoned` + * releases it because no byte ever left this process. + */ + const applyResolve = (send: string, outcome: "settled" | "lost" | "abandoned", tokens: number, at: number): void => { + const reservation = reservations.get(send); + if (!reservation || !isLive(reservation.status)) return; + reservation.status = outcome; + reservation.resolvedAt = at; + for (const ref of reservation.targets) { + const state = scopeState(ref.scope, ref.alias); state.reserved = Math.max(0, state.reserved - reservation.tokens); - // A lost send keeps its whole reservation as unresolved spend; a settled one books - // the real figure, which may be lower OR higher than the ceiling that was reserved. - if (lost) state.unresolved += reservation.tokens; - else state.settled += tokens; + if (outcome === "lost") state.unresolved += reservation.tokens; + else if (outcome === "settled") state.settled += tokens; state.lastSeenAt = Math.max(state.lastSeenAt, at); - }); + } + }; + + const applyDispatch = (send: string, at: number): void => { + const reservation = reservations.get(send); + if (!reservation || reservation.status !== "open") return; + reservation.status = "dispatched"; + reservation.resolvedAt = at; + }; + + /** Tombstone replay: the entry is gone, so a later reuse of the id books a fresh charge. */ + const applyForget = (send: string): void => { + const reservation = reservations.get(send); + if (!reservation || isLive(reservation.status)) return; + reservations.delete(send); + }; + + const applyDrop = (scope: SpendScope, alias: string): void => { + const state = scopes.get(scopeKey(scope, alias)); + if (!state || state.reserved > 0) return; + scopes.delete(scopeKey(scope, alias)); + }; + + const applyCheckpoint = (record: Extract): void => { + scopes.clear(); + reservations.clear(); + for (const entry of record.scopes) { + scopes.set(scopeKey(entry.scope, entry.alias), { + settled: entry.settled, + reserved: 0, + unresolved: entry.unresolved, + lastSeenAt: entry.seenAt, + }); + } + for (const entry of record.sends) { + // `reserved` is rebuilt from the live entries rather than trusted from the snapshot, + // so the two can never disagree about the same tokens. + if (isLive(entry.status)) { + applyReserve(entry.send, entry.targets, entry.tokens, entry.at); + if (entry.status === "dispatched") applyDispatch(entry.send, entry.resolvedAt); + continue; + } + reservations.set(entry.send, { + targets: entry.targets, + tokens: entry.tokens, + status: entry.status, + at: entry.at, + resolvedAt: entry.resolvedAt, + }); + } }; // Rebuild from the journal before serving: an exhausted scope must still be exhausted // after a restart, which is the whole reason this store exists. if (journal) { - for (const line of journal.read()) { - let record: JournalRecord; - try { - record = JSON.parse(line) as JournalRecord; - } catch { + const lines = journal.read(); + recordsOnDisk = lines.length; + for (let index = 0; index < lines.length; index += 1) { + const line = lines[index] as string; + const record = parseSpendJournalRecord(line); + if (!record) { + // A rejected FINAL line is a torn tail write -- the process died between the write + // and its newline -- and is dropped quietly, because that record never completed and + // therefore never authorised anything. A rejected line ANYWHERE ELSE is different: + // the records after it did complete, so skipping it silently undercounts a scope and + // hands back budget. It is counted, and a configured limit refuses on it below. + if (index < lines.length - 1) corruptRecords += 1; continue; } - if (record.v !== 1) continue; - if (record.kind === "reserve") applyReserve(record.sendId, record.scopes, sanitizeTokens(record.tokens), record.at); - else if (record.kind === "settle") applySettle(record.sendId, sanitizeTokens(record.tokens), record.at, false); - else if (record.kind === "lost") applySettle(record.sendId, 0, record.at, true); + switch (record.kind) { + case "reserve": applyReserve(record.send, record.targets, sanitizeTokens(record.tokens), record.at); break; + case "dispatch": applyDispatch(record.send, record.at); break; + case "settle": applyResolve(record.send, "settled", sanitizeTokens(record.tokens), record.at); break; + case "lost": applyResolve(record.send, "lost", 0, record.at); break; + case "abandon": applyResolve(record.send, "abandoned", 0, record.at); break; + case "forget": applyForget(record.send); break; + case "drop": applyDrop(record.scope, record.alias); break; + case "checkpoint": applyCheckpoint(record); break; + } } } + /** + * Bounded cleanup. It runs before every admission, so nothing depends on a caller + * remembering `prune()` -- the first draft exported one and no production path called it. + * Every removal writes a tombstone: without one, replay rebuilds precisely what cleanup + * removed and the file keeps growing while the maps look bounded. + * + * `force` is the at-capacity pass. It ignores the retention window but never the safety + * rule: an ACTIVE or EXHAUSTED scope is not a candidate at any pressure, because dropping + * one hands it a fresh allowance under the same id. When that leaves nothing to remove, + * the caller refuses admission rather than making room by forgetting a spent scope. + */ + const evictScopes = (at: number, force: boolean): number => { + const cutoff = at - policy.retentionMs; + const candidates: { key: string; scope: SpendScope; alias: string; seenAt: number }[] = []; + for (const [key, state] of scopes) { + const separator = key.indexOf("\0"); + const scope = key.slice(0, separator) as SpendScope; + if (state.reserved > 0) continue; + if (isExhausted(scope, state)) continue; + if (!force && state.lastSeenAt >= cutoff) continue; + candidates.push({ key, scope, alias: key.slice(separator + 1), seenAt: state.lastSeenAt }); + } + if (force) { + candidates.sort((a, b) => a.seenAt - b.seenAt); + candidates.length = Math.min(candidates.length, 1); + } + for (const candidate of candidates) { + scopes.delete(candidate.key); + append({ v: 1, kind: "drop", scope: candidate.scope, alias: candidate.alias, at }); + } + return candidates.length; + }; + + /** + * Forget resolved send ids. A forgotten id is forgotten COMPLETELY: reusing it later books + * a fresh reservation against every scope, which is conservative. The state this must never + * produce is the middle one -- an id the ledger recognises but charges nothing for. + */ + const evictSends = (at: number, force: boolean): number => { + const cutoff = at - policy.retentionMs; + const candidates: { send: string; resolvedAt: number }[] = []; + for (const [send, reservation] of reservations) { + if (isLive(reservation.status)) continue; + if (!force && reservation.resolvedAt >= cutoff) continue; + candidates.push({ send, resolvedAt: reservation.resolvedAt }); + } + if (force) { + candidates.sort((a, b) => a.resolvedAt - b.resolvedAt); + candidates.length = Math.min(candidates.length, 1); + } + for (const candidate of candidates) { + reservations.delete(candidate.send); + append({ v: 1, kind: "forget", send: candidate.send, at }); + } + return candidates.length; + }; + + /** + * Replace the journal with a single checkpoint once it has grown past its record budget. + * Bounded maps are not enough on their own: the file behind them is what replay reads, and + * an uncompacted file grows forever on unique root and send ids. + */ + const compact = (at: number): void => { + const rewrite = journal?.rewrite; + if (!journal || !rewrite || recordsOnDisk < compactAfterRecords) return; + const checkpoint: JournalRecord = { + v: 1, + kind: "checkpoint", + at, + scopes: [...scopes].map(([key, state]) => { + const separator = key.indexOf("\0"); + return { + scope: key.slice(0, separator) as SpendScope, + alias: key.slice(separator + 1), + settled: state.settled, + unresolved: state.unresolved, + seenAt: state.lastSeenAt, + }; + }), + sends: [...reservations].map(([send, reservation]) => ({ + send, + status: reservation.status, + targets: [...reservation.targets], + tokens: reservation.tokens, + at: reservation.at, + resolvedAt: reservation.resolvedAt, + })), + }; + try { + rewrite.call(journal, [JSON.stringify(checkpoint)]); + recordsOnDisk = 1; + } catch { + // Compaction is maintenance, not accounting: a failed rewrite leaves the previous + // journal intact and every figure in it still replayable. + persistFailures += 1; + } + }; + + /** The denial when tracking cannot fit this request, or undefined when it can. */ + const makeRoom = (refs: readonly ScopeRef[], at: number): SpendDenial | undefined => { + evictSends(at, false); + evictScopes(at, false); + while (reservations.size >= maxTrackedSends) { + if (evictSends(at, true) === 0) return { reason: "tracking-capacity-exhausted" }; + } + let fresh = 0; + for (const ref of refs) if (!scopes.has(scopeKey(ref.scope, ref.alias))) fresh += 1; + while (scopes.size + fresh > maxTrackedScopes) { + if (evictScopes(at, true) === 0) { + return { reason: "tracking-capacity-exhausted", scope: refs[0]?.scope }; + } + } + return undefined; + }; + return { get persistFailures() { return persistFailures; }, + get corruptRecords() { return corruptRecords; }, + get degraded() { return persistFailures > 0 || corruptRecords > 0; }, reserve(request: SpendReservationRequest): SpendReservationDecision { const tokens = sanitizeTokens(request.inputTokens) + sanitizeTokens(request.outputCeilingTokens); const at = request.at ?? now(); + const send = aliasFor("send", request.sendId); + const refs = refsFor(request.scopes); + const enforced = refs.some((ref) => limitFor(ref.scope) !== undefined); + + // A send id this ledger already knows is REFUSED. Returning success while booking + // nothing -- the old behaviour -- let one id authorise an unlimited number of physical + // sends with the scope totals never moving. + if (reservations.has(send)) { + return { reserved: false, denial: { reason: "duplicate-send-id", sendId: request.sendId } }; + } + // Replay could not prove these totals are complete, so a configured ceiling cannot be + // enforced on them. Observe-only accounting continues and reports the degradation. + if (enforced && corruptRecords > 0) { + return { reserved: false, denial: { reason: "journal-corrupt", corruptRecords } }; + } + const capacity = makeRoom(refs, at); + if (capacity) return { reserved: false, denial: capacity }; + // Check every scope before mutating any: a refusal must not leave a partial - // reservation booked on the scopes that would have passed. - const checks: { scope: SpendScope; id: string; state: ScopeState }[] = []; - eachScope(request.scopes, (scope, id, state) => checks.push({ scope, id, state })); - for (const { scope, id, state } of checks) { - const limit = limitFor(scope); + // reservation booked on the scopes that would have passed. Reading state without + // creating it matters here -- a denied request must not leave a tracked scope behind. + for (const ref of refs) { + const limit = limitFor(ref.scope); if (limit === undefined) continue; - const projected = state.settled + state.reserved + state.unresolved + tokens; + const state = scopes.get(scopeKey(ref.scope, ref.alias)); + const projected = (state ? state.settled + state.reserved + state.unresolved : 0) + tokens; if (projected > limit) { - return { reserved: false, denial: { reason: "spend-limit-exceeded", scope, scopeId: id, limit, projected } }; + const scopeId = ref.scope === "root" + ? request.scopes.rootId + : ref.scope === "identity" ? request.scopes.identityId : request.scopes.poolId; + return { + reserved: false, + denial: { reason: "spend-limit-exceeded", scope: ref.scope, scopeId: scopeId ?? "", limit, projected }, + }; } } - applyReserve(request.sendId, request.scopes, tokens, at); - append({ v: 1, kind: "reserve", sendId: request.sendId, scopes: request.scopes, tokens, at }); - return { reserved: true, sendId: request.sendId, tokens }; + + // Durability BEFORE admission. The record goes to disk first, and under a configured + // limit a failed write refuses the request rather than admitting one that a restart + // would forget -- which is exactly the disk-full and permission case durability is for. + const durable = append({ v: 1, kind: "reserve", send, targets: refs, tokens, at }); + if (!durable && enforced) { + return { reserved: false, denial: { reason: "reserve-not-durable", sendId: request.sendId } }; + } + applyReserve(send, refs, tokens, at); + compact(at); + return { reserved: true, sendId: request.sendId, tokens, durable }; }, - settle(sendId: string, usage: SpendUsage): boolean { - const reservation = reservations.get(sendId); + markDispatched(sendId: string): boolean { + const send = aliasFor("send", sendId); + const reservation = reservations.get(send); if (!reservation || reservation.status !== "open") return false; + const at = now(); + applyDispatch(send, at); + append({ v: 1, kind: "dispatch", send, at }); + return true; + }, + + abandon(sendId: string): boolean { + const send = aliasFor("send", sendId); + const reservation = reservations.get(send); + // Only an UNDISPATCHED reservation may be released for free. Once bytes have left for + // upstream the tokens may already be billed, so the caller owes settle or markLost. + if (!reservation || reservation.status !== "open") return false; + const at = now(); + applyResolve(send, "abandoned", 0, at); + append({ v: 1, kind: "abandon", send, at }); + return true; + }, + + settle(sendId: string, usage: SpendUsage): boolean { + const send = aliasFor("send", sendId); + const reservation = reservations.get(send); + if (!reservation || !isLive(reservation.status)) return false; const tokens = sanitizeTokens(usage.inputTokens) + sanitizeTokens(usage.outputTokens); - applySettle(sendId, tokens, now(), false); - append({ v: 1, kind: "settle", sendId, tokens, at: now() }); + const at = now(); + applyResolve(send, "settled", tokens, at); + append({ v: 1, kind: "settle", send, tokens, at }); return true; }, markLost(sendId: string): boolean { - const reservation = reservations.get(sendId); - if (!reservation || reservation.status !== "open") return false; - applySettle(sendId, 0, now(), true); - append({ v: 1, kind: "lost", sendId, at: now() }); + const send = aliasFor("send", sendId); + const reservation = reservations.get(send); + if (!reservation || !isLive(reservation.status)) return false; + const at = now(); + applyResolve(send, "lost", 0, at); + append({ v: 1, kind: "lost", send, at }); return true; }, + knows(sendId: string): boolean { + return reservations.has(aliasFor("send", sendId)); + }, + snapshot(scope: SpendScope, scopeId: string): ScopeSpendSnapshot | undefined { - const state = scopes.get(scopeKey(scope, scopeId)); + const state = scopes.get(scopeKey(scope, aliasFor(scope, scopeId))); if (!state) return undefined; return { settled: state.settled, @@ -323,25 +902,16 @@ export function createSpendReservationLedger(options: { }, exhausted(scope: SpendScope, scopeId: string): boolean { - const state = scopes.get(scopeKey(scope, scopeId)); + const state = scopes.get(scopeKey(scope, aliasFor(scope, scopeId))); return state !== undefined && isExhausted(scope, state); }, prune(at: number = now()): void { - const cutoff = at - policy.retentionMs; - for (const [key, state] of scopes) { - const scope = key.slice(0, key.indexOf("\0")) as SpendScope; - // Removal requires BOTH inactive and not exhausted inside the window. An - // exhausted-but-idle scope that was dropped would be recreated fresh under the - // same id -- the exact laundering the ceiling exists to stop. - if (state.reserved > 0 || state.lastSeenAt >= cutoff) continue; - if (isExhausted(scope, state)) continue; - scopes.delete(key); - } - for (const [sendId, reservation] of reservations) { - if (reservation.status === "open" || reservation.at >= cutoff) continue; - reservations.delete(sendId); - } + // Removal requires BOTH inactive and not exhausted inside the window. An + // exhausted-but-idle scope that was dropped would be recreated fresh under the + // same id -- the exact laundering the ceiling exists to stop. + evictSends(at, false); + evictScopes(at, false); }, }; } @@ -355,8 +925,10 @@ let sharedLedger: SpendReservationLedger | undefined; */ export function sharedSpendLedger(): SpendReservationLedger { if (!sharedLedger) { + const home = getConfigDir(); sharedLedger = createSpendReservationLedger({ - journal: createFileSpendJournal(join(getConfigDir(), SPEND_LEDGER_JOURNAL_FILENAME)), + journal: createFileSpendJournal(join(home, SPEND_LEDGER_JOURNAL_FILENAME)), + salt: loadOrCreateSpendLedgerSalt(join(home, SPEND_LEDGER_SALT_FILENAME)), }); } return sharedLedger; diff --git a/src/lib/workflow-budget.ts b/src/lib/workflow-budget.ts index 0aa0cb158b..8ca435352c 100644 --- a/src/lib/workflow-budget.ts +++ b/src/lib/workflow-budget.ts @@ -38,7 +38,12 @@ export interface WorkflowBudgetPolicy { * root still gets admitted; without this a worker burst starves the conversation it serves. */ readonly interactiveReserve: number; - /** Roots tracked at once. Bounded so a caller minting new ids cannot grow this forever. */ + /** + * Roots tracked at once, as a hard bound rather than a hint. At the ceiling one idle, + * under-limit root is evicted to make room; when no root may be forgotten safely the new + * root is REFUSED with `workflow-tracking-exhausted`. Admitting it anyway is what made a + * caller minting new ids able to grow this map past the number written here. + */ readonly maxTrackedRoots: number; } @@ -54,12 +59,28 @@ export type WorkflowDenial = | "workflow-concurrency-exhausted" | "workflow-sends-exhausted" | "workflow-children-exhausted" - | "workflow-spend-exhausted"; + | "workflow-spend-exhausted" + /** + * The root table is full and every entry is active or exhausted, so admitting this root + * would mean evicting one whose ceiling has already fired. Refusing is the honest answer: + * `maxTrackedRoots` is a bound, and inserting anyway made it a suggestion. + */ + | "workflow-tracking-exhausted" + /** This send id was already reserved once; a repeat buys no second dispatch. */ + | "workflow-send-replayed" + /** The reservation could not be made durable, and a configured ceiling requires it. */ + | "workflow-spend-undurable"; export type WorkflowLane = "interactive" | "worker"; export interface WorkflowAdmission { readonly rootId: string; + /** + * The request is about to leave for upstream. Call this at the dispatch boundary: until it + * runs, releasing the lease costs nothing, and after it a missing usage frame is booked as + * unresolved spend. + */ + markDispatched(): void; release(): void; } @@ -99,7 +120,14 @@ interface WorkflowState { const roots = new Map(); -function pruneOldestRoot(policy: WorkflowBudgetPolicy, spendLedger?: SpendReservationLedger): void { +/** + * Evict the oldest root that is safe to forget, and report whether one was found. + * + * The return value is the point. An earlier version returned void and the caller inserted + * the new root regardless, so `maxTrackedRoots` bounded nothing whenever every candidate + * was active or exhausted -- which is precisely the fan-out this file exists to bound. + */ +function evictOneRoot(policy: WorkflowBudgetPolicy, spendLedger?: SpendReservationLedger): boolean { let oldestKey: string | undefined; let oldestAt = Number.POSITIVE_INFINITY; for (const [key, state] of roots) { @@ -112,7 +140,9 @@ function pruneOldestRoot(policy: WorkflowBudgetPolicy, spendLedger?: SpendReserv if (spendLedger?.exhausted("root", key) === true) continue; if (state.lastSeenMs < oldestAt) { oldestAt = state.lastSeenMs; oldestKey = key; } } - if (oldestKey !== undefined) roots.delete(oldestKey); + if (oldestKey === undefined) return false; + roots.delete(oldestKey); + return true; } /** @@ -123,9 +153,11 @@ function pruneOldestRoot(policy: WorkflowBudgetPolicy, spendLedger?: SpendReserv * * When `spend` is given, admission also reserves its tokens on the spend ledger -- at the * root, identity and pool scopes at once -- before a concurrency slot is taken. A turn - * released without settlement moves its reservation to unresolved spend, because a send - * whose usage never arrived may still have been billed; releasing it would understate the - * scope. + * released without settlement is resolved by whether it was ever DISPATCHED: an undispatched + * turn gives its tokens back, and a dispatched one keeps them as unresolved spend, because a + * send whose usage never arrived may still have been billed. Call `lease.markDispatched()` + * at the point the request leaves for upstream; without it, admission followed by a local + * validation or routing failure would book spend that never happened. */ export function admitWorkflowTurn( rootId: string | undefined, @@ -142,7 +174,12 @@ export function admitWorkflowTurn( const ledger = spendLedger ?? (spend ? sharedSpendLedger() : undefined); let state = roots.get(rootId); if (!state) { - if (roots.size >= policy.maxTrackedRoots) pruneOldestRoot(policy, ledger); + if (roots.size >= policy.maxTrackedRoots && !evictOneRoot(policy, ledger)) { + // Nothing may be forgotten, so the new root is refused instead of admitted over the + // bound. The alternative -- evicting an exhausted root -- resets the ceiling that + // already fired, and a caller minting fresh ids would get unlimited budget from it. + return { admitted: false, reason: "workflow-tracking-exhausted", rootId }; + } state = { active: 0, sends: 0, children: new Set(), lastSeenMs: now }; roots.set(rootId, state); } @@ -171,11 +208,22 @@ export function admitWorkflowTurn( at: now, }); if (!decision.reserved) { + const denial = decision.denial; + // Every ledger refusal denies a DISPATCH. A duplicate send id and an undurable + // reservation are reported as themselves rather than folded into "exhausted", because + // an operator reading a 429 needs to know which of the three happened. + const reason: WorkflowDenial = denial.reason === "duplicate-send-id" + ? "workflow-send-replayed" + : denial.reason === "reserve-not-durable" || denial.reason === "journal-corrupt" + ? "workflow-spend-undurable" + : denial.reason === "tracking-capacity-exhausted" + ? "workflow-tracking-exhausted" + : "workflow-spend-exhausted"; return { admitted: false, - reason: "workflow-spend-exhausted", + reason, rootId, - spendScope: decision.denial.scope, + spendScope: denial.reason === "spend-limit-exceeded" ? denial.scope : undefined, }; } } @@ -187,6 +235,9 @@ export function admitWorkflowTurn( admitted: true, lease: { rootId, + markDispatched(): void { + if (spend && ledger) ledger.markDispatched(spend.sendId); + }, release(): void { if (released) return; released = true; @@ -195,10 +246,14 @@ export function admitWorkflowTurn( current.active = Math.max(0, current.active - 1); current.lastSeenMs = Date.now(); } - // A turn that ends without a settlement keeps its cost as unresolved spend rather - // than being released: the send may have been billed even though its usage frame - // never arrived. markLost is a no-op once settleWorkflowSpend already ran. - if (spend && ledger) ledger.markLost(spend.sendId); + // Which of the two applies depends on whether the send ever left this process. + // `abandon` succeeds only while the reservation is undispatched -- a turn refused by + // local validation or routing releases its tokens and books nothing, because + // inventing debt the account never incurred breaks the budget in the other + // direction. Once dispatched, abandon refuses and markLost keeps the cost as + // unresolved spend, since a send whose usage frame never arrived may still have been + // billed. Both are no-ops once settleWorkflowSpend already ran. + if (spend && ledger && !ledger.abandon(spend.sendId)) ledger.markLost(spend.sendId); }, }, }; @@ -230,6 +285,27 @@ export function settleWorkflowSpend( return (spendLedger ?? sharedSpendLedger()).settle(sendId, usage); } +/** + * Record that the send left for upstream. + * + * This is the line between "may be released for free" and "may have been billed". Admission + * alone is not dispatch: a turn can be admitted and then fail request validation, provider + * routing, or a local guard without a single byte reaching a model. Booking those as spend + * invents debt the account never incurred, so the reservation only becomes unresolvable + * after this call. + */ +export function dispatchWorkflowSpend(sendId: string, spendLedger?: SpendReservationLedger): boolean { + return (spendLedger ?? sharedSpendLedger()).markDispatched(sendId); +} + +/** + * Give a reservation back because the send never happened. Refused once dispatched, where + * settle or markLost is the only honest outcome. + */ +export function abandonWorkflowSpend(sendId: string, spendLedger?: SpendReservationLedger): boolean { + return (spendLedger ?? sharedSpendLedger()).abandon(sendId); +} + /** * Whether this root has already spent its whole physical-send ceiling. * diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 2b2722d86c..118089f2ae 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1099,6 +1099,7 @@ "skill-ocx.test.ts": "ci-workflows", "slug-codec.test.ts": "codex-integration", "sponsor-presets.test.ts": "providers", + "spend-ledger-file-journal.test.ts": "lib", "spend-reservation-ledger.test.ts": "lib", "sse-client-frame-bounds.test.ts": "responses", "sse-decoder.test.ts": "responses", diff --git a/tests/lib/spend-ledger-file-journal.test.ts b/tests/lib/spend-ledger-file-journal.test.ts new file mode 100644 index 0000000000..f2c10952fd --- /dev/null +++ b/tests/lib/spend-ledger-file-journal.test.ts @@ -0,0 +1,65 @@ +import { describe, expect, test } from "bun:test"; +import { chmodSync, mkdtempSync, readFileSync, readdirSync, statSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + createFileSpendJournal, + loadOrCreateSpendLedgerSalt, +} from "../../src/lib/spend-reservation-ledger"; + +/** POSIX mode bits do not describe a Windows ACL, where hardenSecretPath does the work. */ +const posixModes = process.platform !== "win32"; +const modeOf = (path: string): number => statSync(path).mode & 0o777; +const line = (send: string): string => JSON.stringify({ v: 1, kind: "lost", send, at: 1 }); + +describe("spend ledger file journal", () => { + test.skipIf(!posixModes)("a journal that already exists is re-hardened, not trusted", () => { + const dir = mkdtempSync(join(tmpdir(), "ocx-spend-journal-")); + const path = join(dir, "spend-ledger.jsonl"); + const journal = createFileSpendJournal(path); + + journal.append(line("alias-one")); + expect(modeOf(path)).toBe(0o600); + + // `mode` in a write option applies only when the file is CREATED. A journal left + // group-readable by an older build, a restored backup or a lax umask would keep that mode + // for its whole life, which is the gap this closes. + chmodSync(path, 0o644); + journal.append(line("alias-two")); + expect(modeOf(path)).toBe(0o600); + + chmodSync(path, 0o644); + expect(journal.read()).toHaveLength(2); + expect(modeOf(path)).toBe(0o600); + }); + + test("compaction replaces the journal atomically and leaves no temp behind", () => { + const dir = mkdtempSync(join(tmpdir(), "ocx-spend-compact-")); + const path = join(dir, "spend-ledger.jsonl"); + const journal = createFileSpendJournal(path); + journal.append(line("alias-one")); + journal.append(line("alias-two")); + + const rewrite = journal.rewrite; + expect(rewrite).toBeDefined(); + rewrite?.call(journal, [line("checkpoint-stand-in")]); + + expect(readFileSync(path, "utf8")).toBe(line("checkpoint-stand-in") + "\n"); + expect(journal.read()).toHaveLength(1); + // The temp file is renamed over the journal, never left in the home directory. + expect(readdirSync(dir)).toEqual(["spend-ledger.jsonl"]); + if (posixModes) expect(modeOf(path)).toBe(0o600); + }); + + test("the alias salt is minted once and reused, so replay still matches live requests", () => { + const dir = mkdtempSync(join(tmpdir(), "ocx-spend-salt-")); + const path = join(dir, "spend-ledger.salt"); + + const minted = loadOrCreateSpendLedgerSalt(path); + expect(minted).toMatch(/^[0-9a-f]{64}$/); + // Stability is the whole contract: a salt that changed per process would alias the same + // root id differently after a restart and hand every scope a fresh allowance. + expect(loadOrCreateSpendLedgerSalt(path)).toBe(minted); + if (posixModes) expect(modeOf(path)).toBe(0o600); + }); +}); diff --git a/tests/lib/spend-reservation-ledger.test.ts b/tests/lib/spend-reservation-ledger.test.ts index b2c613dbba..9440504bea 100644 --- a/tests/lib/spend-reservation-ledger.test.ts +++ b/tests/lib/spend-reservation-ledger.test.ts @@ -1,16 +1,28 @@ import { describe, expect, test } from "bun:test"; import { createSpendReservationLedger, + parseSpendJournalRecord, type SpendJournal, type SpendReservationPolicy, } from "../../src/lib/spend-reservation-ledger"; -/** In-memory journal: same replay contract as the file store, without touching disk. */ +/** In-memory journal: same replay and compaction contract as the file store, without disk. */ const memoryJournal = (): SpendJournal & { lines: string[] } => { const lines: string[] = []; - return { lines, read: () => [...lines], append: (line) => { lines.push(line); } }; + return { + lines, + read: () => [...lines], + append: (line) => { lines.push(line); }, + rewrite: (next) => { lines.length = 0; lines.push(...next); }, + }; }; +/** A journal that cannot persist: the disk-full and permission case durability exists for. */ +const unwritableJournal = (): SpendJournal => ({ + read: () => [], + append: () => { throw new Error("ENOSPC: no space left on device"); }, +}); + const policy = (maxTokens: number | undefined, retentionMs = 60_000): SpendReservationPolicy => ({ root: { maxTokens }, identity: { maxTokens }, @@ -151,5 +163,235 @@ describe("spend reservation ledger", () => { journal.lines.push("{not-json"); const second = createSpendReservationLedger({ journal, policy: policy(100), now: () => 2_000 }); expect(second.exhausted("root", "r1")).toBe(true); + // Quietly: the final record is the one that never finished being written, so nothing + // after it is missing and no total is understated. + expect(second.corruptRecords).toBe(0); + expect(second.degraded).toBe(false); + }); +}); + +describe("spend reservation ledger, send identity", () => { + test("a duplicate send id is refused instead of authorising a free dispatch", () => { + const journal = memoryJournal(); + const ledger = createSpendReservationLedger({ journal, policy: policy(1_000), now: () => 1_000 }); + const request = { sendId: "s1", scopes: { rootId: "r1" }, inputTokens: 10, outputCeilingTokens: 10 }; + expect(ledger.reserve(request).reserved).toBe(true); + + // The old behaviour returned success here while booking nothing, so one id bought an + // unlimited number of physical sends with the scope totals frozen. + const repeat = ledger.reserve(request); + expect(repeat.reserved).toBe(false); + if (!repeat.reserved) expect(repeat.denial.reason).toBe("duplicate-send-id"); + expect(ledger.snapshot("root", "r1")?.reserved).toBe(20); + + // Still refused once the original send resolves... + expect(ledger.settle("s1", { inputTokens: 10, outputTokens: 10 })).toBe(true); + expect(ledger.reserve(request).reserved).toBe(false); + expect(ledger.snapshot("root", "r1")?.settled).toBe(20); + + // ...and after a restart rebuilds the ledger from the journal. + const restarted = createSpendReservationLedger({ journal, policy: policy(1_000), now: () => 2_000 }); + expect(restarted.knows("s1")).toBe(true); + expect(restarted.reserve(request).reserved).toBe(false); + }); + + test("an undispatched reservation is released and only a dispatched one becomes unresolved", () => { + const ledger = createSpendReservationLedger({ policy: policy(1_000), now: () => 1_000 }); + ledger.reserve({ sendId: "never-sent", scopes: { rootId: "r1" }, inputTokens: 40, outputCeilingTokens: 10 }); + expect(ledger.abandon("never-sent")).toBe(true); + const released = ledger.snapshot("root", "r1"); + expect(released?.reserved).toBe(0); + expect(released?.unresolved).toBe(0); + expect(released?.settled).toBe(0); + // Abandoning is terminal, and the id stays known so it cannot be replayed. + expect(ledger.markLost("never-sent")).toBe(false); + expect(ledger.knows("never-sent")).toBe(true); + + ledger.reserve({ sendId: "sent", scopes: { rootId: "r1" }, inputTokens: 40, outputCeilingTokens: 10 }); + expect(ledger.markDispatched("sent")).toBe(true); + // Bytes left for upstream, so the tokens may already be billed and cannot be handed back. + expect(ledger.abandon("sent")).toBe(false); + expect(ledger.markLost("sent")).toBe(true); + expect(ledger.snapshot("root", "r1")?.unresolved).toBe(50); + }); +}); + +describe("spend reservation ledger, durability", () => { + test("under a configured limit a reservation that cannot be persisted is refused", () => { + const ledger = createSpendReservationLedger({ + journal: unwritableJournal(), policy: policy(1_000), now: () => 1_000, + }); + const denied = ledger.reserve({ + sendId: "s1", scopes: { rootId: "r1" }, inputTokens: 10, outputCeilingTokens: 10, + }); + // Admitting here would keep the request but forget it across a restart, which defeats the + // durable ceiling in exactly the disk-full and permission cases durability exists for. + expect(denied.reserved).toBe(false); + if (!denied.reserved) expect(denied.denial.reason).toBe("reserve-not-durable"); + // And it booked nothing: no scope was created and the id was not remembered. + expect(ledger.snapshot("root", "r1")).toBeUndefined(); + expect(ledger.knows("s1")).toBe(false); + expect(ledger.persistFailures).toBe(1); + expect(ledger.degraded).toBe(true); + }); + + test("observe-only mode still admits, and says the reservation is not durable", () => { + const ledger = createSpendReservationLedger({ + journal: unwritableJournal(), policy: policy(undefined), now: () => 1_000, + }); + const decision = ledger.reserve({ + sendId: "s1", scopes: { rootId: "r1" }, inputTokens: 10, outputCeilingTokens: 10, + }); + expect(decision.reserved).toBe(true); + if (decision.reserved) expect(decision.durable).toBe(false); + expect(ledger.degraded).toBe(true); + // An unconfigured install refuses nothing, so the accounting continues in memory. + expect(ledger.snapshot("root", "r1")?.reserved).toBe(20); + }); +}); + +describe("spend reservation ledger, journal validation", () => { + test("every malformed record shape is rejected rather than asserted into the replay", () => { + // Each of these used to be type-asserted straight into the rebuild: `null` crashed at + // record.v and the field-less reserve crashed inside applyReserve. + expect(parseSpendJournalRecord("null")).toBeUndefined(); + expect(parseSpendJournalRecord("[]")).toBeUndefined(); + expect(parseSpendJournalRecord("{not-json")).toBeUndefined(); + expect(parseSpendJournalRecord(JSON.stringify({ v: 1, kind: "reserve" }))).toBeUndefined(); + expect(parseSpendJournalRecord(JSON.stringify({ v: 2, kind: "lost", send: "a", at: 1 }))).toBeUndefined(); + expect(parseSpendJournalRecord(JSON.stringify({ v: 1, kind: "nope", send: "a", at: 1 }))).toBeUndefined(); + expect(parseSpendJournalRecord(JSON.stringify({ v: 1, kind: "lost", send: "a", at: -1 }))).toBeUndefined(); + expect(parseSpendJournalRecord(JSON.stringify({ v: 1, kind: "lost", send: "", at: 1 }))).toBeUndefined(); + expect(parseSpendJournalRecord(JSON.stringify({ + v: 1, kind: "reserve", send: "a", targets: [{ scope: "elsewhere", alias: "b" }], tokens: 1, at: 1, + }))).toBeUndefined(); + expect(parseSpendJournalRecord(JSON.stringify({ + v: 1, kind: "reserve", send: "a", targets: [{ scope: "root", alias: "b" }], tokens: Number.NaN, at: 1, + }))).toBeUndefined(); + expect(parseSpendJournalRecord(JSON.stringify({ v: 1, kind: "settle", send: "a", tokens: -5, at: 1 }))).toBeUndefined(); + expect(parseSpendJournalRecord(JSON.stringify({ v: 1, kind: "drop", scope: "root", at: 1 }))).toBeUndefined(); + // The one well-formed shape survives. + expect(parseSpendJournalRecord(JSON.stringify({ + v: 1, kind: "reserve", send: "a", targets: [{ scope: "root", alias: "b" }], tokens: 5, at: 7, + }))).toBeDefined(); + }); + + test("corruption in the middle of the journal fails accounting closed", () => { + const journal = memoryJournal(); + const first = createSpendReservationLedger({ journal, policy: policy(1_000), now: () => 1_000 }); + first.reserve({ sendId: "s1", scopes: { rootId: "r1" }, inputTokens: 10, outputCeilingTokens: 10 }); + first.settle("s1", { inputTokens: 10, outputTokens: 10 }); + // Records AFTER this one completed, so dropping it quietly would understate the root and + // hand back budget. Only a torn tail may be dropped. + journal.lines.splice(1, 0, "null"); + + const second = createSpendReservationLedger({ journal, policy: policy(1_000), now: () => 2_000 }); + expect(second.corruptRecords).toBe(1); + expect(second.degraded).toBe(true); + const denied = second.reserve({ + sendId: "s2", scopes: { rootId: "r1" }, inputTokens: 1, outputCeilingTokens: 0, + }); + expect(denied.reserved).toBe(false); + if (!denied.reserved) expect(denied.denial.reason).toBe("journal-corrupt"); + + // Observe-only accounting is not refused by it: there is no ceiling to enforce wrongly. + const observing = createSpendReservationLedger({ journal, policy: policy(undefined), now: () => 2_000 }); + expect(observing.reserve({ + sendId: "s3", scopes: { rootId: "r1" }, inputTokens: 1, outputCeilingTokens: 0, + }).reserved).toBe(true); + }); +}); + +describe("spend reservation ledger, bounded retention", () => { + const bounded = (overrides: Partial = {}): SpendReservationPolicy => ({ + root: {}, identity: {}, pool: {}, + retentionMs: 1_000, + maxTrackedScopes: 2, + maxTrackedSends: 2, + compactAfterRecords: 6, + ...overrides, + }); + + test("cleanup runs without a caller, and the journal does not resurrect what it removed", () => { + const journal = memoryJournal(); + let clock = 0; + const ledger = createSpendReservationLedger({ journal, policy: bounded(), now: () => clock }); + // Twelve unique root ids and twelve unique send ids, which is the shape that grew both + // Maps and the journal without bound when nothing called prune(). + for (let i = 0; i < 12; i += 1) { + clock = i * 10_000; + expect(ledger.reserve({ + sendId: `s${i}`, scopes: { rootId: `r${i}` }, inputTokens: 1, outputCeilingTokens: 0, + }).reserved).toBe(true); + expect(ledger.settle(`s${i}`, { inputTokens: 1, outputTokens: 0 })).toBe(true); + } + expect(ledger.snapshot("root", "r0")).toBeUndefined(); + expect(ledger.knows("s0")).toBe(false); + expect(ledger.snapshot("root", "r11")?.settled).toBe(1); + + // The tombstones and the checkpoint are what make that durable: a restart rebuilds the + // bounded state rather than every id the process ever saw. + const restarted = createSpendReservationLedger({ journal, policy: bounded(), now: () => clock }); + expect(restarted.snapshot("root", "r0")).toBeUndefined(); + expect(restarted.knows("s0")).toBe(false); + expect(restarted.snapshot("root", "r11")?.settled).toBe(1); + expect(restarted.corruptRecords).toBe(0); + // And the file itself stayed small instead of carrying two records per unique id. + expect(journal.lines.length).toBeLessThan(12); + }); + + test("a full tracking table refuses admission rather than forgetting an exhausted scope", () => { + let clock = 1_000; + const ledger = createSpendReservationLedger({ + policy: bounded({ root: { maxTokens: 100 }, maxTrackedSends: 64 }), + now: () => clock, + }); + for (const root of ["a", "b"]) { + expect(ledger.reserve({ + sendId: `s-${root}`, scopes: { rootId: root }, inputTokens: 100, outputCeilingTokens: 0, + }).reserved).toBe(true); + expect(ledger.settle(`s-${root}`, { inputTokens: 100, outputTokens: 0 })).toBe(true); + } + clock = 9_000; + // Both tracked scopes are spent, so there is no safe eviction candidate. Making room by + // dropping one would hand it a fresh allowance under the same id. + const denied = ledger.reserve({ + sendId: "s-c", scopes: { rootId: "c" }, inputTokens: 1, outputCeilingTokens: 0, + }); + expect(denied.reserved).toBe(false); + if (!denied.reserved) expect(denied.denial.reason).toBe("tracking-capacity-exhausted"); + expect(ledger.exhausted("root", "a")).toBe(true); + expect(ledger.exhausted("root", "b")).toBe(true); + expect(ledger.snapshot("root", "c")).toBeUndefined(); + }); +}); + +describe("spend reservation ledger, privacy", () => { + test("the journal stores salted aliases, never a root header, credential or pool id", () => { + const journal = memoryJournal(); + const scopes = { rootId: "thread_0123456789", identityId: "cred-jun@example.com", poolId: "pool-prod" }; + const ledger = createSpendReservationLedger({ + journal, policy: policy(1_000), now: () => 1_000, salt: "install-one", + }); + ledger.reserve({ sendId: "send-abc", scopes, inputTokens: 10, outputCeilingTokens: 0 }); + ledger.markDispatched("send-abc"); + ledger.settle("send-abc", { inputTokens: 10, outputTokens: 0 }); + + const written = journal.lines.join("\n"); + for (const raw of ["send-abc", "thread_0123456789", "cred-jun@example.com", "pool-prod"]) { + expect(written).not.toContain(raw); + } + + // The alias is stable for one install, so a restart still finds the same spend... + const restarted = createSpendReservationLedger({ + journal, policy: policy(1_000), now: () => 2_000, salt: "install-one", + }); + expect(restarted.snapshot("root", "thread_0123456789")?.settled).toBe(10); + expect(restarted.snapshot("identity", "cred-jun@example.com")?.settled).toBe(10); + // ...and unrecoverable with anything but that install's salt. + const stranger = createSpendReservationLedger({ + journal, policy: policy(1_000), now: () => 2_000, salt: "install-two", + }); + expect(stranger.snapshot("root", "thread_0123456789")).toBeUndefined(); }); }); diff --git a/tests/lib/workflow-budget.test.ts b/tests/lib/workflow-budget.test.ts index 16f711ff70..f21ecc1ef9 100644 --- a/tests/lib/workflow-budget.test.ts +++ b/tests/lib/workflow-budget.test.ts @@ -77,19 +77,32 @@ describe("workflow count caps", () => { if (denied && !denied.admitted) expect(denied.reason).toBe("workflow-children-exhausted"); }); - test("an exhausted-but-idle root is never evicted to make room", () => { - // Fill the root to its send ceiling, then let it go idle: only the new - // exhausted-but-idle rule can still protect it from eviction. + test("a full root table refuses a new root instead of evicting an exhausted one", () => { + // Fill one root to its send ceiling and let it go idle, then take the only other slot + // with an active root. maxTrackedRoots is 2, so the table is now full and neither entry + // may be forgotten. const filled = admitWorkflowTurn("full", "interactive", smallPolicy); chargeWorkflowSends("full", 3); if (filled?.admitted) filled.lease.release(); - // Two more roots arrive, forcing eviction pressure at maxTrackedRoots = 2. - admitWorkflowTurn("n1", "interactive", smallPolicy); - admitWorkflowTurn("n2", "interactive", smallPolicy); - // The exhausted root survived the prune: recreating it must not reset its allowance. + const busy = admitWorkflowTurn("n1", "interactive", smallPolicy); + expect(busy?.admitted).toBe(true); + + // Inserting a third root anyway is what made maxTrackedRoots a suggestion: the bound has + // to refuse, because the only other way to honour it is to reset a ceiling that fired. + const refused = admitWorkflowTurn("n2", "interactive", smallPolicy); + expect(refused?.admitted).toBe(false); + if (refused && !refused.admitted) expect(refused.reason).toBe("workflow-tracking-exhausted"); + expect(workflowBudgetSnapshot("n2")).toBeUndefined(); + + // The exhausted root survived, so recreating it does not reset its allowance. const decision = admitWorkflowTurn("full", "interactive", smallPolicy); expect(decision?.admitted).toBe(false); if (decision && !decision.admitted) expect(decision.reason).toBe("workflow-sends-exhausted"); + + // Once the active root goes idle it becomes a safe candidate and the next root fits. + if (busy?.admitted) busy.lease.release(); + expect(admitWorkflowTurn("n2", "interactive", smallPolicy)?.admitted).toBe(true); + expect(workflowBudgetSnapshot("n1")).toBeUndefined(); }); }); @@ -123,11 +136,12 @@ describe("workflow spend reservation", () => { if (denied && !denied.admitted) expect(denied.spendScope).toBe("identity"); }); - test("settlement is idempotent and a release without it becomes unresolved spend", () => { + test("settlement is idempotent and a dispatched release without it becomes unresolved spend", () => { const ledger = createSpendReservationLedger({ journal: memoryJournal(), policy: spendPolicy(1_000), now: () => 1_000 }); const admitted = admitWorkflowTurn("r1", "interactive", DEFAULT_WORKFLOW_BUDGET_POLICY, undefined, 1_000, { sendId: "s1", inputTokens: 100, outputCeilingTokens: 50 }, ledger); expect(admitted?.admitted).toBe(true); + if (admitted?.admitted) admitted.lease.markDispatched(); expect(settleWorkflowSpend("s1", { inputTokens: 90, outputTokens: 10 }, ledger)).toBe(true); // Double settlement books nothing. expect(settleWorkflowSpend("s1", { inputTokens: 90, outputTokens: 10 }, ledger)).toBe(false); @@ -136,26 +150,59 @@ describe("workflow spend reservation", () => { expect(settled?.settled).toBe(100); expect(settled?.unresolved).toBe(0); - // A turn released without settlement keeps its cost as unresolved spend. + // A DISPATCHED turn released without settlement keeps its cost as unresolved spend: the + // send may have been billed even though its usage frame never arrived. const lost = admitWorkflowTurn("r1", "interactive", DEFAULT_WORKFLOW_BUDGET_POLICY, undefined, 2_000, { sendId: "s2", inputTokens: 30, outputCeilingTokens: 20 }, ledger); - if (lost?.admitted) lost.lease.release(); + if (lost?.admitted) { + lost.lease.markDispatched(); + lost.lease.release(); + } const after = ledger.snapshot("root", "r1"); expect(after?.unresolved).toBe(50); // And a late settle for the lost send is correctly refused. expect(settleWorkflowSpend("s2", { inputTokens: 30, outputTokens: 20 }, ledger)).toBe(false); }); + test("a turn that never reached upstream books no spend at all", () => { + const ledger = createSpendReservationLedger({ journal: memoryJournal(), policy: spendPolicy(1_000), now: () => 1_000 }); + const admitted = admitWorkflowTurn("r1", "interactive", DEFAULT_WORKFLOW_BUDGET_POLICY, + undefined, 1_000, { sendId: "never-sent", inputTokens: 100, outputCeilingTokens: 50 }, ledger); + expect(admitted?.admitted).toBe(true); + // Admission is not dispatch. A local validation or routing failure between the two used + // to be booked as unresolved spend, which invents debt the account never incurred. + if (admitted?.admitted) admitted.lease.release(); + const snapshot = ledger.snapshot("root", "r1"); + expect(snapshot?.reserved).toBe(0); + expect(snapshot?.unresolved).toBe(0); + expect(snapshot?.settled).toBe(0); + + // The send id stays known, so replaying it buys no second dispatch. + const replay = admitWorkflowTurn("r1", "interactive", DEFAULT_WORKFLOW_BUDGET_POLICY, + undefined, 1_000, { sendId: "never-sent", inputTokens: 100, outputCeilingTokens: 50 }, ledger); + expect(replay?.admitted).toBe(false); + if (replay && !replay.admitted) expect(replay.reason).toBe("workflow-send-replayed"); + }); + test("a spend-exhausted idle root survives eviction pressure", () => { const ledger = createSpendReservationLedger({ journal: memoryJournal(), policy: spendPolicy(100), now: () => 1_000 }); const exhausted = admitWorkflowTurn("full", "interactive", smallPolicy, undefined, 1_000, { sendId: "s1", inputTokens: 60, outputCeilingTokens: 40 }, ledger); - // The lease is released so the root is idle, but its spend is exhausted. - if (exhausted?.admitted) exhausted.lease.release(); - admitWorkflowTurn("n1", "interactive", smallPolicy, undefined, 2_000, undefined, ledger); - admitWorkflowTurn("n2", "interactive", smallPolicy, undefined, 3_000, undefined, ledger); - const snap = workflowBudgetSnapshot("full"); - expect(snap).toBeDefined(); + // The send is dispatched and settled, then the lease is released: the root is idle and + // its spend is exhausted. + expect(exhausted?.admitted).toBe(true); + if (exhausted?.admitted) { + exhausted.lease.markDispatched(); + expect(settleWorkflowSpend("s1", { inputTokens: 60, outputTokens: 40 }, ledger)).toBe(true); + exhausted.lease.release(); + } + // An idle, unspent root takes the other slot, then a third root arrives under + // maxTrackedRoots = 2. The evictable one is the unspent root, never the exhausted one. + const spare = admitWorkflowTurn("n1", "interactive", smallPolicy, undefined, 2_000, undefined, ledger); + if (spare?.admitted) spare.lease.release(); + expect(admitWorkflowTurn("n2", "interactive", smallPolicy, undefined, 3_000, undefined, ledger)?.admitted).toBe(true); + expect(workflowBudgetSnapshot("n1")).toBeUndefined(); + expect(workflowBudgetSnapshot("full")).toBeDefined(); const denied = admitWorkflowTurn("full", "interactive", smallPolicy, undefined, 4_000, { sendId: "s2", inputTokens: 1, outputCeilingTokens: 0 }, ledger); expect(denied?.admitted).toBe(false);