From 5fdf8f69082586d8ebd09c7cb6753dcac5faee72 Mon Sep 17 00:00:00 2001 From: Bowen Liang Date: Fri, 4 Sep 2026 00:19:10 +0800 Subject: [PATCH] fix(host): fold provider usage through a deep billed-count read (#44) A nonconforming provider figure (fractional token counts, or a gateway reporting cached_tokens above prompt_tokens driving the disjoint uncached input below zero) rode the state raw and failed the unit's strict `.int().nonnegative()` wire/state gates on EVERY later delivery - the registry drive throws per event, so the contextTimeline push feed dies and every unit registered after it starves: the Context tab froze mid-session while the agent kept running (issue #44: stats stuck at round 26 step 5 of 97, cost 3.49 of ~30 CNY). - fold: every usage bucket passes a deep billed-count read - finite numbers and numeric strings accepted, fractions rounded, negatives clamped to 0, non-finite/garbage read as absent; the request record and the session-cost totals stay complete instead of freezing - semantics change: any readable bucket is now a billing sample (official token-meter parity) - an output-only usage object bills prompt 0 with its output split, where the previous shallow guard dropped the sample whole - stateVersion 13: cached rows refold from the durable log, which recovers the steps the frozen feed never showed (cost/timing are running sums and were never trimmed); upgrade and rollback both discard mismatched rows and refold clean Closes #44 --- src/host/fold.ts | 91 ++++++++++++++++++++++-------- src/host/timeline.ts | 9 ++- tests/host/compat/registry.spec.ts | 70 ++++++++++++++++++++++- tests/host/fold-events.spec.ts | 86 +++++++++++++++++++++++++++- tests/host/helpers/events.ts | 3 +- tests/host/helpers/projection.ts | 8 ++- tests/host/timeline.spec.ts | 2 +- 7 files changed, 238 insertions(+), 31 deletions(-) diff --git a/src/host/fold.ts b/src/host/fold.ts index da8ad7c..dc31597 100644 --- a/src/host/fold.ts +++ b/src/host/fold.ts @@ -428,11 +428,42 @@ function applySurface( return node } +/** The durable usage object, as far as the fold reads it — every bucket is re-proved by `tokenCountOf`, never trusted. */ interface UsageLike { - inputTokens?: number - cacheReadTokens?: number - cacheWriteTokens?: number - outputTokens?: number + inputTokens?: unknown + cacheReadTokens?: unknown + cacheWriteTokens?: unknown + outputTokens?: unknown +} + +/** One usage object's buckets, deeply normalized to billed counts (see {@link tokenCountOf}). */ +interface BilledUsage { + input: number + cacheRead: number + cacheWrite: number + output: number +} + +/** + * One provider-reported usage bucket as a billed count, or null when the + * field carries no readable number. Accepts finite numbers and numeric + * strings; fractions round (some gateways report fractional counts) and + * negatives clamp to 0 — a mis-accounting gateway that reports + * `cached_tokens > prompt_tokens` drives the disjoint uncached-input figure + * below zero, and one raw figure in the state would fail the wire and state + * schemas' `.int().nonnegative()` gates on EVERY later delivery, permanently + * freezing the projection feed for the session (issue #44). NaN, infinities, + * and non-numeric values read as absent. + */ +function tokenCountOf(value: unknown): number | null { + if (typeof value === 'number') { + return Number.isFinite(value) ? Math.max(0, Math.round(value)) : null + } + if (typeof value === 'string' && value.trim() !== '') { + const parsed = Number(value) + return Number.isFinite(parsed) ? Math.max(0, Math.round(parsed)) : null + } + return null } /** @@ -466,9 +497,11 @@ function isPeakUtc(time: number): boolean { /** * Fold one billed request into the session-cost totals, cloning along the * mutated path only (the untouched branch stays shared with the persisted - * previous state — the apply contract never mutates it in place). + * previous state — the apply contract never mutates it in place). The + * buckets arrive sanitized ({@link BilledUsage}), so the totals stay at the + * schemas' non-negative safe integers no matter what the provider reported. */ -function accumulateCost(st: TimelineState, time: number, usage: UsageLike): void { +function accumulateCost(st: TimelineState, time: number, usage: BilledUsage): void { const family = costFamilyOf(st.model) if (family === null) return const prev: SessionCostUsage = st.cost ?? {} @@ -477,13 +510,10 @@ function accumulateCost(st: TimelineState, time: number, usage: UsageLike): void const b = fam[period] ?? { uncached: 0, cacheRead: 0, cacheWrite: 0, output: 0 } const nextFam: CostFamilyUsage = { ...fam } nextFam[period] = { - /* v8 ignore next 1 -- accumulateCost only runs under the - `typeof usage.inputTokens === 'number'` guard; the fallback is - defensive against a hand-built UsageLike. */ - uncached: b.uncached + (usage.inputTokens ?? 0), - cacheRead: b.cacheRead + (usage.cacheReadTokens ?? 0), - cacheWrite: b.cacheWrite + (usage.cacheWriteTokens ?? 0), - output: b.output + (usage.outputTokens ?? 0), + uncached: b.uncached + usage.input, + cacheRead: b.cacheRead + usage.cacheRead, + cacheWrite: b.cacheWrite + usage.cacheWrite, + output: b.output + usage.output, } const next: SessionCostUsage = { ...prev } next[family] = nextFam @@ -738,7 +768,7 @@ export function applyTimeline(state: TimelineState, event: TimelineEvent, bounds case 'assistant/message': { // Snapshot the request exactly as dispatched: current surface + header, // before this response joins the surface. - const usage = data?.usage as UsageLike | undefined + const usage = data?.usage as UsageLike | null | undefined const s = ensure() const total = s.systemTokens + s.toolsTokens + s.sums.user + s.sums.inject + s.sums.assistant + s.sums.tool const record: RequestRecord = { @@ -755,18 +785,35 @@ export function applyTimeline(state: TimelineState, event: TimelineEvent, bounds // materialize an `undefined` property (plain-JSON precondition, the trap that broke the projection cache here). if (data && typeof data.turn === 'number') record.turn = data.turn if (data && typeof data.step === 'number') record.step = data.step - if (usage && typeof usage.inputTokens === 'number') { + if (usage !== null && typeof usage === 'object') { // Official TokenUsage semantics (dsh-llm): the buckets are disjoint — // inputTokens is uncached input only, cache read/write are separate, // and billed prompt-side = input + cacheRead + cacheWrite. outputTokens // already includes reasoningTokens. No separate prompt/output field - // exists in the durable vocabulary. - record.prompt = usage.inputTokens + (usage.cacheReadTokens || 0) + (usage.cacheWriteTokens || 0) - // Cache-hit share of the billed prompt (the step line's 缓存 figure): - // keep the cache-served half of `prompt`; absent = no cache buckets. - if (typeof usage.cacheReadTokens === 'number') record.cacheRead = usage.cacheReadTokens - if (typeof usage.outputTokens === 'number') record.output = usage.outputTokens - accumulateCost(s, event.time, usage) + // exists in the durable vocabulary. Every bucket passes the deep + // `tokenCountOf` read first: the durable log is untrusted input, and a + // raw nonconforming figure must never enter the state (issue #44). + const input = tokenCountOf(usage.inputTokens) + const cacheRead = tokenCountOf(usage.cacheReadTokens) + const cacheWrite = tokenCountOf(usage.cacheWriteTokens) + const output = tokenCountOf(usage.outputTokens) + // Any readable bucket is a billing sample (the official meter folds + // every reported usage object; an output-only sample bills prompt 0 + // there too). A fully unreadable object is treated as absent, so a + // fabricated 0 never reaches the client's derived-occupancy anchor. + if (input !== null || cacheRead !== null || cacheWrite !== null || output !== null) { + record.prompt = (input ?? 0) + (cacheRead ?? 0) + (cacheWrite ?? 0) + // Cache-hit share of the billed prompt (the step line's 缓存 figure): + // keep the cache-served half of `prompt`; absent = no cache bucket. + if (cacheRead !== null) record.cacheRead = cacheRead + if (output !== null) record.output = output + accumulateCost(s, event.time, { + input: input ?? 0, + cacheRead: cacheRead ?? 0, + cacheWrite: cacheWrite ?? 0, + output: output ?? 0, + }) + } } s.requests.push(record) // Timing: one completed model call; its wait/generation split prices diff --git a/src/host/timeline.ts b/src/host/timeline.ts index 05bcc2c..1d98d40 100644 --- a/src/host/timeline.ts +++ b/src/host/timeline.ts @@ -218,7 +218,14 @@ export function createContextTimelineDefinition(config: Config): ProjectionDefin // first token) + `genMs` (first token → message) — `lmMs` left the shape, // and the `stepStart` slot gained the `firstToken` stamp; cached rows // refolded from the log, which rebuilds the split. - stateVersion: 12, + // 13: provider-reported usage buckets are sanitized (deep-read, rounded, + // clamped non-negative — see fold.ts `tokenCountOf`) before they enter + // request records and the session-cost totals. A nonconforming raw figure + // (fractional counts, a gateway reporting cached_tokens > prompt_tokens) + // previously rode the state verbatim and failed the wire/state schemas' + // integer gates on EVERY later delivery, permanently freezing the + // session's projection feed (issue #44); cached rows refold clean. + stateVersion: 13, } return definition } diff --git a/tests/host/compat/registry.spec.ts b/tests/host/compat/registry.spec.ts index 6308b16..e4ac855 100644 --- a/tests/host/compat/registry.spec.ts +++ b/tests/host/compat/registry.spec.ts @@ -117,7 +117,7 @@ for (const [index, baseline] of BASELINES.entries()) { assert.ok(rows !== undefined, 'checkpoint rows are losslessly JSON-serializable') for (const key of ['contextTimeline', 'contextHeaders']) { const row = rows[key] as { ver: number; seq: number } - assert.equal(row.ver, key === 'contextTimeline' ? 12 : 1) + assert.equal(row.ver, key === 'contextTimeline' ? 13 : 1) assert.equal(row.seq, 11) } // And the write-gate equivalent on every intermediate state of a fresh fold. @@ -153,6 +153,72 @@ for (const [index, baseline] of BASELINES.entries()) { assert.ok(viewed.contextHeaders !== undefined) }) + test('issue #44 recovery: the version bump discards the poisoned row and the refold re-tracks every step', () => { + // The reporter's situation: a pre-fix checkpoint row (stateVersion 12) + // carries a raw nonconforming gateway figure, while the durable log + // holds every committed step. The bump makes the stale row unusable, so + // the cold read refolds from init over the FULL log — the steps the + // frozen feed never showed come back, billed with sanitized figures and + // passing the real wire schema on the way out. + const log = [ + ...fullLog(), + assistantMessage(12, { turn: 1, step: 2, usage: { inputTokens: -80, cacheReadTokens: 150.7, outputTokens: 22.4 } }), + assistantMessage(13, { turn: 1, step: 3, usage: { inputTokens: NaN, outputTokens: null } }), + assistantMessage(14, { turn: 1, step: 4, usage: { inputTokens: 12 } }), + ] + const stale: Checkpoint = { + contextTimeline: { + ver: 12, + seq: 14, + val: { + // The poisoned v12 shape: the newest record carries the raw + // gateway figure that failed the integer gates on every delivery. + surface: [], + sums: { user: 0, inject: 0, assistant: 0, tool: 0 }, + systemTokens: 0, + toolsTokens: 0, + requests: [{ seq: 9, time: 0, system: 0, tools: 0, user: 0, inject: 0, assistant: 0, tool: 0, total: 0, prompt: -80, cacheRead: 150, output: 22 }], + events: [], + archived: [], + callNames: {}, + }, + }, + } + // The cache-read rung skips the stale-version row instead of serving it. + const { driver, session } = bootSession(index, log) + assert.equal(driver.viewCheckpoint(stale).contextTimeline, undefined) + // The cold read refolds from init over the full durable log; a schema + // violation anywhere would throw right here (restore parses each wire + // value through the unit's real viewSchema). + const restored = driver.restore(stale, log, 0, session.header) + const timeline = restored.values.contextTimeline as { requests: { seq: number; prompt?: number; cacheRead?: number; output?: number }[] } + // Every step is tracked again — the hostile ones included… + assert.deepEqual(timeline.requests.map(r => r.seq), [5, 9, 12, 13, 14]) + // …each billed from its sanitized buckets (no figure dropped, none raw). + assert.deepEqual( + timeline.requests.filter(r => r.seq >= 12).map(r => [r.prompt, r.cacheRead, r.output]), + [[151, 151, 22], [undefined, undefined, undefined], [12, undefined, undefined]], + ) + }) + + test('issue #44 rollback: a pre-bump plugin discards the v13 rows and refolds clean, never crashes', () => { + // The reverse upgrade direction: v13-written rows read by a plugin + // still pinned at v12. The state SHAPE never changed (only value + // normalization), so the version mismatch just discards the rows and + // refolds from the log — the same graceful ladder as every bump. + const log = fullLog() + const { driver: current, session } = bootSession(index, log) + const rows = current.checkpoint(session) + const rolledBack = new RegistryDriver(baseline) + rolledBack.register({ ...createContextTimelineDefinition({}), stateVersion: 12 }) + rolledBack.register(createContextHeadersDefinition()) + assert.equal(rolledBack.viewCheckpoint(rows).contextTimeline, undefined) + const restored = rolledBack.restore(rows, log, 0, session.header) + const timeline = restored.values.contextTimeline as { current: { total: number }; requests: unknown[] } + assert.ok(timeline.current.total > 0) + assert.equal(timeline.requests.length, 2) + }) + test('the change gate: an uninteresting event keeps its state reference and notifies nothing', () => { const { driver, session } = bootSession(index) const seen: string[] = [] @@ -191,7 +257,7 @@ for (const [index, baseline] of BASELINES.entries()) { driver.register(def) // Same key at the SAME stateVersion shares the registration (preset ref-counting). assert.doesNotThrow(() => driver.register(createContextTimelineDefinition({}))) - assert.throws(() => driver.register({ ...def, stateVersion: 13 }), RegistryViolationError) + assert.throws(() => driver.register({ ...def, stateVersion: 14 }), RegistryViolationError) }) }) } diff --git a/tests/host/fold-events.spec.ts b/tests/host/fold-events.spec.ts index badbf2e..6ddf0a0 100644 --- a/tests/host/fold-events.spec.ts +++ b/tests/host/fold-events.spec.ts @@ -322,10 +322,11 @@ describe('assistant/message request records', () => { assert.equal(state.cost, undefined) }) - test('usage without inputTokens is not a billing sample', () => { + test('output-only usage is a billing sample: prompt 0 with the output split', () => { const { state } = driveTimeline([assistantMessage(1, { usage: { outputTokens: 5 } })]) - assert.ok(!('prompt' in state.requests[0])) - assert.ok(!('output' in state.requests[0])) + assert.equal(state.requests[0].prompt, 0) + assert.equal(state.requests[0].output, 5) + assert.ok(!('cacheRead' in state.requests[0])) assert.equal(state.cost, undefined) }) @@ -356,6 +357,85 @@ describe('assistant/message request records', () => { }) }) +describe('hostile provider usage (issue #44: stats must survive nonconforming figures)', () => { + // The registry parses every served view (and every cold-restored state) + // against the unit's strict integer schemas with no containment: one raw + // nonconforming figure in the state fails the gate on EVERY later delivery + // and permanently freezes the session's projection feed. The fold therefore + // sanitizes each bucket before it touches the state, and the records stay + // complete (billed, not dropped). + test('fractional buckets round to integers', () => { + const { state } = driveTimeline([ + assistantMessage(1, { usage: { inputTokens: 10.4, cacheReadTokens: 20.6, outputTokens: 3.5 } }), + ]) + const rec = state.requests[0] + assert.equal(rec.prompt, 31) + assert.equal(rec.cacheRead, 21) + assert.equal(rec.output, 4) + }) + + test('a gateway reporting cached_tokens above prompt_tokens clamps to billed zeros', () => { + const { state } = driveTimeline([ + assistantMessage(1, { usage: { inputTokens: -80, cacheReadTokens: 150, outputTokens: 22 } }), + ]) + const rec = state.requests[0] + assert.equal(rec.prompt, 150) + assert.equal(rec.cacheRead, 150) + assert.equal(rec.output, 22) + }) + + test('string-reported buckets are read', () => { + const { state } = driveTimeline([ + assistantMessage(1, { usage: { inputTokens: '100', cacheWriteTokens: ' 7 ' } }), + ]) + const rec = state.requests[0] + assert.equal(rec.prompt, 107) + assert.ok(!('cacheRead' in rec)) + assert.ok(!('output' in rec)) + }) + + test('non-finite and garbage buckets read as absent', () => { + const { state } = driveTimeline([ + assistantMessage(1, { usage: { inputTokens: Number.NaN, outputTokens: null, cacheReadTokens: {}, cacheWriteTokens: ' ' } }), + ]) + assert.ok(!('prompt' in state.requests[0]), 'no readable bucket → the sample is absent, not a bogus zero') + assert.equal(state.cost, undefined) + }) + + test('a readable minority still bills: garbage input with a valid output', () => { + const { state } = driveTimeline([ + assistantMessage(1, { usage: { inputTokens: 'n/a', outputTokens: 9 } }), + ]) + const rec = state.requests[0] + assert.equal(rec.prompt, 0) + assert.equal(rec.output, 9) + }) + + test('the folded state and served view stay schema-valid under hostile usage', () => { + const drive = driveTimeline([ + header(1, { model: 'deepseek-v4-flash' }), + assistantMessage(2, { turn: 1, step: 1, usage: { inputTokens: -80, cacheReadTokens: 150.7, outputTokens: '22.2' } }), + assistantMessage(3, { turn: 1, step: 2, usage: { inputTokens: Number.NaN, outputTokens: null } }), + assistantMessage(4, { turn: 1, step: 3, usage: { inputTokens: 12.4 } }), + ]) + // The gates the registry itself parses with: a throw here is exactly the + // issue #44 freeze (the drive loop and every later delivery die on it). + drive.def.stateSchema.parse(assertPlainJson(drive.state)) + drive.def.wire.viewSchema.parse(drive.def.wire.view(drive.state)) + }) + + test('sanitized buckets accumulate into the session-cost totals', () => { + const { state } = driveTimeline([ + header(1, { model: 'deepseek-v4-flash' }), + assistantMessage(2, { + usage: { inputTokens: -100, cacheReadTokens: 300.4, cacheWriteTokens: '10', outputTokens: 0.2 }, + time: Date.UTC(2024, 0, 1, 22, 0, 0), // 06:00 Beijing Monday — off-peak + }), + ]) + assert.deepEqual(state.cost?.flash?.off, { uncached: 0, cacheRead: 300, cacheWrite: 10, output: 0 }) + }) +}) + describe('plan/mode', () => { test('active true records plan.on', () => { const { state } = driveTimeline([planMode(1, { active: true })]) diff --git a/tests/host/helpers/events.ts b/tests/host/helpers/events.ts index e424524..a163cdb 100644 --- a/tests/host/helpers/events.ts +++ b/tests/host/helpers/events.ts @@ -92,7 +92,8 @@ export function assistantMessage(seq: number, opts: { turn?: number step?: number content?: ContentBlock[] - usage?: { inputTokens?: number; outputTokens?: number; cacheReadTokens?: number; cacheWriteTokens?: number } + /** Widened: hostile fixtures ride the same field (the fold re-proves every bucket). */ + usage?: Record time?: number surfaceOp?: TimelineEvent['surfaceOp'] }): TimelineEvent { diff --git a/tests/host/helpers/projection.ts b/tests/host/helpers/projection.ts index d820e43..546492a 100644 --- a/tests/host/helpers/projection.ts +++ b/tests/host/helpers/projection.ts @@ -22,9 +22,15 @@ import { createContextHeadersDefinition } from '../../../src/host/headers' */ export interface TimelineDefLike { key: string + /** The persisted-state gate the projection cache's cold path parses rows with. */ + stateSchema: { parse(value: unknown): unknown } init(): TimelineState apply(state: TimelineState, event: TimelineEvent): TimelineState - wire: { view(state: TimelineState): ContextTimeline } + wire: { + /** The wire gate every delivery channel parses the served value with. */ + viewSchema: { parse(value: unknown): unknown } + view(state: TimelineState): ContextTimeline + } } export interface HeadersDefLike { diff --git a/tests/host/timeline.spec.ts b/tests/host/timeline.spec.ts index 51bcc52..d274414 100644 --- a/tests/host/timeline.spec.ts +++ b/tests/host/timeline.spec.ts @@ -49,7 +49,7 @@ describe('createContextTimelineDefinition', () => { test('carries the supported projection contract on one unit', () => { const def = createContextTimelineDefinition({}) assert.equal(def.key, 'contextTimeline') - assert.equal(def.stateVersion, 12) + assert.equal(def.stateVersion, 13) assert.equal(typeof def.init, 'function') assert.equal(typeof def.apply, 'function') // The supported registry contract: stateSchema + a REQUIRED wire block.