diff options
Diffstat (limited to 'src/core/metrics')
| -rw-r--r-- | src/core/metrics/cache-timer.test.ts | 359 | ||||
| -rw-r--r-- | src/core/metrics/cache-timer.ts | 286 | ||||
| -rw-r--r-- | src/core/metrics/format.test.ts | 92 | ||||
| -rw-r--r-- | src/core/metrics/format.ts | 85 | ||||
| -rw-r--r-- | src/core/metrics/index.ts | 15 | ||||
| -rw-r--r-- | src/core/metrics/place.test.ts | 143 | ||||
| -rw-r--r-- | src/core/metrics/place.ts | 460 | ||||
| -rw-r--r-- | src/core/metrics/placement-scenarios.test.ts | 376 | ||||
| -rw-r--r-- | src/core/metrics/types.ts | 14 |
9 files changed, 1448 insertions, 382 deletions
diff --git a/src/core/metrics/cache-timer.test.ts b/src/core/metrics/cache-timer.test.ts new file mode 100644 index 0000000..d450135 --- /dev/null +++ b/src/core/metrics/cache-timer.test.ts @@ -0,0 +1,359 @@ +import type { AgentEvent, StepId, TurnMetrics } from "@dispatch/wire"; +import { describe, expect, it } from "vitest"; +import type { RenderGroup } from "../chunks"; +import { + CACHE_TIMER_MAX_MS, + CACHE_TTL_MS, + type CacheTimerState, + foldCacheTimerEvent, + initialCacheTimerState, + lastTurnRowIndex, + seedTurnGap, + stopTurnGap, + viewCacheTimer, +} from "./cache-timer"; +import type { MetricsRow } from "./types"; + +const SEC = 1000; +const MIN = 60 * SEC; +const C = "conv"; + +const ev = { + turnStart: (turnId: string): AgentEvent => ({ type: "turn-start", conversationId: C, turnId }), + userMessage: (turnId: string): AgentEvent => ({ + type: "user-message", + conversationId: C, + turnId, + text: "hi", + }), + text: (turnId: string): AgentEvent => ({ + type: "text-delta", + conversationId: C, + turnId, + delta: "x", + }), + reasoningEnd: (turnId: string): AgentEvent => ({ + type: "reasoning-end", + conversationId: C, + turnId, + }), + call: (turnId: string, stepId: string, toolCallId: string): AgentEvent => ({ + type: "tool-call", + conversationId: C, + turnId, + stepId: stepId as StepId, + toolCallId, + toolName: "t", + input: {}, + }), + result: ( + turnId: string, + stepId: string, + toolCallId: string, + durationMs?: number, + ): AgentEvent => ({ + type: "tool-result", + conversationId: C, + turnId, + stepId: stepId as StepId, + toolCallId, + toolName: "t", + content: "ok", + isError: false, + ...(durationMs !== undefined ? { durationMs } : {}), + }), + stepComplete: (turnId: string, stepId: string): AgentEvent => ({ + type: "step-complete", + conversationId: C, + turnId, + stepId: stepId as StepId, + }), + usage: (turnId: string, stepId: string): AgentEvent => ({ + type: "usage", + conversationId: C, + turnId, + stepId: stepId as StepId, + usage: { inputTokens: 1, outputTokens: 1 }, + }), + done: (turnId: string): AgentEvent => ({ + type: "done", + conversationId: C, + turnId, + reason: "end-turn", + }), + error: (turnId: string): AgentEvent => ({ + type: "error", + conversationId: C, + turnId, + message: "boom", + }), +}; + +const noTurn = () => undefined; + +/** Fold `[time, event]` pairs from the initial state. */ +function run( + steps: readonly (readonly [number, AgentEvent])[], + start: CacheTimerState = initialCacheTimerState(), + lastTurnId: () => string | undefined = noTurn, +): CacheTimerState { + return steps.reduce((s, [at, e]) => foldCacheTimerEvent(s, e, at, lastTurnId), start); +} + +describe("foldCacheTimerEvent — step gaps", () => { + it("runs from step-complete while tools execute, and stamps when the last result returns", () => { + const s1 = run([ + [0, ev.turnStart("t1")], + [100, ev.call("t1", "t1#0", "a")], + [110, ev.call("t1", "t1#0", "b")], + [1_000, ev.stepComplete("t1", "t1#0")], + ]); + expect(s1.running).toMatchObject({ + kind: "step", + stepId: "t1#0", + startedAt: 1_000, + toolCount: 2, + }); + + const s2 = run([[4_000, ev.result("t1", "t1#0", "a")]], s1); + expect(s2.running?.kind).toBe("step"); // still waiting on "b" + expect(s2.stepStamps.size).toBe(0); + + const s3 = run([[13_000, ev.result("t1", "t1#0", "b")]], s2); + expect(s3.running).toBeNull(); + expect(s3.stepStamps.get("t1#0")).toBe(12_000); + }); + + it("stamps each step of a multi-step turn separately", () => { + const s = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + [3_000, ev.result("t1", "t1#0", "a")], + [5_000, ev.call("t1", "t1#1", "b")], + [6_000, ev.stepComplete("t1", "t1#1")], + [66_000, ev.result("t1", "t1#1", "b")], + ]); + expect(s.stepStamps.get("t1#0")).toBe(2_000); + expect(s.stepStamps.get("t1#1")).toBe(60_000); + }); + + it("uses the backend tool duration when it exceeds the wall clock (late-join replay burst)", () => { + const s = run([ + [5_000, ev.call("t1", "t1#0", "a")], + [5_000, ev.stepComplete("t1", "t1#0")], + [5_000, ev.result("t1", "t1#0", "a", 42_000)], + ]); + expect(s.stepStamps.get("t1#0")).toBe(42_000); + }); + + it("stops at the next step's output when a result never matched (fallback)", () => { + const s = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + [9_000, ev.text("t1")], + ]); + expect(s.running).toBeNull(); + expect(s.stepStamps.get("t1#0")).toBe(8_000); + }); + + it("treats a reasoning-end boundary as resumed output too", () => { + const s = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + [9_000, ev.reasoningEnd("t1")], + ]); + expect(s.running).toBeNull(); + expect(s.stepStamps.get("t1#0")).toBe(8_000); + }); + + it("stops on a tool-call or usage from a DIFFERENT step, not its own", () => { + const base = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + ]); + expect(run([[2_000, ev.usage("t1", "t1#0")]], base).running?.kind).toBe("step"); + expect(run([[3_000, ev.usage("t1", "t1#1")]], base).stepStamps.get("t1#0")).toBe(2_000); + expect(run([[4_000, ev.call("t1", "t1#1", "z")]], base).stepStamps.get("t1#0")).toBe(3_000); + }); + + it("discards a step with no tool calls (the final step of a turn)", () => { + const s = run([ + [1_000, ev.stepComplete("t1", "t1#0")], + [1_050, ev.done("t1")], + ]); + expect(s.stepStamps.size).toBe(0); + expect(s.running).toEqual({ kind: "turn", turnId: "t1", startedAt: 1_050 }); + }); + + it("stamps a step whose tools were still running when the turn ended (stopped)", () => { + const s = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + [21_000, ev.done("t1")], + ]); + expect(s.stepStamps.get("t1#0")).toBe(20_000); + expect(s.running?.kind).toBe("turn"); + }); +}); + +describe("foldCacheTimerEvent — turn gaps", () => { + it("starts at done and is stamped on that turn when the next turn starts", () => { + const s = run([ + [1_000, ev.done("t1")], + [91_000, ev.userMessage("t2")], + [91_500, ev.turnStart("t2")], + ]); + expect(s.running).toBeNull(); + expect(s.turnStamps.get("t1")).toBe(90_000); + }); + + it("an error also ends the turn", () => { + expect(run([[7, ev.error("t1")]]).running).toEqual({ + kind: "turn", + turnId: "t1", + startedAt: 7, + }); + }); + + it("stopTurnGap stamps on a local send", () => { + const s = stopTurnGap(run([[1_000, ev.done("t1")]]), 4_000, noTurn); + expect(s.running).toBeNull(); + expect(s.turnStamps.get("t1")).toBe(3_000); + }); + + it("stopTurnGap is a no-op while a step gap runs", () => { + const base = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + ]); + expect(stopTurnGap(base, 2_000, noTurn)).toBe(base); + }); +}); + +describe("seedTurnGap", () => { + it("seeds a turn gap with no turnId, stamped on the most recent turn when stopped", () => { + const seeded = seedTurnGap(initialCacheTimerState(), 1_000); + expect(seeded.running).toEqual({ kind: "turn", turnId: null, startedAt: 1_000 }); + const s = stopTurnGap(seeded, 61_000, () => "t9"); + expect(s.turnStamps.get("t9")).toBe(60_000); + }); + + it("drops a seeded gap when the most recent turn is unknown", () => { + const s = stopTurnGap(seedTurnGap(initialCacheTimerState(), 1_000), 2_000, noTurn); + expect(s.running).toBeNull(); + expect(s.turnStamps.size).toBe(0); + }); + + it("keeps the later start, never overrides a step gap, ignores non-finite", () => { + const live = run([[5_000, ev.done("t1")]]); + expect(seedTurnGap(live, 4_000).running).toEqual({ + kind: "turn", + turnId: "t1", + startedAt: 5_000, + }); + expect(seedTurnGap(live, 6_000).running).toEqual({ + kind: "turn", + turnId: "t1", + startedAt: 6_000, + }); + const step = run([ + [0, ev.call("t1", "t1#0", "a")], + [1_000, ev.stepComplete("t1", "t1#0")], + ]); + expect(seedTurnGap(step, 9_000)).toBe(step); + expect(seedTurnGap(initialCacheTimerState(), Number.NaN).running).toBeNull(); + }); +}); + +describe("viewCacheTimer", () => { + it("formats m:ss up to 10 minutes", () => { + expect(viewCacheTimer(0).label).toBe("0:00"); + expect(viewCacheTimer(9 * SEC + 999).label).toBe("0:09"); + expect(viewCacheTimer(MIN + 5 * SEC).label).toBe("1:05"); + expect(viewCacheTimer(5 * MIN + 30 * SEC).label).toBe("5:30"); + expect(viewCacheTimer(9 * MIN + 59 * SEC).label).toBe("9:59"); + expect(viewCacheTimer(9 * MIN + 59 * SEC).capped).toBe(false); + }); + + it("reads '10+ min' and is capped at and past 10 minutes", () => { + expect(viewCacheTimer(CACHE_TIMER_MAX_MS)).toEqual({ + label: "10+ min", + level: "error", + capped: true, + }); + expect(viewCacheTimer(3 * 60 * MIN).label).toBe("10+ min"); + }); + + it("colours by the 5-minute cache TTL", () => { + expect(viewCacheTimer(3 * MIN + 59 * SEC).level).toBe("success"); + expect(viewCacheTimer(4 * MIN).level).toBe("warning"); + expect(viewCacheTimer(CACHE_TTL_MS).level).toBe("error"); + expect(viewCacheTimer(7 * MIN).level).toBe("error"); + }); + + it("clamps negative / non-finite elapsed (clock skew) to 0:00", () => { + expect(viewCacheTimer(-5 * SEC).label).toBe("0:00"); + expect(viewCacheTimer(Number.NaN).label).toBe("0:00"); + }); +}); + +function userRow(text = "hi"): MetricsRow { + const group: RenderGroup = { + kind: "single", + chunk: { seq: null, role: "user", chunk: { type: "text", text }, provisional: true }, + }; + return { kind: "group", group }; +} + +function assistantRow(turnId: string): MetricsRow { + const group: RenderGroup = { + kind: "single", + chunk: { + seq: 1, + role: "assistant", + chunk: { type: "text", text: "ok" }, + provisional: false, + turnId, + }, + }; + return { kind: "group", group }; +} + +function turnRow(turnId: string, turnNumber: number): MetricsRow { + const usage = { inputTokens: 10, outputTokens: 5 }; + const turn: TurnMetrics = { turnId, usage, steps: [{ stepId: `${turnId}#0` as StepId, usage }] }; + return { + kind: "turn-metrics", + turn, + turnNumber, + cumulativeUsage: usage, + prevLastStepUsage: null, + }; +} + +describe("lastTurnRowIndex", () => { + it("picks the LAST turn-metrics row", () => { + const rows = [ + userRow(), + assistantRow("t1"), + turnRow("t1", 1), + userRow(), + assistantRow("t2"), + turnRow("t2", 2), + ]; + expect(lastTurnRowIndex(rows)).toBe(5); + }); + + it("is null once a user message follows the last turn", () => { + expect(lastTurnRowIndex([userRow(), turnRow("t1", 1), userRow("next")])).toBeNull(); + }); + + it("skips trailing non-user rows after the turn row", () => { + expect(lastTurnRowIndex([userRow(), turnRow("t1", 1), assistantRow("t1")])).toBe(1); + }); + + it("is null with no turn-metrics rows", () => { + expect(lastTurnRowIndex([])).toBeNull(); + expect(lastTurnRowIndex([userRow(), assistantRow("t1")])).toBeNull(); + }); +}); diff --git a/src/core/metrics/cache-timer.ts b/src/core/metrics/cache-timer.ts new file mode 100644 index 0000000..3d7270d --- /dev/null +++ b/src/core/metrics/cache-timer.ts @@ -0,0 +1,286 @@ +import type { AgentEvent } from "@dispatch/wire"; +import type { MetricsRow } from "./types"; + +/** + * Pure core for the **cache timer** — how long the prompt cache has sat idle + * between model requests. Anthropic's prompt cache (as used by the Claude Code + * subscription) expires 5 minutes after its last hit, so the gaps BETWEEN + * requests decide whether the next request reads from cache: + * + * - **step gap**: a step (one LLM round-trip) finished and its tools are + * executing. Starts at `step-complete`; stops when the step's last tool + * result arrives (the next request fires right after) or, failing that, when + * anything from the next step shows up. Its final value is **stamped** on + * that step (keyed by `stepId`). + * - **turn gap**: the turn ended (`done` / `error`) and the conversation is + * waiting on the user. Stops when a new message is sent (locally, or a + * `user-message` / `turn-start` from another client) and is stamped on that + * turn (keyed by `turnId`). + * + * At most ONE gap runs at a time (`running`). Zero DOM, zero clock: every fold + * takes the current time as an argument. + */ + +/** Prompt-cache time-to-live: 5 minutes (drives the badge colour). */ +export const CACHE_TTL_MS = 5 * 60 * 1000; + +/** The timer counts up to this, then reads "10+ min". */ +export const CACHE_TIMER_MAX_MS = 10 * 60 * 1000; + +/** Within this much of the TTL the timer warns (colour only). */ +const WARN_AT_MS = CACHE_TTL_MS - 60 * 1000; + +/** A step gap in progress: the step's tools are executing. */ +export interface RunningStepGap { + readonly kind: "step"; + readonly turnId: string; + readonly stepId: string; + /** Epoch ms the step completed. */ + readonly startedAt: number; + /** How many tool calls the step made (`0` → nothing to show; never stamped). */ + readonly toolCount: number; + /** Tool calls still awaiting their result. */ + readonly pending: ReadonlySet<string>; + /** Longest backend-reported tool `durationMs` seen so far for this step. */ + readonly maxToolMs: number; +} + +/** A turn gap in progress: the turn ended and nothing has been sent since. */ +export interface RunningTurnGap { + readonly kind: "turn"; + /** + * The turn that ended, or `null` when seeded from the backend's + * `lastActivityAt` after a reload (no live `done` seen) — then it belongs to + * the most recent turn, resolved positionally / via `lastTurnId` on stop. + */ + readonly turnId: string | null; + /** Epoch ms the turn ended. */ + readonly startedAt: number; +} + +export type RunningGap = RunningStepGap | RunningTurnGap; + +export interface CacheTimerState { + readonly running: RunningGap | null; + /** Stamped step gaps: `stepId` → elapsed ms. */ + readonly stepStamps: ReadonlyMap<string, number>; + /** Stamped turn gaps: `turnId` → elapsed ms. */ + readonly turnStamps: ReadonlyMap<string, number>; + /** Tool calls seen per step in the CURRENT turn (`stepId` → call ids). */ + readonly calls: ReadonlyMap<string, ReadonlySet<string>>; + /** Tool results seen per step in the CURRENT turn (`stepId` → call ids). */ + readonly results: ReadonlyMap<string, ReadonlySet<string>>; + /** Longest tool `durationMs` per step in the CURRENT turn. */ + readonly toolMs: ReadonlyMap<string, number>; +} + +export function initialCacheTimerState(): CacheTimerState { + return { + running: null, + stepStamps: new Map(), + turnStamps: new Map(), + calls: new Map(), + results: new Map(), + toolMs: new Map(), + }; +} + +function addTo( + map: ReadonlyMap<string, ReadonlySet<string>>, + key: string, + value: string, +): ReadonlyMap<string, ReadonlySet<string>> { + const next = new Map(map); + next.set(key, new Set([...(map.get(key) ?? []), value])); + return next; +} + +function withEntry(map: ReadonlyMap<string, number>, key: string, value: number) { + const next = new Map(map); + next.set(key, value); + return next; +} + +/** + * Stop the running STEP gap (if any). A step that made tool calls is stamped + * with `max(wall-clock elapsed, longest backend tool duration)` — the backend + * figure keeps a late-join replay (whole step delivered in one burst, ~0 + * elapsed) honest. A step with no tool calls is discarded. + */ +function stopStep(state: CacheTimerState, now: number): CacheTimerState { + const r = state.running; + if (r === null || r.kind !== "step") return state; + if (r.toolCount === 0) return { ...state, running: null }; + const elapsed = Math.max(0, now - r.startedAt, r.maxToolMs); + return { ...state, running: null, stepStamps: withEntry(state.stepStamps, r.stepId, elapsed) }; +} + +/** + * Stop the running TURN gap (if any) and stamp it on its turn. A seeded gap + * (`turnId === null`) is stamped on `lastTurnId()` — the most recent turn's + * id — or dropped when that is unknown. + */ +export function stopTurnGap( + state: CacheTimerState, + now: number, + lastTurnId: () => string | undefined, +): CacheTimerState { + const r = state.running; + if (r === null || r.kind !== "turn") return state; + const turnId = r.turnId ?? lastTurnId(); + if (turnId === undefined) return { ...state, running: null }; + const elapsed = Math.max(0, now - r.startedAt); + return { ...state, running: null, turnStamps: withEntry(state.turnStamps, turnId, elapsed) }; +} + +/** A turn ended: close any step gap, forget the turn's tool bookkeeping, start the turn gap. */ +function endTurn(state: CacheTimerState, turnId: string, now: number): CacheTimerState { + const s = stopStep(state, now); + return { + ...s, + running: { kind: "turn", turnId, startedAt: now }, + calls: new Map(), + results: new Map(), + toolMs: new Map(), + }; +} + +/** + * Fold one live event into the cache-timer state at time `now`. `lastTurnId` + * resolves a seeded turn gap's turn when a new turn starts (see + * `stopTurnGap`); it is only called in that case. + */ +export function foldCacheTimerEvent( + state: CacheTimerState, + event: AgentEvent, + now: number, + lastTurnId: () => string | undefined, +): CacheTimerState { + const r = state.running; + switch (event.type) { + case "user-message": + case "turn-start": + // A new turn is starting — the user (here or elsewhere) sent a message. + return stopTurnGap(state, now, lastTurnId); + + case "tool-call": { + // A call from a DIFFERENT step means the next step is already streaming. + const s = r?.kind === "step" && r.stepId !== event.stepId ? stopStep(state, now) : state; + return { ...s, calls: addTo(s.calls, event.stepId, event.toolCallId) }; + } + + case "tool-result": { + const results = addTo(state.results, event.stepId, event.toolCallId); + const dur = event.durationMs ?? 0; + const toolMs = + dur > (state.toolMs.get(event.stepId) ?? 0) + ? withEntry(state.toolMs, event.stepId, dur) + : state.toolMs; + let s: CacheTimerState = { ...state, results, toolMs }; + if (r?.kind === "step" && r.stepId === event.stepId) { + const pending = new Set(r.pending); + pending.delete(event.toolCallId); + s = { ...s, running: { ...r, pending, maxToolMs: Math.max(r.maxToolMs, dur) } }; + // Every tool of the step has returned → the next request fires now. + if (pending.size === 0) s = stopStep(s, now); + } + return s; + } + + case "step-complete": { + const s = stopStep(stopTurnGap(state, now, lastTurnId), now); + const calls = s.calls.get(event.stepId) ?? new Set<string>(); + const done = s.results.get(event.stepId) ?? new Set<string>(); + const pending = new Set([...calls].filter((id) => !done.has(id))); + return { + ...s, + running: { + kind: "step", + turnId: event.turnId, + stepId: event.stepId, + startedAt: now, + toolCount: calls.size, + pending, + maxToolMs: s.toolMs.get(event.stepId) ?? 0, + }, + }; + } + + case "text-delta": + case "reasoning-delta": + case "reasoning-end": + case "provider-retry": + // Output (or a retry) from the next request: the gap is over. + return r?.kind === "step" ? stopStep(state, now) : state; + + case "usage": + return r?.kind === "step" && event.stepId !== undefined && event.stepId !== r.stepId + ? stopStep(state, now) + : state; + + case "done": + case "error": + return endTurn(state, event.turnId, now); + + default: + return state; + } +} + +/** + * Seed a turn gap from an authoritative "last activity" timestamp (the + * backend's `lastActivityAt`, for a conversation that is idle — e.g. after a + * reload, when no live `done` was seen). Never overrides a running step gap; + * a running turn gap keeps the LATER start. Non-finite input is ignored. + */ +export function seedTurnGap(state: CacheTimerState, at: number): CacheTimerState { + if (!Number.isFinite(at)) return state; + const r = state.running; + if (r === null) return { ...state, running: { kind: "turn", turnId: null, startedAt: at } }; + if (r.kind === "turn" && at > r.startedAt) return { ...state, running: { ...r, startedAt: at } }; + return state; +} + +// ── View ────────────────────────────────────────────────────────────────────── + +/** Formatted cache-timer view: label + colour severity + capped flag. */ +export interface CacheTimerView { + /** `"m:ss"` up to 10 minutes, then `"10+ min"`. */ + readonly label: string; + /** Colour by the 5-minute cache TTL (maps to DaisyUI `badge-{level}`). */ + readonly level: "success" | "warning" | "error"; + /** Whether the 10-minute display cap was reached (a live timer may stop ticking). */ + readonly capped: boolean; +} + +/** + * Build the view for `elapsedMs`. Colour tracks the cache TTL: success under + * 4 min, warning in the last minute, error once the cache has likely expired. + * Negative / non-finite elapsed (clock skew) clamps to `0:00`. + */ +export function viewCacheTimer(elapsedMs: number): CacheTimerView { + const ms = Number.isFinite(elapsedMs) && elapsedMs > 0 ? elapsedMs : 0; + const level = ms >= CACHE_TTL_MS ? "error" : ms >= WARN_AT_MS ? "warning" : "success"; + if (ms >= CACHE_TIMER_MAX_MS) return { label: "10+ min", level, capped: true }; + const totalSeconds = Math.floor(ms / 1000); + const minutes = Math.floor(totalSeconds / 60); + const seconds = totalSeconds % 60; + return { label: `${minutes}:${seconds.toString().padStart(2, "0")}`, level, capped: false }; +} + +/** + * Index of the most recent turn's metrics row — the LAST `turn-metrics` row, + * provided no user message follows it. Where a SEEDED turn gap (no `turnId`) + * renders. `null` when there is no such row. + */ +export function lastTurnRowIndex(rows: readonly MetricsRow[]): number | null { + for (let i = rows.length - 1; i >= 0; i--) { + const row = rows[i]; + if (row === undefined) continue; + if (row.kind === "turn-metrics") return i; + if (row.kind === "group" && row.group.kind === "single" && row.group.chunk.role === "user") { + return null; + } + } + return null; +} diff --git a/src/core/metrics/format.test.ts b/src/core/metrics/format.test.ts index 97170d0..3e8ebdd 100644 --- a/src/core/metrics/format.test.ts +++ b/src/core/metrics/format.test.ts @@ -46,7 +46,7 @@ describe("viewStepMetrics", () => { }; const view = viewStepMetrics(step, 0); expect(view.label).toBe("step 1"); - expect(view.tokensLabel).toBe("1,801 tok"); + expect(view.tokensLabel).toBe("1,234 in / 567 out"); expect(view.tps).toBe("473 tok/s"); expect(view.ttft).toBe("820ms"); expect(view.decode).toBe("1.2s"); @@ -105,15 +105,14 @@ describe("viewStepMetrics", () => { expect(view.tps).toBe("100 tok/s"); }); - it("tps falls back to genTotalMs when decodeMs absent", () => { + it("tps is null when decodeMs is absent (never falls back to genTotalMs, which includes TTFT)", () => { const step: StepMetrics = { stepId: "s1" as StepId, usage: { inputTokens: 100, outputTokens: 50 }, genTotalMs: 800, }; const view = viewStepMetrics(step, 0); - // 50 / (800/1000) = 62.5 → rounds to 63 - expect(view.tps).toBe("63 tok/s"); + expect(view.tps).toBeNull(); }); }); @@ -133,8 +132,8 @@ describe("viewTurnMetrics", () => { ], }; const view = viewTurnMetrics(turn); - expect(view.tokensLabel).toBe("1,234 tok"); - expect(view.breakdown).toBe("1,000 in / 234 out"); + expect(view.tokensLabel).toBe("234 tok out"); + expect(view.breakdown).toBe("1,000 in"); expect(view.tps).toBe("78 tok/s"); expect(view.duration).toBe("5.0s"); }); @@ -146,7 +145,7 @@ describe("viewTurnMetrics", () => { steps: [], }; const view = viewTurnMetrics(turn); - expect(view.breakdown).toBe("1,000 in / 234 out / 500 cache"); + expect(view.breakdown).toBe("1,000 in (500 cached)"); }); it("breakdown omits cache when not present", () => { @@ -156,7 +155,7 @@ describe("viewTurnMetrics", () => { steps: [], }; const view = viewTurnMetrics(turn); - expect(view.breakdown).toBe("100 in / 50 out"); + expect(view.breakdown).toBe("100 in"); }); it("tps is null when no step has decodeMs or genTotalMs", () => { @@ -184,7 +183,7 @@ describe("viewTurnMetrics", () => { expect(view.duration).toBeNull(); }); - it("sums decodeMs across steps (fallback genTotalMs per step) for tps", () => { + it("turn tps counts only steps that report decodeMs (both tokens and time)", () => { const turn: TurnMetrics = { turnId: "t1", usage: { inputTokens: 300, outputTokens: 150 }, @@ -203,9 +202,9 @@ describe("viewTurnMetrics", () => { ], }; const view = viewTurnMetrics(turn); - // step1 uses decodeMs=800, step2 falls back to genTotalMs=2000 → total=2800ms - // 150 / (2800/1000) = 53.57 → rounds to 54 - expect(view.tps).toBe("54 tok/s"); + // only step1 is timed: 50 / (800/1000) = 62.5 → 63. step2 (no decodeMs) + // contributes neither tokens nor time — never its TTFT-inflated genTotalMs. + expect(view.tps).toBe("63 tok/s"); }); }); @@ -260,40 +259,47 @@ describe("computeExpectedCachePct", () => { expect(computeExpectedCachePct({ inputTokens: 100, outputTokens: 0 }, null)).toBeNull(); }); - it("null when the prior turn cached nothing (denominator 0)", () => { - const prev = { inputTokens: 100, outputTokens: 0 }; - const current = { inputTokens: 200, outputTokens: 0, cacheReadTokens: 50 }; - expect(computeExpectedCachePct(current, prev)).toBeNull(); + it("null when this turn has no first step", () => { + expect(computeExpectedCachePct(undefined, { inputTokens: 100, outputTokens: 0 })).toBeNull(); }); - it("100% when the whole prior cached prefix was read back (backend worked example)", () => { - // turn 1: cacheRead 0, cacheWrite 5146 → prefix 5146; turn 2 reads 5146 back. - const prev = { inputTokens: 5149, outputTokens: 0, cacheReadTokens: 0, cacheWriteTokens: 5146 }; - const current = { - inputTokens: 8462, - outputTokens: 0, - cacheReadTokens: 5146, - cacheWriteTokens: 3313, - }; - expect(computeExpectedCachePct(current, prev)).toBe(100); + it("null when the prior prompt was empty (denominator 0)", () => { + const prev = { inputTokens: 0, outputTokens: 0 }; + expect(computeExpectedCachePct({ inputTokens: 10, outputTokens: 0 }, prev)).toBeNull(); }); - it("drops below 100% when the cache busted (read < prior prefix)", () => { - const prev = { - inputTokens: 1000, - outputTokens: 0, - cacheReadTokens: 100, - cacheWriteTokens: 900, - }; - const current = { inputTokens: 1000, outputTokens: 0, cacheReadTokens: 500 }; - // 500 / (100 + 900) = 50% - expect(computeExpectedCachePct(current, prev)).toBe(50); + it("100% when the first step reads back the whole prior final prompt", () => { + // prev turn's last step sent 5149 tokens; this turn's first step (a + // superset prompt) reads 5149 of them from cache. + const prevLast = { inputTokens: 5149, outputTokens: 40, cacheReadTokens: 3000 }; + const first = { inputTokens: 8462, outputTokens: 0, cacheReadTokens: 5149 }; + expect(computeExpectedCachePct(first, prevLast)).toBe(100); + }); + + it("drops below 100% when the cache busted", () => { + const prevLast = { inputTokens: 1000, outputTokens: 0, cacheReadTokens: 900 }; + const first = { inputTokens: 1200, outputTokens: 0, cacheReadTokens: 500 }; + expect(computeExpectedCachePct(first, prevLast)).toBe(50); + }); + + it("0% (not null) on a full miss", () => { + const prevLast = { inputTokens: 1000, outputTokens: 0 }; + const first = { inputTokens: 1200, outputTokens: 0 }; + expect(computeExpectedCachePct(first, prevLast)).toBe(0); + }); + + it("is NOT inflated by multi-step turns (step-level, not turn sums)", () => { + // A 10-step turn reads the same prefix 10x; the old turn-sum ratio read + // ~10x the baseline and pinned at 100%. Step-level reflects the bust. + const prevLast = { inputTokens: 50_000, outputTokens: 0, cacheReadTokens: 49_000 }; + const first = { inputTokens: 51_000, outputTokens: 0, cacheReadTokens: 10_000 }; + expect(computeExpectedCachePct(first, prevLast)).toBe(20); }); - it("clamps to 100 if read somehow exceeds the prior prefix", () => { - const prev = { inputTokens: 100, outputTokens: 0, cacheWriteTokens: 100 }; - const current = { inputTokens: 100, outputTokens: 0, cacheReadTokens: 250 }; - expect(computeExpectedCachePct(current, prev)).toBe(100); + it("clamps to 100", () => { + const prevLast = { inputTokens: 100, outputTokens: 0 }; + const first = { inputTokens: 300, outputTokens: 0, cacheReadTokens: 250 }; + expect(computeExpectedCachePct(first, prevLast)).toBe(100); }); }); @@ -303,9 +309,9 @@ describe("viewExpectedCache", () => { }); it("success level + hit flag for full retention", () => { - const prev = { inputTokens: 5149, outputTokens: 0, cacheWriteTokens: 5146 }; - const current = { inputTokens: 8462, outputTokens: 0, cacheReadTokens: 5146 }; - const v = viewExpectedCache(current, prev); + const prevLast = { inputTokens: 5146, outputTokens: 0 }; + const first = { inputTokens: 8462, outputTokens: 0, cacheReadTokens: 5146 }; + const v = viewExpectedCache(first, prevLast); expect(v?.pct).toBe(100); expect(v?.level).toBe("success"); expect(v?.isHit).toBe(true); diff --git a/src/core/metrics/format.ts b/src/core/metrics/format.ts index 894bd54..5a13a9a 100644 --- a/src/core/metrics/format.ts +++ b/src/core/metrics/format.ts @@ -77,26 +77,31 @@ export function computeTps(outputTokens: number, elapsedMs: number | undefined): return outputTokens / (elapsedMs / 1000); } -function totalTokens(u: Usage): number { - return u.inputTokens + u.outputTokens; -} - function formatBreakdown(u: Usage): string { - let s = `${formatTokens(u.inputTokens)} in / ${formatTokens(u.outputTokens)} out`; + let s = `${formatTokens(u.inputTokens)} in`; if (u.cacheReadTokens !== undefined && u.cacheReadTokens > 0) { - s += ` / ${formatTokens(u.cacheReadTokens)} cache`; + s += ` (${formatTokens(u.cacheReadTokens)} cached)`; } return s; } +/** + * A step's tokens-per-second: `outputTokens / decodeMs`. Decode time is the + * model's generation window (first output token → stream end) with tool + * execution excluded by the backend. `genTotalMs` is deliberately NOT a + * fallback: it includes time-to-first-token (prompt processing / queueing), + * which would understate the rate by multiples. No decode time → no TPS. + */ +function stepTps(step: StepMetrics): number | null { + return computeTps(step.usage.outputTokens, step.decodeMs); +} + /** Build a formatted view of a single step's metrics. */ export function viewStepMetrics(step: StepMetrics, index: number): StepMetricsView { - const total = totalTokens(step.usage); - const tps = computeTps(step.usage.outputTokens, step.decodeMs ?? step.genTotalMs); return { label: `step ${index + 1}`, - tokensLabel: `${formatTokens(total)} tok`, - tps: formatTps(tps), + tokensLabel: `${formatTokens(step.usage.inputTokens)} in / ${formatTokens(step.usage.outputTokens)} out`, + tps: formatTps(stepTps(step)), ttft: formatDuration(step.ttftMs), decode: formatDuration(step.decodeMs), genTotal: formatDuration(step.genTotalMs), @@ -130,19 +135,25 @@ export function viewCacheRate(u: Usage): CacheRateView { } /** - * Expected cache (retention): of the cache that existed going INTO this turn, how - * much was read back — `clamp01(cacheRead_N / (cacheRead_{N-1} + cacheWrite_{N-1}))`. - * The denominator is the PRIOR turn's cached prefix (what it read + what it wrote). - * Ideally ~100% on every turn after the first; <100% = the cache busted/expired. + * Expected cache (retention): of the prompt the PREVIOUS turn last sent (its + * final step's input), how much this turn's FIRST step read back from cache — + * `clamp01(firstStep.cacheRead / prevLastStep.input)`. Everything in that + * prompt is a prefix of this turn's first prompt, so ~100% means the cache + * survived between turns; lower means it busted/expired. Step-level on both + * sides: comparing whole-turn sums is meaningless (a multi-step turn reads + * the same prefix once per step). * - * Returns `null` when it cannot be derived: no prior turn (`prev === null`) or the - * prior turn cached nothing (denominator <= 0) — distinct from a real 0%. + * Returns `null` when it cannot be derived: no earlier turn, no first step, + * or the previous prompt was empty — distinct from a real 0%. */ -export function computeExpectedCachePct(current: Usage, prev: Usage | null): number | null { - if (prev === null) return null; - const denom = (prev.cacheReadTokens ?? 0) + (prev.cacheWriteTokens ?? 0); +export function computeExpectedCachePct( + firstStep: Usage | undefined, + prevLastStep: Usage | null, +): number | null { + if (prevLastStep === null || firstStep === undefined) return null; + const denom = prevLastStep.inputTokens; if (denom <= 0) return null; - const read = current.cacheReadTokens ?? 0; + const read = firstStep.cacheReadTokens ?? 0; const rate = read / denom; const clamped = rate < 0 ? 0 : rate > 1 ? 1 : rate; return Math.round(clamped * 100); @@ -152,28 +163,36 @@ export function computeExpectedCachePct(current: Usage, prev: Usage | null): num * Build a view of the cross-turn retention (percentage + colour level + hit flag), * or `null` when it can't be derived (see `computeExpectedCachePct`). */ -export function viewExpectedCache(current: Usage, prev: Usage | null): CacheRateView | null { - const pct = computeExpectedCachePct(current, prev); +export function viewExpectedCache( + firstStep: Usage | undefined, + prevLastStep: Usage | null, +): CacheRateView | null { + const pct = computeExpectedCachePct(firstStep, prevLastStep); if (pct === null) return null; - return { pct, level: cacheLevel(pct), isHit: (current.cacheReadTokens ?? 0) > 0 }; + return { pct, level: cacheLevel(pct), isHit: (firstStep?.cacheReadTokens ?? 0) > 0 }; } -/** Build a formatted view of a turn's aggregate metrics. */ +/** + * Build a formatted view of a turn's aggregate metrics. + * + * Turn TPS = output tokens ÷ decode time, summed over ONLY the steps that + * report a decode time (both numerator and denominator), so a step without + * timing can neither inflate nor deflate it. Tool execution between steps is + * never part of any step's decode time. + */ export function viewTurnMetrics(turn: TurnMetrics, turnNumber?: number): TurnMetricsView { - const total = totalTokens(turn.usage); - let totalGenMs: number | undefined; + let timedOut = 0; + let decodeMs: number | undefined; for (const step of turn.steps) { - const stepMs = step.decodeMs ?? step.genTotalMs; - if (stepMs !== undefined) { - totalGenMs = (totalGenMs ?? 0) + stepMs; - } + if (step.decodeMs === undefined || step.decodeMs <= 0) continue; + timedOut += step.usage.outputTokens; + decodeMs = (decodeMs ?? 0) + step.decodeMs; } - const tps = computeTps(turn.usage.outputTokens, totalGenMs); return { label: turnNumber !== undefined ? `turn ${turnNumber}` : "turn", - tokensLabel: `${formatTokens(total)} tok`, + tokensLabel: `${formatTokens(turn.usage.outputTokens)} tok out`, breakdown: formatBreakdown(turn.usage), - tps: formatTps(tps), + tps: formatTps(computeTps(timedOut, decodeMs)), duration: formatDuration(turn.durationMs), }; } diff --git a/src/core/metrics/index.ts b/src/core/metrics/index.ts index d3c9669..d9df7f5 100644 --- a/src/core/metrics/index.ts +++ b/src/core/metrics/index.ts @@ -1,4 +1,19 @@ export { + CACHE_TIMER_MAX_MS, + CACHE_TTL_MS, + type CacheTimerState, + type CacheTimerView, + foldCacheTimerEvent, + initialCacheTimerState, + lastTurnRowIndex, + type RunningGap, + type RunningStepGap, + type RunningTurnGap, + seedTurnGap, + stopTurnGap, + viewCacheTimer, +} from "./cache-timer"; +export { type ContextUsage, computeCachePct, computeContextUsage, diff --git a/src/core/metrics/place.test.ts b/src/core/metrics/place.test.ts index 9c925a3..7695e21 100644 --- a/src/core/metrics/place.test.ts +++ b/src/core/metrics/place.test.ts @@ -28,31 +28,36 @@ function assistantGroup(seq: number, text: string): RenderGroup { }; } -function toolCallGroup(seq: number, stepId: string, toolCallId: string): RenderGroup { +function toolCallGroup(stepId: string, toolCallId: string): RenderGroup { return { - kind: "single", - chunk: { - seq, - role: "assistant", - chunk: { + kind: "tool", + stepId, + provisional: false, + entry: { + toolCallId, + toolName: "test", + result: null, + call: { type: "tool-call", toolCallId, toolName: "test", input: {}, stepId: stepId as StepId, }, - provisional: false, }, }; } -function toolResultGroup(seq: number, stepId: string, toolCallId: string): RenderGroup { +function toolResultGroup(stepId: string, toolCallId: string): RenderGroup { return { - kind: "single", - chunk: { - seq, - role: "tool", - chunk: { + kind: "tool", + stepId, + provisional: false, + entry: { + toolCallId, + toolName: "test", + call: null, + result: { type: "tool-result", toolCallId, toolName: "test", @@ -60,29 +65,10 @@ function toolResultGroup(seq: number, stepId: string, toolCallId: string): Rende isError: false, stepId: stepId as StepId, }, - provisional: false, }, }; } -function toolBatchGroup(stepId: string, toolCallIds: string[]): RenderGroup { - return { - kind: "tool-batch", - stepId, - entries: toolCallIds.map((id) => ({ - call: { - type: "tool-call" as const, - toolCallId: id, - toolName: "test", - input: {}, - stepId: stepId as StepId, - }, - result: null, - })), - provisional: false, - }; -} - function makeStep(stepId: string, inputTokens: number, outputTokens: number): StepMetrics { return { stepId: stepId as StepId, @@ -169,9 +155,9 @@ describe("interleaveTurnMetrics", () => { it("head-aligned: segment i gets entries[i]", () => { const g1 = userGroup(1, "q1"); - const g2 = toolCallGroup(2, "s1", "c1"); + const g2 = toolCallGroup("s1", "c1"); const g3 = userGroup(3, "q2"); - const g4 = toolCallGroup(4, "s2", "c2"); + const g4 = toolCallGroup("s2", "c2"); const step1 = makeStep("s1", 100, 50); const step2 = makeStep("s2", 200, 80); const rows = interleaveTurnMetrics( @@ -192,7 +178,7 @@ describe("interleaveTurnMetrics", () => { it("a trailing segment with no entry (in-flight turn) renders no metrics", () => { const g1 = userGroup(1, "q1"); - const g2 = toolCallGroup(2, "s1", "c1"); + const g2 = toolCallGroup("s1", "c1"); const g3 = userGroup(3, "q2"); const g4 = assistantGroup(4, "a2"); const step = makeStep("s1", 100, 50); @@ -220,26 +206,28 @@ describe("interleaveTurnMetrics", () => { expectTurnMetricsAt(rows, 2, "t1"); }); - it("tool step anchors inline after its tool-batch group", () => { + it("tool step anchors inline after its LAST independent tool card", () => { const g1 = userGroup(1, "q1"); - const g2 = toolBatchGroup("t#0", ["c1", "c2"]); + const g2 = toolCallGroup("t#0", "c1"); + const parallel = toolCallGroup("t#0", "c2"); const g3 = assistantGroup(3, "a1"); const step0 = makeStep("t#0", 100, 50); const step1 = makeStep("t#1", 200, 80); const turn = makeEntry("t1", 300, 130, [step0, step1]); - const rows = interleaveTurnMetrics([g1, g2, g3], [turn]); + const rows = interleaveTurnMetrics([g1, g2, parallel, g3], [turn]); - expect(rows).toHaveLength(5); + expect(rows).toHaveLength(6); expectGroupAt(rows, 0, g1); expectGroupAt(rows, 1, g2); - expectStepMetricsAt(rows, 2, "t#0", 0); - expectGroupAt(rows, 3, g3); - expectTurnMetricsAt(rows, 4, "t1"); + expectGroupAt(rows, 2, parallel); + expectStepMetricsAt(rows, 3, "t#0", 0); + expectGroupAt(rows, 4, g3); + expectTurnMetricsAt(rows, 5, "t1"); }); it("single tool-call group anchors its step", () => { const g1 = userGroup(1, "q1"); - const g2 = toolCallGroup(2, "s1", "c1"); + const g2 = toolCallGroup("s1", "c1"); const g3 = assistantGroup(3, "a1"); const step = makeStep("s1", 100, 50); const turn = makeEntry("t1", 100, 50, [step]); @@ -255,7 +243,7 @@ describe("interleaveTurnMetrics", () => { it("single tool-result group anchors its step", () => { const g1 = userGroup(1, "q1"); - const g2 = toolResultGroup(2, "s1", "c1"); + const g2 = toolResultGroup("s1", "c1"); const g3 = assistantGroup(3, "a1"); const step = makeStep("s1", 100, 50); const turn = makeEntry("t1", 100, 50, [step]); @@ -271,9 +259,9 @@ describe("interleaveTurnMetrics", () => { it("multi-step: each tool step inline, unanchored text step skipped", () => { const g1 = userGroup(1, "q1"); - const g2 = toolBatchGroup("t#0", ["c1"]); + const g2 = toolCallGroup("t#0", "c1"); const g3 = assistantGroup(2, "thinking"); - const g4 = toolBatchGroup("t#1", ["c2", "c3"]); + const g4 = toolCallGroup("t#1", "c2"); const g5 = assistantGroup(3, "a1"); const step0 = makeStep("t#0", 100, 50); const step1 = makeStep("t#1", 200, 80); @@ -294,10 +282,10 @@ describe("interleaveTurnMetrics", () => { it("multiple turns head-aligned with inline steps", () => { const g1 = userGroup(1, "q1"); - const g2 = toolBatchGroup("s1", ["c1"]); + const g2 = toolCallGroup("s1", "c1"); const g3 = assistantGroup(2, "a1"); const g4 = userGroup(3, "q2"); - const g5 = toolCallGroup(4, "s2", "c2"); + const g5 = toolCallGroup("s2", "c2"); const step1 = makeStep("s1", 100, 50); const step2 = makeStep("s2", 200, 80); const rows = interleaveTurnMetrics( @@ -332,7 +320,7 @@ describe("interleaveTurnMetrics", () => { it("fewer metrics than segments: trailing segments are bare", () => { const g1 = userGroup(1, "q1"); - const g2 = toolCallGroup(2, "s1", "c1"); + const g2 = toolCallGroup("s1", "c1"); const g3 = userGroup(3, "q2"); const g4 = assistantGroup(4, "a2"); const g5 = userGroup(5, "q3"); @@ -361,7 +349,7 @@ describe("interleaveTurnMetrics", () => { // stepId groups) — the failure case. The text-only turn MUST get its OWN // entry (t3), NOT the trimmed t1's stale metrics. const g3 = userGroup(3, "q2"); - const g4 = toolBatchGroup("s2", ["c2"]); + const g4 = toolCallGroup("s2", "c2"); const g5 = assistantGroup(4, "tool-reply"); const g6 = userGroup(5, "q3"); const g7 = assistantGroup(6, "text-reply"); @@ -401,7 +389,12 @@ describe("interleaveTurnMetrics", () => { total: { turnId: "t1", usage: { inputTokens: 1000, outputTokens: 10, cacheReadTokens: 500 }, - steps: [], + steps: [ + { + stepId: "t1#0" as StepId, + usage: { inputTokens: 1000, outputTokens: 10, cacheReadTokens: 500 }, + }, + ], }, }, { @@ -424,14 +417,14 @@ describe("interleaveTurnMetrics", () => { // Cumulative includes BOTH turns (t1 + t2): input 3000, cacheRead 2100. expect(tmRows[0]?.cumulativeUsage.inputTokens).toBe(3000); expect(tmRows[0]?.cumulativeUsage.cacheReadTokens).toBe(2100); - // Retention baseline is the prior finalized turn (t1, even though trimmed). - expect(tmRows[0]?.prevTurnUsage?.inputTokens).toBe(1000); - expect(tmRows[0]?.prevTurnUsage?.cacheReadTokens).toBe(500); + // Retention baseline is the prior finalized turn's LAST step (t1, even though trimmed). + expect(tmRows[0]?.prevLastStepUsage?.inputTokens).toBe(1000); + expect(tmRows[0]?.prevLastStepUsage?.cacheReadTokens).toBe(500); }); it("in-flight turn (no durationMs) still produces turn row", () => { const g1 = userGroup(1, "q1"); - const g2 = toolCallGroup(2, "s1", "c1"); + const g2 = toolCallGroup("s1", "c1"); const step = makeStep("s1", 100, 50); const turn: TurnMetricsEntry = { turnId: "t1", @@ -454,7 +447,7 @@ describe("interleaveTurnMetrics", () => { it("leading non-turn groups emit as plain group rows", () => { const g0 = assistantGroup(1, "system msg"); const g1 = userGroup(2, "q1"); - const g2 = toolCallGroup(3, "s1", "c1"); + const g2 = toolCallGroup("s1", "c1"); const step = makeStep("s1", 100, 50); const rows = interleaveTurnMetrics([g0, g1, g2], [makeEntry("t1", 100, 50, [step])]); @@ -468,7 +461,7 @@ describe("interleaveTurnMetrics", () => { it("trimmed turn (more metrics than segments) does NOT emit a standalone row at the top", () => { const g1 = userGroup(1, "q1"); - const g2 = toolCallGroup(2, "s1", "c1"); + const g2 = toolCallGroup("s1", "c1"); const step1 = makeStep("s1", 100, 50); const step2 = makeStep("s2", 200, 80); const rows = interleaveTurnMetrics( @@ -504,7 +497,7 @@ describe("interleaveTurnMetrics", () => { it("progressive: entry with steps but total=null emits step rows and NO turn-metrics row", () => { const g1 = userGroup(1, "q1"); - const g2 = toolBatchGroup("s1", ["c1"]); + const g2 = toolCallGroup("s1", "c1"); const g3 = assistantGroup(2, "a1"); const step1 = makeStep("s1", 100, 50); const entry = makeProgressiveEntry("t1", [step1]); @@ -519,7 +512,7 @@ describe("interleaveTurnMetrics", () => { it("entry with total emits step rows + a turn-metrics row", () => { const g1 = userGroup(1, "q1"); - const g2 = toolBatchGroup("s1", ["c1"]); + const g2 = toolCallGroup("s1", "c1"); const g3 = assistantGroup(2, "a1"); const step1 = makeStep("s1", 100, 50); const entry = makeEntry("t1", 100, 50, [step1]); @@ -606,16 +599,36 @@ describe("interleaveTurnMetrics — cumulative usage (cache total)", () => { expect(tm[0]?.cumulativeUsage.cacheReadTokens).toBe(500); }); - it("carries the prior finalized turn's usage as the retention baseline", () => { + it("carries the prior finalized turn's LAST STEP usage as the retention baseline", () => { + const steps = (id: string): StepMetrics[] => [ + { stepId: `${id}#0` as StepId, usage: { inputTokens: 1000, outputTokens: 5 } }, + { + stepId: `${id}#1` as StepId, + usage: { inputTokens: 1669, outputTokens: 5, cacheReadTokens: 384 }, + }, + ]; + const t1: TurnMetrics = { + turnId: "t1", + usage: { inputTokens: 2669, outputTokens: 10, cacheReadTokens: 384 }, + steps: steps("t1"), + }; + const t2: TurnMetrics = { + turnId: "t2", + usage: { inputTokens: 2737, outputTokens: 10, cacheReadTokens: 2560 }, + steps: steps("t2"), + }; const rows = interleaveTurnMetrics( [userGroup(1, "q1"), assistantGroup(2, "a1"), userGroup(3, "q2"), assistantGroup(4, "a2")], - [cacheEntry("t1", 2669, 10, 384), cacheEntry("t2", 2737, 10, 2560)], + [ + { turnId: "t1", steps: t1.steps, total: t1 }, + { turnId: "t2", steps: t2.steps, total: t2 }, + ], ); const tm = turnMetricsRows(rows); // first finalized turn has no earlier baseline - expect(tm[0]?.prevTurnUsage).toBeNull(); - // second turn's baseline is the first turn's usage - expect(tm[1]?.prevTurnUsage?.inputTokens).toBe(2669); - expect(tm[1]?.prevTurnUsage?.cacheReadTokens).toBe(384); + expect(tm[0]?.prevLastStepUsage).toBeNull(); + // second turn's baseline is the first turn's LAST step (not the turn sum) + expect(tm[1]?.prevLastStepUsage?.inputTokens).toBe(1669); + expect(tm[1]?.prevLastStepUsage?.cacheReadTokens).toBe(384); }); }); diff --git a/src/core/metrics/place.ts b/src/core/metrics/place.ts index 7122b09..d4c343a 100644 --- a/src/core/metrics/place.ts +++ b/src/core/metrics/place.ts @@ -1,11 +1,17 @@ import type { Usage } from "@dispatch/wire"; import type { RenderGroup } from "../chunks"; -import type { MetricsRow, TurnMetricsEntry } from "./types"; +import type { MetricsRow, StepMetrics, TurnMetricsEntry } from "./types"; function groupStepId(g: RenderGroup): string | undefined { - if (g.kind === "tool-batch") return g.stepId; - const c = g.chunk.chunk; - return c.type === "tool-call" || c.type === "tool-result" ? c.stepId : undefined; + return g.kind === "tool" ? g.stepId : undefined; +} + +function groupTurnId(g: RenderGroup): string | undefined { + return g.kind === "tool" ? g.turnId : g.chunk.turnId; +} + +function isUser(g: RenderGroup): boolean { + return g.kind === "single" && g.chunk.role === "user"; } /** Element-wise sum of two token usages (cache fields included only when nonzero). */ @@ -21,277 +27,255 @@ function addUsage(a: Usage, b: Usage): Usage { return out; } +/** A run of consecutive groups that belong to one (possibly unknown) turn. */ +interface Segment { + readonly start: number; + readonly end: number; // exclusive + /** Turn resolved from the segment's own groups (turnId / stepId), if any. */ + readonly known: string | undefined; + /** Whether the segment holds model output (anything but user / error chunks). */ + readonly hasResponse: boolean; +} + /** - * Interleave turn metrics into the rendered transcript. + * Assign every group to the turn that produced it. EXACT whenever the data + * carries provenance — which is always for anything produced since the backend + * began stamping `StoredChunk.turnId` (and always for live chunks, which carry + * the live event's `turnId`): * - * Splits groups into per-turn segments: a new segment begins at each `single` - * group with `group.chunk.role === "user"`. Segments are matched to entries - * by `stepId` presence when possible (robust against chat-limit trimming: when - * a turn's user message is trimmed, positional alignment would be off, but - * stepId matching still finds the right entry). Segments with no stepId-bearing - * groups (text-only turns) fall back to POSITIONAL tail-alignment: since the - * loaded transcript is always a SUFFIX of the full turn history (the chat limit - * keeps the newest and unloads the oldest), segment `seg` ↔ entry `K - T + seg`. + * 1. A group's own `turnId` (committed `StoredChunk.turnId` / live event). + * 2. Else the turn whose metrics contain the group's `stepId`. * - * Within a segment that has a matched entry, each completed step's metrics - * are placed INLINE right after the last group bearing that step's `stepId`. - * Steps whose `stepId` does not appear in any group ("unanchored"): - * - If the segment HAS stepId-bearing groups (tool chunks exist but this step's - * were trimmed): SKIPPED (no blank "step N · 0 tok" bubbles). - * - If the segment has NO stepId-bearing groups (text-only turn): placed at the - * segment tail before the turn-metrics row (the original behavior). - * - * A `turn-metrics` row is emitted ONLY when `entry.total !== null` (i.e. the turn - * is finalized via `done` or durable data). A still-generating turn emits no - * turn-total row. - * - * Fully trimmed turns (entries whose content was unloaded by the chat limit and - * which match no segment) are NOT rendered as standalone rows — that previously - * piled a wall of stale cache badges at the top of a long, trimmed transcript. - * Their usage still counts toward the per-turn "chat total" cumulative (computed - * across ALL finalized turns in entry-array order), so the running cache rate - * stays correct regardless of which turns were trimmed; paging earlier history - * back in ("Show earlier messages") re-matches them and re-renders their rows. + * Only LEGACY groups (persisted before turn provenance existed, with no + * stepId) are left; they are resolved per segment. A segment boundary is a + * user message start (a user chunk after a non-user one, or any user TEXT + * chunk — a message carries at most one, first; image chunks continue it) or a + * change of known turn. A segment with a known turn assigns it to its legacy + * groups. Runs of still-unknown (text-only legacy) segments are matched to the + * UNCLAIMED metrics entries that lie strictly between their resolved + * neighbours' entries — order-preserving, and only when the count is + * unambiguous (leading runs tail-align, since chat-limit trimming removes the + * oldest turns). An ambiguous run gets NO metrics rather than wrong ones. */ -export function interleaveTurnMetrics( +function assignTurns( groups: readonly RenderGroup[], entries: readonly TurnMetricsEntry[], -): readonly MetricsRow[] { - if (entries.length === 0) { - return groups.map((g) => ({ kind: "group" as const, group: g })); - } +): (string | undefined)[] { + const turnOfStep = new Map<string, string>(); + const entryIndex = new Map<string, number>(); + entries.forEach((e, i) => { + entryIndex.set(e.turnId, i); + for (const s of e.steps) turnOfStep.set(s.stepId, e.turnId); + }); + + const direct: (string | undefined)[] = groups.map((g) => { + const t = groupTurnId(g); + if (t !== undefined) return t; + const sid = groupStepId(g); + return sid !== undefined ? turnOfStep.get(sid) : undefined; + }); - const segmentStarts: number[] = []; + // Fast path: everything resolved directly (the normal case). + if (direct.every((t) => t !== undefined)) return direct; + + // Segment the transcript (legacy resolution). + const segments: Segment[] = []; + let segStart = 0; + let segKnown: string | undefined; + let segHasResponse = false; + const close = (end: number) => { + if (end > segStart) { + segments.push({ start: segStart, end, known: segKnown, hasResponse: segHasResponse }); + } + }; for (let i = 0; i < groups.length; i++) { const g = groups[i]; - if (g !== undefined && g.kind === "single" && g.chunk.role === "user") { - segmentStarts.push(i); + if (g === undefined) continue; + const prev = i > 0 ? groups[i - 1] : undefined; + const t = direct[i]; + const userStart = + isUser(g) && + (prev === undefined || + !isUser(prev) || + (g.kind === "single" && g.chunk.chunk.type === "text")); + const turnChange = t !== undefined && segKnown !== undefined && t !== segKnown; + if (i > segStart && (userStart || turnChange)) { + close(i); + segStart = i; + segKnown = undefined; + segHasResponse = false; + } + if (t !== undefined && segKnown === undefined) segKnown = t; + if (!isUser(g) && !(g.kind === "single" && g.chunk.chunk.type === "error")) { + segHasResponse = true; } } + close(groups.length); - let T = segmentStarts.length; - - // No user messages — e.g. a compacted conversation whose history starts - // with a system summary. Treat the entire transcript as one segment so - // turn/step metrics can still be placed. - if (T === 0 && entries.length > 0) { - segmentStarts.push(0); - T = 1; - } - - if (T === 0) { - return groups.map((g) => ({ kind: "group" as const, group: g })); + const segTurn: (string | undefined)[] = segments.map((s) => s.known); + const claimed = new Set<number>(); + for (const t of segTurn) { + const idx = t !== undefined ? entryIndex.get(t) : undefined; + if (idx !== undefined) claimed.add(idx); } + // Entry index of a resolved segment; a known turn with no entry yet (an + // in-flight turn whose first step hasn't completed) sits after every entry. + const segEntryIdx = (k: number): number => { + const t = segTurn[k]; + if (t === undefined) return -1; + return entryIndex.get(t) ?? entries.length; + }; - const K = entries.length; - - // Build stepId → entry-index lookup for matching. - const entryStepIds: Set<string>[] = entries.map((e) => new Set(e.steps.map((s) => s.stepId))); - - // Match segments to entries. Pass 1: match by stepId overlap (handles - // trimming where positional alignment alone could be ambiguous). Pass 2: - // positional tail-alignment fallback for unmatched segments (text-only turns - // with no stepId-bearing groups). - const usedEntries = new Set<number>(); - const segmentEntry = new Map<number, TurnMetricsEntry>(); - const segmentEntryIndex = new Map<number, number>(); - - // Pass 1: stepId matching. - for (let seg = 0; seg < T; seg++) { - const start = segmentStarts[seg] ?? 0; - const end = seg + 1 < T ? (segmentStarts[seg + 1] ?? groups.length) : groups.length; - - const segStepIds = new Set<string>(); - for (let i = start; i < end; i++) { - const g = groups[i]; - if (g === undefined) continue; - const sid = groupStepId(g); - if (sid !== undefined) segStepIds.add(sid); + let k = 0; + while (k < segments.length) { + if (segTurn[k] !== undefined) { + k++; + continue; } - if (segStepIds.size === 0) continue; // text-only — defer to pass 2 + let runEnd = k; + while (runEnd < segments.length && segTurn[runEnd] === undefined) runEnd++; + const leading = k === 0; + const lower = leading ? -1 : segEntryIdx(k - 1); + const upper = runEnd < segments.length ? segEntryIdx(runEnd) : entries.length; + const candidates: number[] = []; + for (let e = lower + 1; e < upper; e++) if (!claimed.has(e)) candidates.push(e); - let bestEntry = -1; - let bestMatch = 0; - for (let i = 0; i < K; i++) { - if (usedEntries.has(i)) continue; - let match = 0; - for (const sid of segStepIds) { - if (entryStepIds[i]?.has(sid)) match++; - } - if (match > bestMatch) { - bestMatch = match; - bestEntry = i; - } + let run: number[] = []; + for (let s = k; s < runEnd; s++) run.push(s); + if (run.length !== candidates.length) { + // A segment with no model output (only the user message, or an error) is + // a turn that failed before producing anything — or one just sent, whose + // turn hasn't started yet — and legitimately has no metrics. + const responsive = run.filter((s) => segments[s]?.hasResponse === true); + if (responsive.length <= candidates.length) run = responsive; } - if (bestEntry >= 0) { - usedEntries.add(bestEntry); - const e = entries[bestEntry]; - if (e !== undefined) { - segmentEntry.set(seg, e); - segmentEntryIndex.set(seg, bestEntry); - } + let pick: number[] | null = null; + if (run.length === candidates.length) pick = candidates; + else if (leading && run.length < candidates.length) pick = candidates.slice(-run.length); + if (pick !== null) { + run.forEach((s, i) => { + const e = pick?.[i]; + const entry = e !== undefined ? entries[e] : undefined; + if (entry !== undefined && e !== undefined) { + segTurn[s] = entry.turnId; + claimed.add(e); + } + }); } + k = runEnd; } - // Pass 2: positional fallback for segments pass 1 left unmatched - // (text-only turns with no stepId-bearing groups to anchor on). - // - // The loaded transcript is always a SUFFIX of the full turn history — - // chat-limit/windowing keeps the NEWEST chunks and unloads the OLDEST — so - // the T loaded segments correspond to the LAST T entries. TAIL-ALIGNMENT - // (segment `seg` ↔ entry `K - T + seg`) is therefore correct whenever the - // metrics hold at least as many turns as there are loaded segments - // (`K >= T`): the leading `K - T` entries are TRIMMED turns (their content - // was unloaded) and must be skipped, never matched to a newer segment. - // - // This MUST run even when pass 1 matched SOME segments (tool turns). The - // earlier code only tail-aligned when pass 1 matched NONE, falling back to - // HEAD-alignment otherwise — which, with leading trimmed entries, matched a - // brand-new text-only turn to an old (trimmed) entry's STALE metrics (the - // "new steps show no / wrong cache" failure). Tail-aligning by position is - // safe alongside pass 1: stepIds are unique per turn, so pass 1 already - // grabbed each tool turn's positionally-correct entry, leaving the right - // entry free for each text-only turn. - // - // Only when `K < T` (fewer entries than segments — some loaded turns have no - // metrics yet, e.g. a metrics sync still pending or a freshly loaded - // transcript) do we head-align, assigning the first K entries to the first K - // unmatched segments (the turns that DO have metrics sit at the front). - if (K >= T) { - // Tail-align: skip the first K-T entries (trimmed turns). - for (let seg = 0; seg < T; seg++) { - if (segmentEntry.has(seg)) continue; - const entryIdx = K - T + seg; - if (entryIdx >= 0 && entryIdx < K && !usedEntries.has(entryIdx)) { - usedEntries.add(entryIdx); - const e = entries[entryIdx]; - if (e !== undefined) { - segmentEntry.set(seg, e); - segmentEntryIndex.set(seg, entryIdx); - } - } - } - } else { - // Head-align fallback (K < T): first K entries to first K unmatched segments. - let nextUnused = 0; - for (let seg = 0; seg < T; seg++) { - if (segmentEntry.has(seg)) continue; - while (nextUnused < K && usedEntries.has(nextUnused)) nextUnused++; - if (nextUnused < K) { - usedEntries.add(nextUnused); - const e = entries[nextUnused]; - if (e !== undefined) { - segmentEntry.set(seg, e); - segmentEntryIndex.set(seg, nextUnused); - } - nextUnused++; - } - } + const out = [...direct]; + segments.forEach((s, si) => { + const t = segTurn[si]; + if (t === undefined) return; + for (let i = s.start; i < s.end; i++) if (out[i] === undefined) out[i] = t; + }); + return out; +} + +/** + * Interleave per-turn / per-step metrics into the rendered transcript. + * + * Every group is attributed to the turn that produced it (see `assignTurns` — + * exact via `turnId` provenance; a guarded order-preserving fallback only for + * legacy rows). Then: + * - each completed step's metrics row goes right after the LAST group bearing + * that step's `stepId` (steps with no visible content — text-only steps, or + * content unloaded by the chat limit — get no row); + * - a finalized turn's (`entry.total !== null`) metrics row goes right after + * the LAST group of that turn — so a turn whose user message was unloaded + * still shows its total under its visible tail, and a mid-turn steering + * message never pulls the total up. + * + * Because attribution never depends on counting user messages, sending a new + * message, a new turn streaming, steering, image attachments, errored turns, or + * chat-limit unloading can never shift one turn's metrics onto another. + * + * Turns whose content is fully unloaded are not rendered (no wall of stale + * rows at the top) but still count toward the running "chat total". + */ +export function interleaveTurnMetrics( + groups: readonly RenderGroup[], + entries: readonly TurnMetricsEntry[], +): readonly MetricsRow[] { + if (entries.length === 0) { + return groups.map((g) => ({ kind: "group" as const, group: g })); } - // Running cumulative usage across ALL finalized turns (in entry order), for - // the per-turn "chat total" cache rate. Alongside it, the previous finalized - // turn's usage at each index — the baseline for cross-turn retention. + const turnOf = assignTurns(groups, entries); + + const entryByTurn = new Map<string, { entry: TurnMetricsEntry; index: number }>(); + entries.forEach((entry, index) => { + entryByTurn.set(entry.turnId, { entry, index }); + }); + + // Running cumulative usage across ALL finalized turns (entry order), and the + // previous finalized turn's LAST step usage (the retention baseline: the + // prompt that turn last sent, which the next turn's first step should hit in + // cache). const cumulativeByEntry: Usage[] = []; - const prevUsageByEntry: (Usage | null)[] = []; + const prevLastStepByEntry: (Usage | null)[] = []; let runningUsage: Usage = { inputTokens: 0, outputTokens: 0 }; - let lastFinalizedUsage: Usage | null = null; + let lastFinalizedStep: Usage | null = null; for (const e of entries) { - prevUsageByEntry.push(lastFinalizedUsage); + prevLastStepByEntry.push(lastFinalizedStep); if (e.total !== null) { runningUsage = addUsage(runningUsage, e.total.usage); - lastFinalizedUsage = e.total.usage; + const last = e.total.steps[e.total.steps.length - 1]; + if (last !== undefined) lastFinalizedStep = last.usage; } cumulativeByEntry.push(runningUsage); } - const rows: MetricsRow[] = []; - - const firstUserIdx = segmentStarts[0] ?? 0; - - for (let i = 0; i < firstUserIdx; i++) { - const g = groups[i]; - if (g !== undefined) { - rows.push({ kind: "group", group: g }); - } - } - - for (let seg = 0; seg < T; seg++) { - const start = segmentStarts[seg] ?? 0; - const end = seg + 1 < T ? (segmentStarts[seg + 1] ?? groups.length) : groups.length; - - const entry = segmentEntry.get(seg); + // Anchor positions: last group of each step (within its turn) / each turn. + const lastGroupOfStep = new Map<string, number>(); + const lastGroupOfTurn = new Map<string, number>(); + groups.forEach((g, i) => { + const t = turnOf[i]; + if (t === undefined) return; + lastGroupOfTurn.set(t, i); + const sid = groupStepId(g); + if (sid !== undefined) lastGroupOfStep.set(sid, i); + }); - if (entry === undefined) { - for (let i = start; i < end; i++) { - const g = groups[i]; - if (g !== undefined) { - rows.push({ kind: "group", group: g }); - } - } - continue; - } - - const entryIdx = segmentEntryIndex.get(seg) ?? 0; - - // Build anchor map: for each stepId, the LAST group index in this segment. - const anchorByStepId = new Map<string, number>(); - for (let i = start; i < end; i++) { - const g = groups[i]; - if (g === undefined) continue; - const sid = groupStepId(g); - if (sid !== undefined) { - anchorByStepId.set(sid, i); + const stepsAt = new Map<number, { step: StepMetrics; index: number }[]>(); + for (const entry of entries) { + entry.steps.forEach((step, index) => { + const at = lastGroupOfStep.get(step.stepId); + if (at === undefined || turnOf[at] !== entry.turnId) return; + let arr = stepsAt.get(at); + if (arr === undefined) { + arr = []; + stepsAt.set(at, arr); } - } - - // Classify each step as anchored or unanchored. Unanchored steps - // (content trimmed, or text-only steps with no tool chunks) are SKIPPED — - // step-metrics are only shown inline next to the content they describe. - const anchored: Map<number, { stepIndex: number; step: (typeof entry.steps)[number] }[]> = - new Map(); + arr.push({ step, index }); + }); + } - for (let i = 0; i < entry.steps.length; i++) { - const step = entry.steps[i]; - if (step === undefined) continue; - const anchorGroupIdx = anchorByStepId.get(step.stepId); - if (anchorGroupIdx !== undefined) { - let arr = anchored.get(anchorGroupIdx); - if (arr === undefined) { - arr = []; - anchored.set(anchorGroupIdx, arr); - } - arr.push({ stepIndex: i, step }); - } - // Unanchored steps (no matching group) are skipped — no tail bubbles. - } + const rows: MetricsRow[] = []; + for (let i = 0; i < groups.length; i++) { + const g = groups[i]; + if (g === undefined) continue; + rows.push({ kind: "group", group: g }); - // Emit groups; after each anchored group, emit its step-metrics rows. - for (let i = start; i < end; i++) { - const g = groups[i]; - if (g !== undefined) { - rows.push({ kind: "group", group: g }); - } - const stepsHere = anchored.get(i); - if (stepsHere !== undefined) { - stepsHere.sort((a, b) => a.stepIndex - b.stepIndex); - for (const { step, stepIndex } of stepsHere) { - rows.push({ kind: "step-metrics", step, index: stepIndex }); - } - } + const steps = stepsAt.get(i); + if (steps !== undefined) { + steps.sort((a, b) => a.index - b.index); + for (const { step, index } of steps) rows.push({ kind: "step-metrics", step, index }); } - // Turn-metrics row (only when the turn is finalized). Unanchored steps - // are skipped — no tail bubbles. - if (entry.total !== null) { - rows.push({ - kind: "turn-metrics", - turn: entry.total, - turnNumber: entryIdx + 1, - cumulativeUsage: cumulativeByEntry[entryIdx] ?? entry.total.usage, - prevTurnUsage: prevUsageByEntry[entryIdx] ?? null, - }); - } + const t = turnOf[i]; + if (t === undefined || lastGroupOfTurn.get(t) !== i) continue; + const found = entryByTurn.get(t); + if (found === undefined || found.entry.total === null) continue; + rows.push({ + kind: "turn-metrics", + turn: found.entry.total, + turnNumber: found.index + 1, + cumulativeUsage: cumulativeByEntry[found.index] ?? found.entry.total.usage, + prevLastStepUsage: prevLastStepByEntry[found.index] ?? null, + }); } return rows; diff --git a/src/core/metrics/placement-scenarios.test.ts b/src/core/metrics/placement-scenarios.test.ts new file mode 100644 index 0000000..ce3ed5b --- /dev/null +++ b/src/core/metrics/placement-scenarios.test.ts @@ -0,0 +1,376 @@ +// End-to-end placement scenarios: real transcript + metrics reducers fed real +// AgentEvents, rendered through groupRenderedChunks → interleaveTurnMetrics. +// The invariant under test: at EVERY point of a turn (not just after it ends), +// each turn's metrics sit under that turn — never under a neighbour. +import type { AgentEvent, StepId, StoredChunk, TurnMetrics, Usage } from "@dispatch/wire"; +import { describe, expect, it } from "vitest"; +import { + appendUserMessage, + applyHistory, + foldEvent, + groupRenderedChunks, + initialState, + selectChunks, + type TranscriptState, + trimTranscript, +} from "../chunks"; +import { interleaveTurnMetrics } from "./place"; +import { + applyDurableMetrics, + foldMetricsEvent, + initialMetricsState, + selectOrderedTurnMetrics, +} from "./reducer"; +import type { MetricsRow, MetricsState } from "./types"; + +const C = "c1"; +const u = (i: number, o: number): Usage => ({ inputTokens: i, outputTokens: o }); + +/** Ordered list of "what follows each user text bubble / which turn rows appear where". */ +function layout(rows: readonly MetricsRow[]): string[] { + const out: string[] = []; + for (const r of rows) { + if (r.kind === "turn-metrics") out.push(`[turn ${r.turn.turnId}]`); + else if (r.kind === "step-metrics") out.push(`[step ${r.step.stepId}]`); + else if (r.group.kind === "single") { + const c = r.group.chunk.chunk; + if (c.type === "text") out.push(`${r.group.chunk.role}:${c.text}`); + else if (c.type === "image") out.push(`${r.group.chunk.role}:<img>`); + else if (c.type === "error") out.push(`error:${c.message}`); + } else { + out.push(`tool:${r.group.entry.toolCallId}`); + } + } + return out; +} + +class Sim { + t: TranscriptState = initialState(); + m: MetricsState = initialMetricsState(); + seq = 0; + /** Durable server-side log (what loadSince returns). */ + log: StoredChunk[] = []; + durable: TurnMetrics[] = []; + + constructor(private readonly legacy = false) {} + + private ev(e: AgentEvent): void { + this.t = foldEvent(this.t, e); + this.m = foldMetricsEvent(this.m, e); + } + + private persist(role: StoredChunk["role"], chunk: StoredChunk["chunk"], turnId: string): void { + this.seq++; + this.log.push( + this.legacy ? { seq: this.seq, role, chunk } : { seq: this.seq, role, chunk, turnId }, + ); + } + + /** The sender types + sends (optimistic echo), optionally with an image. */ + send(text: string, image = false): void { + this.t = appendUserMessage( + this.t, + text, + image ? [{ url: "data:image/png;base64,AA", mimeType: "image/png" }] : undefined, + ); + } + + start(turnId: string, text: string, image = false): void { + this.ev({ type: "turn-start", conversationId: C, turnId } as AgentEvent); + this.ev({ type: "user-message", conversationId: C, turnId, text } as AgentEvent); + this.persist("user", { type: "text", text }, turnId); + if (image) this.persist("user", { type: "image", url: "/images/c1/x.png" }, turnId); + } + + text(turnId: string, text: string): void { + this.ev({ type: "text-delta", conversationId: C, turnId, delta: text } as AgentEvent); + } + + toolStep(turnId: string, step: number, id: string): void { + const stepId = `${turnId}#${step}` as StepId; + this.ev({ + type: "tool-call", + conversationId: C, + turnId, + stepId, + toolCallId: id, + toolName: "t", + input: {}, + } as AgentEvent); + this.completeStep(turnId, step); + this.ev({ + type: "tool-result", + conversationId: C, + turnId, + stepId, + toolCallId: id, + toolName: "t", + content: "ok", + isError: false, + } as AgentEvent); + this.persist( + "assistant", + { type: "tool-call", toolCallId: id, toolName: "t", input: {}, stepId }, + turnId, + ); + this.persist( + "tool", + { type: "tool-result", toolCallId: id, toolName: "t", content: "ok", isError: false, stepId }, + turnId, + ); + } + + completeStep(turnId: string, step: number): void { + const stepId = `${turnId}#${step}` as StepId; + this.ev({ type: "usage", conversationId: C, turnId, stepId, usage: u(100, 10) } as AgentEvent); + this.ev({ + type: "step-complete", + conversationId: C, + turnId, + stepId, + ttftMs: 100, + decodeMs: 100, + genTotalMs: 200, + } as AgentEvent); + } + + steer(turnId: string, text: string): void { + this.ev({ type: "steering", conversationId: C, turnId, text } as AgentEvent); + this.persist("user", { type: "text", text }, turnId); + } + + /** Finish a turn: final text step, done, seal, then history + metrics sync. */ + finish(turnId: string, text: string, finalStep: number): void { + this.text(turnId, text); + this.completeStep(turnId, finalStep); + this.ev({ + type: "done", + conversationId: C, + turnId, + reason: "stop", + durationMs: 500, + usage: u(100, 10), + } as AgentEvent); + this.persist("assistant", { type: "text", text }, turnId); + this.ev({ type: "turn-sealed", conversationId: C, turnId } as AgentEvent); + this.sync(turnId, finalStep); + } + + /** A turn that errors (never seals); backend persists its partial metrics. */ + fail(turnId: string, message: string, steps: number): void { + this.ev({ type: "error", conversationId: C, turnId, message } as AgentEvent); + this.persist("assistant", { type: "error", message }, turnId); + this.t = applyHistory(this.t, this.log); + if (steps > 0) this.syncMetricsFor(turnId, steps - 1); + } + + private syncMetricsFor(turnId: string, lastStep: number): void { + const steps = []; + for (let i = 0; i <= lastStep; i++) { + steps.push({ stepId: `${turnId}#${i}` as StepId, usage: u(100, 10), decodeMs: 100 }); + } + this.durable.push({ turnId, usage: u(100 * steps.length, 10 * steps.length), steps }); + this.m = applyDurableMetrics(this.m, this.durable); + } + + private sync(turnId: string, lastStep: number): void { + this.t = applyHistory(this.t, this.log); + this.syncMetricsFor(turnId, lastStep); + } + + trim(limit: number): void { + this.t = trimTranscript(this.t, limit); + } + + layout(): string[] { + const groups = groupRenderedChunks(selectChunks(this.t)); + return layout(interleaveTurnMetrics(groups, selectOrderedTurnMetrics(this.m))); + } +} + +/** Build a history of N text-only turns (t1..tN), each "qK" → "aK". */ +function textHistory(sim: Sim, n: number): void { + for (let k = 1; k <= n; k++) { + sim.send(`q${k}`); + sim.start(`t${k}`, `q${k}`); + sim.finish(`t${k}`, `a${k}`, 0); + } +} + +for (const legacy of [false, true]) { + // `legacy`: the persisted rows carry NO turnId (history written before the + // backend stamped it). Live chunks always carry it. + describe(`placement stays exact through a turn (${legacy ? "legacy rows" : "turnId rows"})`, () => { + it("plain text turns", () => { + const sim = new Sim(legacy); + textHistory(sim, 3); + expect(sim.layout()).toEqual([ + "user:q1", + "assistant:a1", + "[turn t1]", + "user:q2", + "assistant:a2", + "[turn t2]", + "user:q3", + "assistant:a3", + "[turn t3]", + ]); + }); + + it("never shifts while a new turn is sent and streams — even after older turns were unloaded", () => { + const sim = new Sim(legacy); + textHistory(sim, 4); + sim.trim(5); // 8 chunks > 5 → unloads whole quarters: q1 a1 q2 a2 + const before = sim.layout(); + expect(before).toEqual([ + "user:q3", + "assistant:a3", + "[turn t3]", + "user:q4", + "assistant:a4", + "[turn t4]", + ]); + + const expectStable = () => expect(sim.layout().slice(0, before.length)).toEqual(before); + + sim.send("q5"); + expectStable(); // just sent (optimistic echo, no turn yet) + sim.start("t5", "q5"); + expectStable(); // turn started + sim.toolStep("t5", 0, "k1"); + expectStable(); // tool step completed (metrics entry exists, not done) + sim.text("t5", "partial…"); + expectStable(); // streaming text + sim.finish("t5", "a5", 1); + expect(sim.layout()).toEqual([ + ...before, + "user:q5", + "tool:k1", + "[step t5#0]", + "assistant:a5", + "[turn t5]", + ]); + }); + + it("image attachments never shift other turns' metrics", () => { + const sim = new Sim(legacy); + sim.send("q1"); + sim.start("t1", "q1"); + sim.finish("t1", "a1", 0); + sim.send("q2", true); + sim.start("t2", "q2", true); + sim.finish("t2", "a2", 0); + sim.send("q3"); + sim.start("t3", "q3"); + sim.finish("t3", "a3", 0); + expect(sim.layout()).toEqual([ + "user:q1", + "assistant:a1", + "[turn t1]", + "user:q2", + "user:<img>", + "assistant:a2", + "[turn t2]", + "user:q3", + "assistant:a3", + "[turn t3]", + ]); + }); + + it("a mid-turn steering message keeps the turn total at the END of the turn", () => { + const sim = new Sim(legacy); + textHistory(sim, 1); + sim.send("q2"); + sim.start("t2", "q2"); + sim.toolStep("t2", 0, "k1"); + sim.steer("t2", "steer!"); + // live, mid-turn: t1 untouched, steering bubble shows no totals + expect(sim.layout()).toEqual([ + "user:q1", + "assistant:a1", + "[turn t1]", + "user:q2", + "tool:k1", + "[step t2#0]", + "user:steer!", + ]); + sim.toolStep("t2", 1, "k2"); + sim.finish("t2", "a2", 2); + textHistoryFrom(sim, 3); + expect(sim.layout()).toEqual([ + "user:q1", + "assistant:a1", + "[turn t1]", + "user:q2", + "tool:k1", + "[step t2#0]", + "user:steer!", + "tool:k2", + "[step t2#1]", + "assistant:a2", + "[turn t2]", + "user:q3", + "assistant:a3", + "[turn t3]", + ]); + }); + + it("an errored turn gets its own metrics and never shifts the others", () => { + const sim = new Sim(legacy); + textHistory(sim, 1); + sim.send("q2"); + sim.start("t2", "q2"); + sim.text("t2", "partial"); + sim.completeStep("t2", 0); + sim.fail("t2", "boom", 1); + textHistoryFrom(sim, 3); + const l = sim.layout(); + expect(l.indexOf("[turn t1]")).toBe(l.indexOf("assistant:a1") + 1); + expect(l.indexOf("[turn t2]")).toBeGreaterThan(l.indexOf("error:boom")); + expect(l.indexOf("[turn t2]")).toBeLessThan(l.indexOf("user:q3")); + expect(l.indexOf("[turn t3]")).toBe(l.indexOf("assistant:a3") + 1); + }); + + it("an errored turn with NO metrics never steals another turn's", () => { + const sim = new Sim(legacy); + textHistory(sim, 1); + sim.send("q2"); + sim.start("t2", "q2"); + sim.fail("t2", "boom", 0); // failed before any step → no metrics row + textHistoryFrom(sim, 3); + expect(sim.layout()).toEqual([ + "user:q1", + "assistant:a1", + "[turn t1]", + "user:q2", + "error:boom", + "user:q3", + "assistant:a3", + "[turn t3]", + ]); + }); + + it("a turn cut in half by the chat limit still shows its total under its visible tail", () => { + const sim = new Sim(legacy); + sim.send("q1"); + sim.start("t1", "q1"); + sim.toolStep("t1", 0, "k1"); + sim.toolStep("t1", 1, "k2"); + sim.finish("t1", "a1", 2); + textHistoryFrom(sim, 2); + // log: q1 call result call result a1 | q2 a2 → keep newest 6 + sim.trim(7); + const l = sim.layout(); + expect(l).not.toContain("user:q1"); + expect(l.indexOf("[turn t1]")).toBe(l.indexOf("assistant:a1") + 1); + expect(l.indexOf("[turn t2]")).toBe(l.indexOf("assistant:a2") + 1); + }); + }); +} + +/** Continue a text history from turn `from` (one more turn). */ +function textHistoryFrom(sim: Sim, from: number): void { + sim.send(`q${from}`); + sim.start(`t${from}`, `q${from}`); + sim.finish(`t${from}`, `a${from}`, 0); +} diff --git a/src/core/metrics/types.ts b/src/core/metrics/types.ts index 84d1904..95b5b02 100644 --- a/src/core/metrics/types.ts +++ b/src/core/metrics/types.ts @@ -61,10 +61,11 @@ export type MetricsRow = /** Cumulative usage across all finalized turns up to and including this one. */ readonly cumulativeUsage: Usage; /** - * Usage of the most recent EARLIER finalized turn, or `null` when this is the - * first finalized turn. The baseline for cross-turn retention (expected cache). + * Usage of the LAST step of the most recent EARLIER finalized turn, or + * `null` when there is none. The baseline for cross-turn retention: that + * step's prompt is what this turn's first step should find in cache. */ - readonly prevTurnUsage: Usage | null; + readonly prevLastStepUsage: Usage | null; }; /** Formatted cache hit-rate view: percentage + colour severity + hit flag. */ @@ -80,6 +81,7 @@ export interface CacheRateView { /** Formatted per-step view for display. */ export interface StepMetricsView { readonly label: string; + /** `"<in> in / <out> out"` — this step's prompt and generated tokens. */ readonly tokensLabel: string; readonly tps: string | null; readonly ttft: string | null; @@ -90,7 +92,13 @@ export interface StepMetricsView { /** Formatted per-turn view for display. */ export interface TurnMetricsView { readonly label: string; + /** `"<out> tok out"` — tokens the model generated this turn. */ readonly tokensLabel: string; + /** + * Prompt-side breakdown: input tokens processed (summed over the turn's + * steps — each step re-sends the whole prompt, so this is NOT the context + * size) and how many of them were cache reads. + */ readonly breakdown: string; readonly tps: string | null; readonly duration: string | null; |