From 01f9fb2a4ad6110def531ead16b370c818e8ee81 Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Mon, 19 Oct 2026 17:02:40 +0900 Subject: feat(chat): cache timer for Claude Code subscription models Per-step and per-turn timers for the gaps between model requests (the Claude prompt cache expires after 5 min idle). A step gap runs live under the executing tools from step-complete until its last tool result (or the next step's output), then is stamped on that step's metrics line beside its cache badge. A turn gap runs live on the turn's metrics line from done/error until the next message is sent, then is stamped there. Counts m:ss up to 10 min, then '10+ min'; colour tracks the 5-min TTL. Gated on claude/. Pure reducer in core/metrics/cache-timer.ts; the turn gap is seeded from an idle conversation's lastActivityAt so it survives a reload. Stamps are in-memory only. --- src/core/metrics/cache-timer.test.ts | 344 +++++++++++++++++++++++++++++++++++ 1 file changed, 344 insertions(+) create mode 100644 src/core/metrics/cache-timer.test.ts (limited to 'src/core/metrics/cache-timer.test.ts') diff --git a/src/core/metrics/cache-timer.test.ts b/src/core/metrics/cache-timer.test.ts new file mode 100644 index 0000000..e55eff4 --- /dev/null +++ b/src/core/metrics/cache-timer.test.ts @@ -0,0 +1,344 @@ +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", + }), + 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("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(); + }); +}); -- cgit v1.2.3 From 75680eac7071c6968eaa5470686ad102a82103db Mon Sep 17 00:00:00 2001 From: Adam Malczewski Date: Thu, 22 Oct 2026 17:51:15 +0900 Subject: feat(chunks): split live thinking at reasoning-end block boundaries --- .dispatch/wire.reference.md | 26 +++++++++++++ backend-handoff.md | 12 ++++++ src/core/chunks/reducer.test.ts | 73 ++++++++++++++++++++++++++++++++++++ src/core/chunks/reducer.ts | 27 +++++++++++++ src/core/chunks/selectors.test.ts | 67 +++++++++++++++++++++++++++++++++ src/core/metrics/cache-timer.test.ts | 15 ++++++++ src/core/metrics/cache-timer.ts | 1 + src/core/wire/conformance.ts | 2 + 8 files changed, 223 insertions(+) (limited to 'src/core/metrics/cache-timer.test.ts') diff --git a/.dispatch/wire.reference.md b/.dispatch/wire.reference.md index 3743bd5..777a2ba 100644 --- a/.dispatch/wire.reference.md +++ b/.dispatch/wire.reference.md @@ -1,5 +1,15 @@ # `@dispatch/wire` — in-repo reference (read THIS, not node_modules) +> **2026-10-22 delta (reasoning block boundaries — ADDITIVE, no version bump):** `AgentEvent` +> gains `TurnReasoningEndEvent` `{ type: "reasoning-end", conversationId, turnId }`. The kernel emits +> one per provider reasoning block (Anthropic signs each thinking block separately), carrying NO +> opaque payload (signature/redacted data never leave the server). A live consumer splits thinking +> exactly as HTTP history persists it: the open thinking chunk closes; with none open (a redacted +> empty block) an empty thinking chunk is implied. Providers without block structure never emit it. +> The FE reducer mirrors the kernel's chunk builder; the old merged-run alignment stays as a fallback +> for older servers/replays. See `backend-handoff.md` and the backend's +> `notes/reasoning-end-live-event-plan.md`. +> > **2026-10-20 review 3.5:** history replacements preserve monotonic seqs (retained history need not start at 1). HTTP history and WS compaction carry required `historyEpoch`; unknown cached epochs must be reloaded. Consumed by review 3.6. > > MIRRORS the backend's `@dispatch/wire` package source so headless FE agents can read the wire @@ -414,6 +424,7 @@ export type AgentEvent = | TurnInputEvent | TurnTextDeltaEvent | TurnReasoningDeltaEvent + | TurnReasoningEndEvent | TurnToolCallEvent | TurnToolResultEvent | TurnToolOutputEvent @@ -474,6 +485,21 @@ export interface TurnReasoningDeltaEvent { readonly delta: string; } +/** + * End of one reasoning / thinking block. Emitted once per block by providers with + * block-structured reasoning (e.g. Anthropic, whose consecutive thinking blocks — each + * separately signed — persist as separate thinking chunks). Lets a live consumer split + * thinking exactly as history does: the open thinking chunk closes; if none is open (a + * redacted/empty block with no deltas) an EMPTY thinking chunk is implied. Carries no + * content: the opaque replay payloads (signature / redacted data) never leave the + * server. Providers without block structure never emit it (deltas merge as before). + */ +export interface TurnReasoningEndEvent { + readonly type: "reasoning-end"; + readonly conversationId: string; + readonly turnId: string; +} + /** The model has requested a tool to be run. */ export interface TurnToolCallEvent { readonly type: "tool-call"; diff --git a/backend-handoff.md b/backend-handoff.md index b4d53bc..0878b93 100644 --- a/backend-handoff.md +++ b/backend-handoff.md @@ -18,6 +18,18 @@ > ?archived=`) was made in the same branch and the FE consumes it via the `file:` deps (dists > rebuilt). Recorded below. +## Milestone: reasoning block boundaries on the live stream (2026-10-22) + +FE slice consuming a new ADDITIVE backend wire event. The kernel now emits one payload-free +`reasoning-end` `AgentEvent` per provider reasoning block (no version bump; opaque +signature/redacted data never leave the server). The pure chunk reducer mirrors the kernel's +builder — an open thinking chunk is flushed as its own provisional chunk; with none open it +appends an empty thinking chunk — so live thinking splits identically to HTTP history and the +replay/history de-dup lines up 1:1. The old merged-run alignment (`mergedRun`) stays as a +fallback for older servers/replays. The cache timer treats `reasoning-end` as resumed output; +`assertAgentEventExhaustive` covers it. Mirrored in `.dispatch/wire.reference.md` +(`TurnReasoningEndEvent`). No backend asks. + ## Milestone: Delivery warning display delay + Flush label (2026-10-21) Frontend-only: each keyed pending-input warning waits 1000 ms on mount (including diff --git a/src/core/chunks/reducer.test.ts b/src/core/chunks/reducer.test.ts index 4996e88..10d662e 100644 --- a/src/core/chunks/reducer.test.ts +++ b/src/core/chunks/reducer.test.ts @@ -7,6 +7,7 @@ import type { TurnInputEvent, TurnProviderRetryEvent, TurnReasoningDeltaEvent, + TurnReasoningEndEvent, TurnSealedEvent, TurnStartEvent, TurnSteeringEvent, @@ -45,6 +46,12 @@ const reasoningDelta = (turnId: string, delta: string): TurnReasoningDeltaEvent delta, }); +const reasoningEnd = (turnId: string): TurnReasoningEndEvent => ({ + type: "reasoning-end", + conversationId: "c1", + turnId, +}); + const toolCall = ( turnId: string, toolCallId: string, @@ -278,6 +285,72 @@ describe("foldEvent — reasoning-delta", () => { }); }); +describe("foldEvent — reasoning-end (thinking block boundary)", () => { + it("closes each block so consecutive deltas become separate chunks", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, reasoningDelta("t1", "a")); + s = foldEvent(s, reasoningEnd("t1")); + s = foldEvent(s, reasoningDelta("t1", "b")); + s = foldEvent(s, reasoningEnd("t1")); + expect(s.provisional.map((p) => p.chunk)).toEqual([ + { type: "thinking", text: "a" }, + { type: "thinking", text: "b" }, + ]); + expect(s.accumulating).toBeNull(); + }); + + it("still merges legacy deltas with no boundary (no reasoning-end)", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, reasoningDelta("t1", "a")); + s = foldEvent(s, reasoningDelta("t1", "b")); + expect(s.accumulating).toEqual({ kind: "thinking", text: "ab", turnId: "t1" }); + }); + + it("end with nothing open appends an empty thinking chunk for the turn", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, reasoningEnd("t1")); + expect(s.provisional).toHaveLength(1); + expect(s.provisional[0]?.chunk).toEqual({ type: "thinking", text: "" }); + expect(s.provisional[0]?.turnId).toBe("t1"); + expect(s.accumulating).toBeNull(); + }); + + it("text-delta then end flushes the text, then appends an empty thinking chunk", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, textDelta("t1", "x")); + s = foldEvent(s, reasoningEnd("t1")); + expect(s.provisional.map((p) => p.chunk)).toEqual([ + { type: "text", text: "x" }, + { type: "thinking", text: "" }, + ]); + }); + + it("end for another turn flushes the open thinking and appends an empty one", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, reasoningDelta("t1", "open")); + s = foldEvent(s, reasoningEnd("t2")); + expect(s.provisional.map((p) => [p.turnId, p.chunk])).toEqual([ + ["t1", { type: "thinking", text: "open" }], + ["t2", { type: "thinking", text: "" }], + ]); + expect(s.accumulating).toBeNull(); + }); + + it("clears a pending provider-retry banner", () => { + let s = initialState(); + s = foldEvent(s, turnStart("t1")); + s = foldEvent(s, providerRetry("t1", 0, 5000)); + expect(selectProviderRetry(s)).not.toBeNull(); + s = foldEvent(s, reasoningEnd("t1")); + expect(selectProviderRetry(s)).toBeNull(); + }); +}); + describe("foldEvent — tool-call then tool-result", () => { it("tool-call then tool-result render in order", () => { let s = initialState(); diff --git a/src/core/chunks/reducer.ts b/src/core/chunks/reducer.ts index 35bb1bb..71df4dd 100644 --- a/src/core/chunks/reducer.ts +++ b/src/core/chunks/reducer.ts @@ -284,6 +284,8 @@ export function applyHistory( * optimistic echo) so a watcher renders it mid-turn. * - `text-delta` extends the current accumulating TextChunk (or starts one). * - `reasoning-delta` extends the current accumulating ThinkingChunk (or starts one). + * - `reasoning-end` closes the open ThinkingChunk as its own chunk; with none open + * (a redacted/empty block) it appends an empty thinking chunk, mirroring history. * - `tool-call` / `tool-result` / `error` finalize any accumulating chunk and * add a new provisional chunk. * - `steering` appends a user bubble mid-turn (drained from the message queue @@ -415,6 +417,30 @@ function reduceEvent(state: TranscriptState, event: AgentEvent): TranscriptState }; } + case "reasoning-end": { + const acc = state.accumulating; + // Close the open thinking block of this turn: flush it as its own provisional chunk. + if (acc !== null && acc.kind === "thinking" && acc.turnId === event.turnId) { + return { + ...state, + provisional: flushAccumulating(state.provisional, acc), + accumulating: null, + generating: true, + }; + } + // No open thinking (a redacted/empty block with no deltas): history persists an + // EMPTY thinking chunk for it, so mirror that 1:1. + return { + ...state, + provisional: [ + ...flushAccumulating(state.provisional, acc), + { role: "assistant", chunk: { type: "thinking", text: "" }, turnId: event.turnId }, + ], + accumulating: null, + generating: true, + }; + } + case "tool-call": { const provisional = flushAccumulating(state.provisional, state.accumulating); const chunk: Chunk = { @@ -568,6 +594,7 @@ const RETRY_CLEARING_EVENTS: ReadonlySet = new Set([ "turn-start", "text-delta", "reasoning-delta", + "reasoning-end", "tool-call", "tool-result", "error", diff --git a/src/core/chunks/selectors.test.ts b/src/core/chunks/selectors.test.ts index 3afc324..90bb5cc 100644 --- a/src/core/chunks/selectors.test.ts +++ b/src/core/chunks/selectors.test.ts @@ -20,6 +20,7 @@ const think = (delta: string): AgentEvent => ({ turnId: T, delta, }); +const thinkEnd: AgentEvent = { type: "reasoning-end", conversationId: C, turnId: T }; const text = (delta: string): AgentEvent => ({ type: "text-delta", conversationId: C, @@ -299,6 +300,72 @@ describe("selectChunks — de-dup of a late-join replay against committed histor expect(provisional.map((r) => r.chunk)).toMatchObject([{ type: "tool-call", toolCallId: "b" }]); }); + it("aligns a boundary-carrying replay 1:1 against split committed blocks", () => { + const split: StoredChunk[] = [ + committedStep0[0] as StoredChunk, + { seq: 2, role: "assistant", chunk: { type: "thinking", text: "first. " }, turnId: T }, + { seq: 3, role: "assistant", chunk: { type: "thinking", text: "second." }, turnId: T }, + { ...(committedStep0[2] as StoredChunk), seq: 4 }, + { ...(committedStep0[3] as StoredChunk), seq: 5 }, + ]; + let s = applyHistory(initialState(), split); + s = fold(s, [ + userMsg, + turnStart, + think("first. "), + thinkEnd, + think("second."), + thinkEnd, + call("a"), + result("a"), + text("next"), + ]); + const rendered = selectChunks(s); + expect(rendered.map((r) => [r.seq, r.chunk.type])).toEqual([ + [1, "text"], + [2, "thinking"], + [3, "thinking"], + [4, "tool-call"], + [5, "tool-result"], + [null, "text"], + ]); + expect(rendered[5]?.streaming).toBe(true); + }); + + it("aligns a redacted empty block with a bare reasoning-end", () => { + const split: StoredChunk[] = [ + committedStep0[0] as StoredChunk, + { seq: 2, role: "assistant", chunk: { type: "thinking", text: "first. " }, turnId: T }, + { seq: 3, role: "assistant", chunk: { type: "thinking", text: "" }, turnId: T }, + { seq: 4, role: "assistant", chunk: { type: "thinking", text: "second." }, turnId: T }, + { ...(committedStep0[2] as StoredChunk), seq: 5 }, + { ...(committedStep0[3] as StoredChunk), seq: 6 }, + ]; + let s = applyHistory(initialState(), split); + s = fold(s, [ + userMsg, + turnStart, + think("first. "), + thinkEnd, + thinkEnd, + think("second."), + thinkEnd, + call("a"), + result("a"), + text("next"), + ]); + const rendered = selectChunks(s); + expect(rendered.map((r) => [r.seq, r.chunk.type])).toEqual([ + [1, "text"], + [2, "thinking"], + [3, "thinking"], + [4, "thinking"], + [5, "tool-call"], + [6, "tool-result"], + [null, "text"], + ]); + }); + it("is the identity when there is no committed history", () => { const s = fold(initialState(), [userMsg, turnStart, text("hello")]); expect(selectChunks(s)).toHaveLength(2); diff --git a/src/core/metrics/cache-timer.test.ts b/src/core/metrics/cache-timer.test.ts index e55eff4..d450135 100644 --- a/src/core/metrics/cache-timer.test.ts +++ b/src/core/metrics/cache-timer.test.ts @@ -32,6 +32,11 @@ const ev = { 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, @@ -151,6 +156,16 @@ describe("foldCacheTimerEvent — step gaps", () => { 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")], diff --git a/src/core/metrics/cache-timer.ts b/src/core/metrics/cache-timer.ts index 66d5fac..3d7270d 100644 --- a/src/core/metrics/cache-timer.ts +++ b/src/core/metrics/cache-timer.ts @@ -208,6 +208,7 @@ export function foldCacheTimerEvent( 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; diff --git a/src/core/wire/conformance.ts b/src/core/wire/conformance.ts index cd53b65..eff2330 100644 --- a/src/core/wire/conformance.ts +++ b/src/core/wire/conformance.ts @@ -18,6 +18,8 @@ export function assertAgentEventExhaustive(event: AgentEvent): string { return "text-delta"; case "reasoning-delta": return "reasoning-delta"; + case "reasoning-end": + return "reasoning-end"; case "tool-call": return "tool-call"; case "tool-result": -- cgit v1.2.3