diff options
| author | Adam Malczewski <[email protected]> | 2026-10-10 00:06:31 +0900 |
|---|---|---|
| committer | Adam Malczewski <[email protected]> | 2026-10-10 00:06:31 +0900 |
| commit | 9f2dcfaab421fa84837685a0b65678acf03d9ded (patch) | |
| tree | a1ef40fbbf98b136a471889c5b02795b739753be /src/core | |
| parent | 591e2dbe5cbf2283d58debdf92943a075edc4569 (diff) | |
| download | dispatch-web-9f2dcfaab421fa84837685a0b65678acf03d9ded.tar.gz dispatch-web-9f2dcfaab421fa84837685a0b65678acf03d9ded.zip | |
fix(turn-metrics): exact per-turn placement via turnId, decode-only TPS, step-level retention
Placement
- Chunks carry their producing turn: StoredChunk.turnId (backend [email protected]) for
committed rows (persisted in the IDB cache; old cached rows are upgraded on
sync) and the live event's turnId for provisional ones (the optimistic echo is
tagged on turn-start).
- interleaveTurnMetrics attributes every group to its turn (turnId, else the
turn owning its stepId) and puts the turn total after the turn's LAST visible
group. No user-message counting: sending, streaming, steering, images, errored
turns and chat-limit unloading can no longer shift metrics between turns — at
any point during a turn, not just after it ends. A turn cut in half by the
chat limit keeps its total under its visible tail.
- Legacy rows (no turnId) fall back to an order-preserving, count-guarded match
that leaves a turn blank rather than show another turn's numbers.
- A turn that errors re-syncs metrics (the backend now persists them).
TPS
- Step TPS = outputTokens / decodeMs only (no genTotalMs fallback — it includes
time-to-first-token). Turn TPS sums tokens and decode time over timed steps
only. Tool execution is excluded by the backend.
Display
- Retention = this turn's first-step cache reads / previous turn's last-step
prompt (step-level; turn sums made it pin at 100%).
- Turn row shows output tokens + input processed (with cached), instead of a
misleading 'total tok' that summed every step's full prompt.
Diffstat (limited to 'src/core')
| -rw-r--r-- | src/core/chunks/groups.ts | 10 | ||||
| -rw-r--r-- | src/core/chunks/reducer.test.ts | 45 | ||||
| -rw-r--r-- | src/core/chunks/reducer.ts | 65 | ||||
| -rw-r--r-- | src/core/chunks/selectors.ts | 28 | ||||
| -rw-r--r-- | src/core/chunks/types.ts | 15 | ||||
| -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/place.test.ts | 45 | ||||
| -rw-r--r-- | src/core/metrics/place.ts | 456 | ||||
| -rw-r--r-- | src/core/metrics/placement-scenarios.test.ts | 380 | ||||
| -rw-r--r-- | src/core/metrics/types.ts | 14 |
11 files changed, 883 insertions, 352 deletions
diff --git a/src/core/chunks/groups.ts b/src/core/chunks/groups.ts index c1c80c9..9df0fa5 100644 --- a/src/core/chunks/groups.ts +++ b/src/core/chunks/groups.ts @@ -22,6 +22,8 @@ export type RenderGroup = readonly stepId: string; readonly entries: readonly ToolBatchEntry[]; readonly provisional: boolean; + /** The producing turn (from the batch's first call chunk), when known. */ + readonly turnId?: string; }; /** @@ -79,6 +81,7 @@ export function groupRenderedChunks(rendered: readonly RenderedChunk[]): readonl // (Svelte `each_key_duplicate` aborts the reactive flush — tab appears dead). const seenCallIds = new Set<string>(); let provisional = false; + let turnId: string | undefined; for (const inner of rendered) { if (inner.chunk.type === "tool-call" && inner.chunk.stepId === stepId) { if (seenCallIds.has(inner.chunk.toolCallId)) continue; @@ -86,9 +89,14 @@ export function groupRenderedChunks(rendered: readonly RenderedChunk[]): readonl const result = resultByCallId.get(inner.chunk.toolCallId) ?? null; entries.push({ call: inner.chunk, result }); if (inner.provisional) provisional = true; + turnId ??= inner.turnId; } } - groups.push({ kind: "tool-batch", stepId, entries, provisional }); + groups.push( + turnId !== undefined + ? { kind: "tool-batch", stepId, entries, provisional, turnId } + : { kind: "tool-batch", stepId, entries, provisional }, + ); continue; } diff --git a/src/core/chunks/reducer.test.ts b/src/core/chunks/reducer.test.ts index 058552e..cba6fad 100644 --- a/src/core/chunks/reducer.test.ts +++ b/src/core/chunks/reducer.test.ts @@ -226,7 +226,7 @@ describe("clearGenerating", () => { s = foldEvent(s, textDelta("t1", "partial")); const cleared = clearGenerating(s); expect(cleared.generating).toBe(false); - expect(cleared.accumulating).toEqual({ kind: "text", text: "partial" }); + expect(cleared.accumulating).toEqual({ kind: "text", text: "partial", turnId: "t1" }); expect(cleared.currentTurnId).toBe("t1"); }); }); @@ -236,7 +236,7 @@ describe("foldEvent — text-delta", () => { let s = initialState(); s = foldEvent(s, turnStart("t1")); s = foldEvent(s, textDelta("t1", "hello")); - expect(s.accumulating).toEqual({ kind: "text", text: "hello" }); + expect(s.accumulating).toEqual({ kind: "text", text: "hello", turnId: "t1" }); expect(s.provisional).toEqual([]); }); @@ -245,7 +245,7 @@ describe("foldEvent — text-delta", () => { s = foldEvent(s, turnStart("t1")); s = foldEvent(s, textDelta("t1", "hello ")); s = foldEvent(s, textDelta("t1", "world")); - expect(s.accumulating).toEqual({ kind: "text", text: "hello world" }); + expect(s.accumulating).toEqual({ kind: "text", text: "hello world", turnId: "t1" }); expect(s.provisional).toEqual([]); }); @@ -254,7 +254,7 @@ describe("foldEvent — text-delta", () => { s = foldEvent(s, turnStart("t1")); s = foldEvent(s, reasoningDelta("t1", "thinking...")); s = foldEvent(s, textDelta("t1", "answer")); - expect(s.accumulating).toEqual({ kind: "text", text: "answer" }); + expect(s.accumulating).toEqual({ kind: "text", text: "answer", turnId: "t1" }); expect(s.provisional).toHaveLength(1); expect(s.provisional[0]?.chunk).toEqual({ type: "thinking", text: "thinking..." }); expect(s.provisional[0]?.role).toBe("assistant"); @@ -266,7 +266,7 @@ describe("foldEvent — reasoning-delta", () => { let s = initialState(); s = foldEvent(s, turnStart("t1")); s = foldEvent(s, reasoningDelta("t1", "hmm")); - expect(s.accumulating).toEqual({ kind: "thinking", text: "hmm" }); + expect(s.accumulating).toEqual({ kind: "thinking", text: "hmm", turnId: "t1" }); }); it("successive reasoning-deltas extend the same chunk", () => { @@ -274,7 +274,7 @@ describe("foldEvent — reasoning-delta", () => { s = foldEvent(s, turnStart("t1")); s = foldEvent(s, reasoningDelta("t1", "hmm ")); s = foldEvent(s, reasoningDelta("t1", "ok")); - expect(s.accumulating).toEqual({ kind: "thinking", text: "hmm ok" }); + expect(s.accumulating).toEqual({ kind: "thinking", text: "hmm ok", turnId: "t1" }); }); }); @@ -531,7 +531,7 @@ describe("clearGenerating also clears a stale provider-retry banner (reconnect)" s = foldEvent(s, textDelta("t1", "partial")); s = foldEvent(s, providerRetry("t1", 0, 5000)); const cleared = clearGenerating(s); - expect(cleared.accumulating).toEqual({ kind: "text", text: "partial" }); + expect(cleared.accumulating).toEqual({ kind: "text", text: "partial", turnId: "t1" }); expect(selectProviderRetry(cleared)).toBeNull(); }); }); @@ -859,7 +859,7 @@ describe("appendUserMessage", () => { let s = initialState(); s = foldEvent(s, turnStart("t1")); s = foldEvent(s, textDelta("t1", "partial")); - expect(s.accumulating).toEqual({ kind: "text", text: "partial" }); + expect(s.accumulating).toEqual({ kind: "text", text: "partial", turnId: "t1" }); s = appendUserMessage(s, "user msg"); expect(s.accumulating).toBeNull(); @@ -1022,3 +1022,32 @@ describe("applyHistory — multi-chunk image echo is superseded by committed", ( expect(msgs[0]?.chunks[1]).toEqual({ type: "image", url: PNG, mimeType: "image/png" }); }); }); + +describe("turnId provenance on live chunks", () => { + it("tags every live chunk with its event's turnId, and the optimistic echo on turn-start", () => { + let s = initialState(); + s = appendUserMessage(s, "q", [{ url: "data:image/png;base64,AA", mimeType: "image/png" }]); + expect(selectChunks(s).map((c) => c.turnId)).toEqual([undefined, undefined]); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "a")); + const chunks = selectChunks(s); + expect(chunks.map((c) => c.turnId)).toEqual(["t1", "t1", "t1"]); + }); + + it("turn-start does not re-tag an echo already owned by an earlier turn", () => { + let s = initialState(); + s = appendUserMessage(s, "q1"); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "a1")); + s = foldEvent(s, turnStart("t2")); + expect(selectChunks(s).map((c) => c.turnId)).toEqual(["t1", "t1"]); + }); + + it("committed chunks expose StoredChunk.turnId", () => { + const s = applyHistory(initialState(), [ + { seq: 1, role: "user", chunk: { type: "text", text: "q" }, turnId: "t9" }, + { seq: 2, role: "assistant", chunk: { type: "text", text: "a" } }, + ]); + expect(selectChunks(s).map((c) => c.turnId)).toEqual(["t9", undefined]); + }); +}); diff --git a/src/core/chunks/reducer.ts b/src/core/chunks/reducer.ts index 64e41b9..4eb08c1 100644 --- a/src/core/chunks/reducer.ts +++ b/src/core/chunks/reducer.ts @@ -40,7 +40,33 @@ function flushAccumulating( if (acc === null) return provisional; const chunk: Chunk = acc.kind === "text" ? { type: "text", text: acc.text } : { type: "thinking", text: acc.text }; - return [...provisional, { role: "assistant", chunk }]; + return [...provisional, withTurn({ role: "assistant", chunk }, acc.turnId)]; +} + +/** Attach a `turnId` to a provisional chunk (omitted when unknown). */ +function withTurn(p: ProvisionalChunk, turnId: string | undefined): ProvisionalChunk { + return turnId !== undefined ? { ...p, turnId } : p; +} + +/** + * Tag the trailing UNTAGGED user run (the sender's optimistic echo, appended + * before the server assigned a turn id) with the turn that just started. + */ +function tagTrailingUserEcho( + provisional: readonly ProvisionalChunk[], + turnId: string, +): readonly ProvisionalChunk[] { + let start = provisional.length; + while (start > 0) { + const p = provisional[start - 1]; + if (p === undefined || p.role !== "user" || p.turnId !== undefined) break; + start--; + } + if (start === provisional.length) return provisional; + return [ + ...provisional.slice(0, start), + ...provisional.slice(start).map((p) => ({ ...p, turnId })), + ]; } /** @@ -199,7 +225,12 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState return state; case "turn-start": - return { ...state, currentTurnId: event.turnId, generating: true }; + return { + ...state, + provisional: tagTrailingUserEcho(state.provisional, event.turnId), + currentTurnId: event.turnId, + generating: true, + }; case "user-message": { // The turn's USER prompt, surfaced on the event stream (backend CR-3) so a @@ -217,12 +248,19 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState const run = trailingRun(state.provisional, "user"); const alreadyEchoed = run.some((p) => p.chunk.type === "text" && p.chunk.text === event.text); if (alreadyEchoed) { - return { ...state, generating: true }; + return { + ...state, + provisional: tagTrailingUserEcho(state.provisional, event.turnId), + generating: true, + }; } const provisional = flushAccumulating(state.provisional, state.accumulating); return { ...state, - provisional: [...provisional, { role: "user", chunk: { type: "text", text: event.text } }], + provisional: [ + ...provisional, + { role: "user", chunk: { type: "text", text: event.text }, turnId: event.turnId }, + ], accumulating: null, generating: true, }; @@ -233,7 +271,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState if (acc !== null && acc.kind === "text") { return { ...state, - accumulating: { kind: "text", text: acc.text + event.delta }, + accumulating: { ...acc, text: acc.text + event.delta }, generating: true, }; } @@ -241,7 +279,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState return { ...state, provisional, - accumulating: { kind: "text", text: event.delta }, + accumulating: { kind: "text", text: event.delta, turnId: event.turnId }, generating: true, }; } @@ -251,7 +289,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState if (acc !== null && acc.kind === "thinking") { return { ...state, - accumulating: { kind: "thinking", text: acc.text + event.delta }, + accumulating: { ...acc, text: acc.text + event.delta }, generating: true, }; } @@ -259,7 +297,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState return { ...state, provisional, - accumulating: { kind: "thinking", text: event.delta }, + accumulating: { kind: "thinking", text: event.delta, turnId: event.turnId }, generating: true, }; } @@ -275,7 +313,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState }; return { ...state, - provisional: [...provisional, { role: "assistant", chunk }], + provisional: [...provisional, { role: "assistant", chunk, turnId: event.turnId }], accumulating: null, generating: true, }; @@ -293,7 +331,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState }; return { ...state, - provisional: [...provisional, { role: "tool", chunk }], + provisional: [...provisional, { role: "tool", chunk, turnId: event.turnId }], accumulating: null, generating: true, }; @@ -307,7 +345,7 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState : { type: "error", message: event.message }; return { ...state, - provisional: [...provisional, { role: "assistant", chunk }], + provisional: [...provisional, { role: "assistant", chunk, turnId: event.turnId }], accumulating: null, generating: false, }; @@ -352,7 +390,10 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState const provisional = flushAccumulating(state.provisional, state.accumulating); return { ...state, - provisional: [...provisional, { role: "user", chunk: { type: "text", text: event.text } }], + provisional: [ + ...provisional, + { role: "user", chunk: { type: "text", text: event.text }, turnId: event.turnId }, + ], accumulating: null, generating: true, }; diff --git a/src/core/chunks/selectors.ts b/src/core/chunks/selectors.ts index e46d5f5..8b265a5 100644 --- a/src/core/chunks/selectors.ts +++ b/src/core/chunks/selectors.ts @@ -8,17 +8,31 @@ import type { RenderedChunk, TranscriptState } from "./types"; export function selectChunks(state: TranscriptState): readonly RenderedChunk[] { const result: RenderedChunk[] = []; for (const c of state.committed) { - result.push({ seq: c.seq, role: c.role, chunk: c.chunk, provisional: false }); + result.push( + c.turnId !== undefined + ? { seq: c.seq, role: c.role, chunk: c.chunk, provisional: false, turnId: c.turnId } + : { seq: c.seq, role: c.role, chunk: c.chunk, provisional: false }, + ); } for (const p of state.provisional) { - result.push({ seq: null, role: p.role, chunk: p.chunk, provisional: true }); + result.push( + p.turnId !== undefined + ? { seq: null, role: p.role, chunk: p.chunk, provisional: true, turnId: p.turnId } + : { seq: null, role: p.role, chunk: p.chunk, provisional: true }, + ); } - if (state.accumulating !== null) { + const acc = state.accumulating; + if (acc !== null) { const chunk: Chunk = - state.accumulating.kind === "text" - ? { type: "text", text: state.accumulating.text } - : { type: "thinking", text: state.accumulating.text }; - result.push({ seq: null, role: "assistant", chunk, provisional: true, streaming: true }); + acc.kind === "text" ? { type: "text", text: acc.text } : { type: "thinking", text: acc.text }; + result.push({ + seq: null, + role: "assistant", + chunk, + provisional: true, + streaming: true, + ...(acc.turnId !== undefined ? { turnId: acc.turnId } : {}), + }); } return result; } diff --git a/src/core/chunks/types.ts b/src/core/chunks/types.ts index 2ced736..949857a 100644 --- a/src/core/chunks/types.ts +++ b/src/core/chunks/types.ts @@ -4,12 +4,20 @@ import type { Chunk, Role, StoredChunk, TurnProviderRetryEvent, Usage } from "@d export interface AccumulatingChunk { readonly kind: "text" | "thinking"; readonly text: string; + /** The turn generating this chunk (from the live event's `turnId`). */ + readonly turnId?: string; } /** A provisional chunk that has no authoritative seq yet. */ export interface ProvisionalChunk { readonly role: Role; readonly chunk: Chunk; + /** + * The turn this chunk belongs to (from the live event's `turnId`). Absent + * only on the sender's optimistic user echo until its `turn-start` arrives + * (which tags it). + */ + readonly turnId?: string; } /** The transcript reducer state. Holds committed history + live in-flight turn. */ @@ -74,4 +82,11 @@ export interface RenderedChunk { * dots on streaming thinking) and drop it the moment generation moves on. */ readonly streaming?: boolean; + /** + * The turn that produced this chunk — `StoredChunk.turnId` for committed + * chunks, the live event's `turnId` for provisional ones. The exact join key + * for placing per-turn metrics. Absent on legacy rows (persisted before the + * backend stamped it) and on the optimistic echo before `turn-start`. + */ + readonly turnId?: string; } 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/place.test.ts b/src/core/metrics/place.test.ts index 9c925a3..393d23c 100644 --- a/src/core/metrics/place.test.ts +++ b/src/core/metrics/place.test.ts @@ -401,7 +401,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,9 +429,9 @@ 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", () => { @@ -606,16 +611,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..9f8b109 100644 --- a/src/core/metrics/place.ts +++ b/src/core/metrics/place.ts @@ -1,6 +1,6 @@ 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; @@ -8,6 +8,14 @@ function groupStepId(g: RenderGroup): string | undefined { return c.type === "tool-call" || c.type === "tool-result" ? c.stepId : undefined; } +function groupTurnId(g: RenderGroup): string | undefined { + return g.kind === "tool-batch" ? 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). */ function addUsage(a: Usage, b: Usage): Usage { const out: Usage = { @@ -21,277 +29,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..16499f7 --- /dev/null +++ b/src/core/metrics/placement-scenarios.test.ts @@ -0,0 +1,380 @@ +// 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 === "tool-call") out.push(`call:${c.toolCallId}`); + else if (c.type === "tool-result") out.push(`result:${c.toolCallId}`); + else if (c.type === "error") out.push(`error:${c.message}`); + } + } + 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", + "call:k1", + "result: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", + "call:k1", + "result: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", + "call:k1", + "result:k1", + "[step t2#0]", + "user:steer!", + "call:k2", + "result: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; |
