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

@@ -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

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
 }

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)
     })
   })
 }

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 })])

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<string, unknown>
   time?: number
   surfaceOp?: TimelineEvent['surfaceOp']
 }): TimelineEvent {

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 {

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.