diff options
Diffstat (limited to 'src/core/metrics/reducer.ts')
| -rw-r--r-- | src/core/metrics/reducer.ts | 450 |
1 files changed, 266 insertions, 184 deletions
diff --git a/src/core/metrics/reducer.ts b/src/core/metrics/reducer.ts index 1e66cc8..39fc5ee 100644 --- a/src/core/metrics/reducer.ts +++ b/src/core/metrics/reducer.ts @@ -2,127 +2,172 @@ import type { AgentEvent, StepId, StepMetrics, TurnMetrics, Usage } from "@dispa import type { BuildingStep, LiveTurn, MetricsState, TurnMetricsEntry } from "./types"; function sumStepUsages(steps: readonly BuildingStep[]): Usage { - let inputTokens = 0; - let outputTokens = 0; - let hasCacheRead = false; - let hasCacheWrite = false; - let cacheReadTokens = 0; - let cacheWriteTokens = 0; + let inputTokens = 0; + let outputTokens = 0; + let hasCacheRead = false; + let hasCacheWrite = false; + let cacheReadTokens = 0; + let cacheWriteTokens = 0; - for (const step of steps) { - if (step.usage === undefined) continue; - inputTokens += step.usage.inputTokens; - outputTokens += step.usage.outputTokens; - if (step.usage.cacheReadTokens !== undefined && step.usage.cacheReadTokens > 0) { - hasCacheRead = true; - cacheReadTokens += step.usage.cacheReadTokens; - } - if (step.usage.cacheWriteTokens !== undefined && step.usage.cacheWriteTokens > 0) { - hasCacheWrite = true; - cacheWriteTokens += step.usage.cacheWriteTokens; - } - } + for (const step of steps) { + if (step.usage === undefined) continue; + inputTokens += step.usage.inputTokens; + outputTokens += step.usage.outputTokens; + if (step.usage.cacheReadTokens !== undefined && step.usage.cacheReadTokens > 0) { + hasCacheRead = true; + cacheReadTokens += step.usage.cacheReadTokens; + } + if (step.usage.cacheWriteTokens !== undefined && step.usage.cacheWriteTokens > 0) { + hasCacheWrite = true; + cacheWriteTokens += step.usage.cacheWriteTokens; + } + } - const base: Usage = { inputTokens, outputTokens }; - if (hasCacheRead) { - (base as { cacheReadTokens?: number }).cacheReadTokens = cacheReadTokens; - } - if (hasCacheWrite) { - (base as { cacheWriteTokens?: number }).cacheWriteTokens = cacheWriteTokens; - } - return base; + const base: Usage = { inputTokens, outputTokens }; + if (hasCacheRead) { + (base as { cacheReadTokens?: number }).cacheReadTokens = cacheReadTokens; + } + if (hasCacheWrite) { + (base as { cacheWriteTokens?: number }).cacheWriteTokens = cacheWriteTokens; + } + return base; } function buildingStepToMetrics(bs: BuildingStep): StepMetrics { - const usage: Usage = bs.usage ?? { inputTokens: 0, outputTokens: 0 }; - const base: StepMetrics = { stepId: bs.stepId as StepId, usage }; - if (bs.ttftMs !== undefined) { - (base as { ttftMs?: number }).ttftMs = bs.ttftMs; - } - if (bs.decodeMs !== undefined) { - (base as { decodeMs?: number }).decodeMs = bs.decodeMs; - } - if (bs.genTotalMs !== undefined) { - (base as { genTotalMs?: number }).genTotalMs = bs.genTotalMs; - } - return base; + const usage: Usage = bs.usage ?? { inputTokens: 0, outputTokens: 0 }; + const base: StepMetrics = { stepId: bs.stepId as StepId, usage }; + if (bs.ttftMs !== undefined) { + (base as { ttftMs?: number }).ttftMs = bs.ttftMs; + } + if (bs.decodeMs !== undefined) { + (base as { decodeMs?: number }).decodeMs = bs.decodeMs; + } + if (bs.genTotalMs !== undefined) { + (base as { genTotalMs?: number }).genTotalMs = bs.genTotalMs; + } + return base; } function getStep(lt: LiveTurn, id: string): BuildingStep { - const step = lt.stepMap.get(id); - if (step === undefined) throw new Error(`Missing step ${id} in live turn`); - return step; + const step = lt.stepMap.get(id); + if (step === undefined) throw new Error(`Missing step ${id} in live turn`); + return step; } function liveTurnToMetrics(lt: LiveTurn): TurnMetrics { - const buildingSteps = lt.stepOrder.map((id) => getStep(lt, id)); - const steps = buildingSteps.map((bs) => buildingStepToMetrics(bs)); - const usage = lt.doneUsage ?? sumStepUsages(buildingSteps); - const base: TurnMetrics = { turnId: lt.turnId, usage, steps }; - if (lt.durationMs !== undefined) { - (base as { durationMs?: number }).durationMs = lt.durationMs; - } - if (lt.doneContextSize !== undefined) { - (base as { contextSize?: number }).contextSize = lt.doneContextSize; - } - return base; + const buildingSteps = lt.stepOrder.map((id) => getStep(lt, id)); + const steps = buildingSteps.map((bs) => buildingStepToMetrics(bs)); + const usage = lt.doneUsage ?? sumStepUsages(buildingSteps); + const base: TurnMetrics = { turnId: lt.turnId, usage, steps }; + if (lt.durationMs !== undefined) { + (base as { durationMs?: number }).durationMs = lt.durationMs; + } + if (lt.doneContextSize !== undefined) { + (base as { contextSize?: number }).contextSize = lt.doneContextSize; + } + return base; +} + +/** + * A step's contribution to the live context size: `inputTokens + outputTokens`, + * or `undefined` when the step has no usage yet OR its counters are not safe to + * sum (non-finite / negative — defensive: a corrupt provider report must never + * reach the status bar as NaN/Infinity). Cache tokens are deliberately NOT + * included: `cacheReadTokens` / `cacheWriteTokens` are a SUBSET of + * `inputTokens`, so adding them would double-count. + */ +function stepContextSize(usage: Usage | undefined): number | undefined { + if (usage === undefined) return undefined; + const { inputTokens, outputTokens } = usage; + if (!Number.isFinite(inputTokens) || !Number.isFinite(outputTokens)) return undefined; + if (inputTokens < 0 || outputTokens < 0) return undefined; + return inputTokens + outputTokens; +} + +/** + * The context size an IN-FLIGHT (not-done) turn occupies right now — for + * progressive display DURING a turn (before it seals), so the indicator updates + * after each step instead of waiting for `done`. + * + * CONTRACT: only call this on a turn whose `done` event has NOT arrived (the + * caller, `selectCurrentContextSize`, reaches it solely for entries with + * `total === null`, i.e. `lt.done === false`). Finalized turns use their + * authoritative `contextSize` instead; `doneContextSize` is read on the + * `total` path, never here. + * + * Returns the most recent step WITH USABLE USAGE's `inputTokens + outputTokens` + * (scanning newest → oldest by first-seen step order): each step's input + * already includes all prior context (the prompt is re-prefilled every step), so + * the last step's input+output is the true occupancy — the same definition + * `TurnDoneEvent.contextSize` stamps at turn end. A just-reported step's usage + * wins immediately, even mid-stream. Steps with no usage or unsafe usage are + * skipped, falling back to the next older usable step. `undefined` when no step + * has reported usable usage yet. + */ +function liveTurnContextSize(lt: LiveTurn): number | undefined { + for (let i = lt.stepOrder.length - 1; i >= 0; i--) { + const step = lt.stepMap.get(lt.stepOrder[i] ?? ""); + const ctx = stepContextSize(step?.usage); + if (ctx !== undefined) return ctx; + } + return undefined; } function ensureLiveTurn(state: MetricsState, turnId: string): [MetricsState, LiveTurn] { - const existing = state.live.get(turnId); - if (existing !== undefined) return [state, existing]; + const existing = state.live.get(turnId); + if (existing !== undefined) return [state, existing]; - const newTurn: LiveTurn = { - turnId, - done: false, - durationMs: undefined, - doneUsage: undefined, - doneContextSize: undefined, - stepMap: new Map(), - stepOrder: [], - }; - const newLive = new Map(state.live); - newLive.set(turnId, newTurn); - return [{ ...state, live: newLive, liveOrder: [...state.liveOrder, turnId] }, newTurn]; + const newTurn: LiveTurn = { + turnId, + done: false, + durationMs: undefined, + doneUsage: undefined, + doneContextSize: undefined, + stepMap: new Map(), + stepOrder: [], + }; + const newLive = new Map(state.live); + newLive.set(turnId, newTurn); + return [{ ...state, live: newLive, liveOrder: [...state.liveOrder, turnId] }, newTurn]; } function upsertStep(lt: LiveTurn, stepId: string, update: Partial<BuildingStep>): LiveTurn { - const existing = lt.stepMap.get(stepId); - if (existing !== undefined) { - const merged: BuildingStep = { - stepId, - usage: update.usage ?? existing.usage, - ttftMs: update.ttftMs ?? existing.ttftMs, - decodeMs: update.decodeMs ?? existing.decodeMs, - genTotalMs: update.genTotalMs ?? existing.genTotalMs, - complete: update.complete ?? existing.complete, - }; - const newMap = new Map(lt.stepMap); - newMap.set(stepId, merged); - return { ...lt, stepMap: newMap }; - } + const existing = lt.stepMap.get(stepId); + if (existing !== undefined) { + const merged: BuildingStep = { + stepId, + usage: update.usage ?? existing.usage, + ttftMs: update.ttftMs ?? existing.ttftMs, + decodeMs: update.decodeMs ?? existing.decodeMs, + genTotalMs: update.genTotalMs ?? existing.genTotalMs, + complete: update.complete ?? existing.complete, + }; + const newMap = new Map(lt.stepMap); + newMap.set(stepId, merged); + return { ...lt, stepMap: newMap }; + } - const fresh: BuildingStep = { - stepId, - usage: update.usage, - ttftMs: update.ttftMs, - decodeMs: update.decodeMs, - genTotalMs: update.genTotalMs, - complete: update.complete ?? false, - }; - const newMap = new Map(lt.stepMap); - newMap.set(stepId, fresh); - return { ...lt, stepMap: newMap, stepOrder: [...lt.stepOrder, stepId] }; + const fresh: BuildingStep = { + stepId, + usage: update.usage, + ttftMs: update.ttftMs, + decodeMs: update.decodeMs, + genTotalMs: update.genTotalMs, + complete: update.complete ?? false, + }; + const newMap = new Map(lt.stepMap); + newMap.set(stepId, fresh); + return { ...lt, stepMap: newMap, stepOrder: [...lt.stepOrder, stepId] }; } /** The initial empty metrics state. */ export function initialMetricsState(): MetricsState { - return { - live: new Map(), - liveOrder: [], - durable: new Map(), - durableOrder: [], - }; + return { + live: new Map(), + liveOrder: [], + durable: new Map(), + durableOrder: [], + }; } /** @@ -135,69 +180,88 @@ export function initialMetricsState(): MetricsState { * - All other event types: return state unchanged. */ export function foldMetricsEvent(state: MetricsState, event: AgentEvent): MetricsState { - switch (event.type) { - case "usage": { - if (event.stepId === undefined) return state; - const [s1, lt] = ensureLiveTurn(state, event.turnId); - const updated = upsertStep(lt, event.stepId, { usage: event.usage }); - const newLive = new Map(s1.live); - newLive.set(event.turnId, updated); - return { ...s1, live: newLive }; - } + switch (event.type) { + case "usage": { + if (event.stepId === undefined) return state; + const [s1, lt] = ensureLiveTurn(state, event.turnId); + const updated = upsertStep(lt, event.stepId, { usage: event.usage }); + const newLive = new Map(s1.live); + newLive.set(event.turnId, updated); + return { ...s1, live: newLive }; + } - case "step-complete": { - const [s1, lt] = ensureLiveTurn(state, event.turnId); - const updated = upsertStep(lt, event.stepId, { - ttftMs: event.ttftMs, - decodeMs: event.decodeMs, - genTotalMs: event.genTotalMs, - complete: true, - }); - const newLive = new Map(s1.live); - newLive.set(event.turnId, updated); - return { ...s1, live: newLive }; - } + case "step-complete": { + const [s1, lt] = ensureLiveTurn(state, event.turnId); + const updated = upsertStep(lt, event.stepId, { + ttftMs: event.ttftMs, + decodeMs: event.decodeMs, + genTotalMs: event.genTotalMs, + complete: true, + }); + const newLive = new Map(s1.live); + newLive.set(event.turnId, updated); + return { ...s1, live: newLive }; + } - case "done": { - const [s1, lt] = ensureLiveTurn(state, event.turnId); - const updated: LiveTurn = { - ...lt, - done: true, - durationMs: event.durationMs ?? lt.durationMs, - doneUsage: event.usage ?? lt.doneUsage, - doneContextSize: event.contextSize ?? lt.doneContextSize, - }; - const newLive = new Map(s1.live); - newLive.set(event.turnId, updated); - return { ...s1, live: newLive }; - } + case "done": { + const [s1, lt] = ensureLiveTurn(state, event.turnId); + const updated: LiveTurn = { + ...lt, + done: true, + durationMs: event.durationMs ?? lt.durationMs, + doneUsage: event.usage ?? lt.doneUsage, + doneContextSize: event.contextSize ?? lt.doneContextSize, + }; + const newLive = new Map(s1.live); + newLive.set(event.turnId, updated); + return { ...s1, live: newLive }; + } - default: - return state; - } + default: + return state; + } } /** * Store durable (sealed) metrics from the backend. These win over live data * for any shared `turnId`. + * + * Once durable (authoritative) data covers a turn, its live (in-memory) copy + * is REDUNDANT and is pruned from `state.live` / `liveOrder` so the live map + * doesn't grow unbounded over a long conversation. There is no display gap: + * the durable entry replaces the live one atomically in the same fold, and + * `selectOrderedTurnMetrics` / `selectCurrentContextSize` read durable for it. */ export function applyDurableMetrics( - state: MetricsState, - turns: readonly TurnMetrics[], + state: MetricsState, + turns: readonly TurnMetrics[], ): MetricsState { - const newDurable = new Map(state.durable); - const newDurableOrder = [...state.durableOrder]; - for (const turn of turns) { - if (!newDurable.has(turn.turnId)) { - newDurableOrder.push(turn.turnId); - } - newDurable.set(turn.turnId, turn); - } - return { - ...state, - durable: newDurable, - durableOrder: newDurableOrder, - }; + const newDurable = new Map(state.durable); + const newDurableOrder = [...state.durableOrder]; + const prunedIds = new Set<string>(); + for (const turn of turns) { + if (!newDurable.has(turn.turnId)) { + newDurableOrder.push(turn.turnId); + } + newDurable.set(turn.turnId, turn); + if (state.live.has(turn.turnId)) prunedIds.add(turn.turnId); + } + + if (prunedIds.size === 0) { + return { ...state, durable: newDurable, durableOrder: newDurableOrder }; + } + + const newLive = new Map(state.live); + for (const id of prunedIds) newLive.delete(id); + const newLiveOrder = state.liveOrder.filter((id) => !prunedIds.has(id)); + + return { + ...state, + live: newLive, + liveOrder: newLiveOrder, + durable: newDurable, + durableOrder: newDurableOrder, + }; } /** @@ -210,54 +274,72 @@ export function applyDurableMetrics( * Live turns with no completed steps and not done are omitted. */ export function selectOrderedTurnMetrics(state: MetricsState): readonly TurnMetricsEntry[] { - const result: TurnMetricsEntry[] = []; - const seen = new Set<string>(); + const result: TurnMetricsEntry[] = []; + const seen = new Set<string>(); - for (const turnId of state.durableOrder) { - const tm = state.durable.get(turnId); - if (tm !== undefined) { - result.push({ turnId, steps: tm.steps, total: tm }); - seen.add(turnId); - } - } + for (const turnId of state.durableOrder) { + const tm = state.durable.get(turnId); + if (tm !== undefined) { + result.push({ turnId, steps: tm.steps, total: tm }); + seen.add(turnId); + } + } - for (const turnId of state.liveOrder) { - if (seen.has(turnId)) continue; - const lt = state.live.get(turnId); - if (lt === undefined) continue; + for (const turnId of state.liveOrder) { + if (seen.has(turnId)) continue; + const lt = state.live.get(turnId); + if (lt === undefined) continue; - const completeSteps = lt.stepOrder - .map((id) => lt.stepMap.get(id)) - .filter((s): s is BuildingStep => s?.complete === true) - .map((s) => buildingStepToMetrics(s)); + const completeSteps = lt.stepOrder + .map((id) => lt.stepMap.get(id)) + .filter((s): s is BuildingStep => s?.complete === true) + .map((s) => buildingStepToMetrics(s)); - if (completeSteps.length === 0 && !lt.done) continue; + if (completeSteps.length === 0 && !lt.done) continue; - result.push({ - turnId, - steps: completeSteps, - total: lt.done ? liveTurnToMetrics(lt) : null, - }); - } + result.push({ + turnId, + steps: completeSteps, + total: lt.done ? liveTurnToMetrics(lt) : null, + }); + } - return result; + return result; } /** * Select the conversation's CURRENT context size — the tokens it occupies right * now. Per the wire contract a client reads the LATEST turn's `contextSize`; we * scan the merged ordered turns NEWEST → OLDEST and return the first DEFINED - * `contextSize` (a finalized turn whose provider reported per-step usage). + * value. + * + * For a FINALIZED turn (`done` event or durable data) we use its authoritative + * `contextSize`. For an IN-FLIGHT (not-done) turn we compute it PROGRESSIVELY + * from the most recent step WITH USAGE — its `inputTokens + outputTokens` is the + * current occupancy (mirroring `TurnDoneEvent.contextSize`'s definition) — so + * the indicator updates after each step completes instead of waiting for the + * turn to seal. An in-flight turn with no step usage yet is skipped, falling + * back to the next older finalized turn. * - * Returns `undefined` ("unknown") when no finalized turn carries a context size — - * the caller renders a placeholder, NEVER `0`. Durable (sealed) data wins over + * Returns `undefined` ("unknown") when no turn carries a context size — the + * caller renders a placeholder, NEVER `0`. Durable (sealed) data wins over * live for a shared `turnId` (it is the persisted, authoritative value). */ export function selectCurrentContextSize(state: MetricsState): number | undefined { - const ordered = selectOrderedTurnMetrics(state); - for (let i = ordered.length - 1; i >= 0; i--) { - const total = ordered[i]?.total; - if (total?.contextSize !== undefined) return total.contextSize; - } - return undefined; + const ordered = selectOrderedTurnMetrics(state); + for (let i = ordered.length - 1; i >= 0; i--) { + const entry = ordered[i]; + if (entry === undefined) continue; + if (entry.total !== null) { + if (entry.total.contextSize !== undefined) return entry.total.contextSize; + continue; + } + // In-flight turn: progressive context size from the latest step with usage. + const lt = state.live.get(entry.turnId); + if (lt !== undefined) { + const live = liveTurnContextSize(lt); + if (live !== undefined) return live; + } + } + return undefined; } |