From c99932b16d84a30cc558b01a951b3e4d97728177 Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Wed, 14 Oct 2026 19:57:32 +0200 Subject: [PATCH] fix(claude): show marked progress updates before questions The completed block identifies narration after its thinking deltas have already arrived. Buffer thinking until that classification is known, then emit marked blocks as assistant text and unmarked blocks as reasoning. Reuse per-block reconciliation so completed snapshots do not repeat output. Co-authored-by: ThePharmer --- .../agent.progress-updates.real.e2e.test.ts | 112 +++++++++++++ .../providers/claude/agent.redesign.test.ts | 157 +++++++++++++----- .../server/agent/providers/claude/agent.ts | 58 +++++-- 3 files changed, 272 insertions(+), 55 deletions(-) create mode 100644 packages/server/src/server/agent/providers/claude/agent.progress-updates.real.e2e.test.ts diff --git a/packages/server/src/server/agent/providers/claude/agent.progress-updates.real.e2e.test.ts b/packages/server/src/server/agent/providers/claude/agent.progress-updates.real.e2e.test.ts new file mode 100644 index 0000000000..9ca6488336 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/agent.progress-updates.real.e2e.test.ts @@ -0,0 +1,112 @@ +import { afterEach, expect, test } from "vitest"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import pino from "pino"; + +import type { AgentSession, AgentStreamEvent, AgentTimelineItem } from "../../agent-sdk-types.js"; +import { ClaudeAgentClient } from "./agent.js"; + +// Reproduces #6199 through the real Claude provider. Claude Opus 5.5 writes the +// text it means for the user between tool calls as a "progress update": a +// thinking block that Claude Code marks with `narration_block_indexes`. +const PROMPT = + "There are three plan files in this directory: plan-a.txt, plan-b.txt and plan-c.txt. Read each one, tell me how they compare on cost, time and risk, and then ask me which plan to pick using your AskUserQuestion tool. Keep me posted as you go."; + +const handles: Array<{ cwd: string; session: AgentSession }> = []; + +afterEach(async () => { + for (const handle of handles.splice(0)) { + await handle.session.close().catch(() => undefined); + rmSync(handle.cwd, { recursive: true, force: true }); + } +}); + +function writePlans(cwd: string): void { + writeFileSync( + path.join(cwd, "plan-a.txt"), + "Plan A\nCost: 3 credits\nTime: 2 days\nRisk: needs a manual review step\n", + ); + writeFileSync( + path.join(cwd, "plan-b.txt"), + "Plan B\nCost: 5 credits\nTime: 1 day\nRisk: none known\n", + ); + writeFileSync( + path.join(cwd, "plan-c.txt"), + "Plan C\nCost: 2 credits\nTime: 4 days\nRisk: depends on an external vendor\n", + ); +} + +test("the comparison written before AskUserQuestion reaches the timeline as assistant text", async () => { + const cwd = mkdtempSync(path.join(tmpdir(), "claude-progress-updates-")); + writePlans(cwd); + const narration: string[] = []; + const logger = pino( + { level: "trace" }, + { + write(line) { + const record = JSON.parse(line) as { + rawEvent?: { + type?: string; + narration_block_indexes?: number[]; + message?: { content?: Array<{ type: string; thinking?: string }> }; + }; + }; + const frame = record.rawEvent; + if (frame?.type !== "assistant") return; + for (const index of frame.narration_block_indexes ?? []) { + const block = frame.message?.content?.[index]; + if (block?.type === "thinking" && block.thinking) narration.push(block.thinking); + } + }, + }, + ); + const client = new ClaudeAgentClient({ logger }); + const session = await client.createSession({ + provider: "claude", + cwd, + model: "claude-opus-5-5", + thinkingOptionId: "xhigh", + modeId: "bypassPermissions", + }); + handles.push({ cwd, session }); + + const items: AgentTimelineItem[] = []; + await new Promise((resolve, reject) => { + session.subscribe((event: AgentStreamEvent) => { + if (event.type === "timeline") items.push(event.item); + if (event.type === "turn_failed") reject(new Error(event.error)); + if ( + event.type === "permission_requested" || + event.type === "turn_completed" || + event.type === "turn_failed" + ) { + resolve(); + } + }); + void session.startTurn(PROMPT).catch(reject); + }); + + const lastReadIndex = items.findLastIndex( + (item) => item.type === "tool_call" && item.name !== "AskUserQuestion", + ); + const askIndex = items.findIndex( + (item) => item.type === "tool_call" && item.name === "AskUserQuestion", + ); + expect(askIndex).toBeGreaterThan(lastReadIndex); + const between = items.slice(lastReadIndex + 1, askIndex); + const assistantText = between + .flatMap((item) => (item.type === "assistant_message" ? [item.text] : [])) + .join(""); + expect(assistantText).toMatch(/plan/i); + // A plain-text model response would not exercise narration classification. + expect(narration.length).toBeGreaterThan(0); + for (const text of narration) { + expect( + items.flatMap((item) => (item.type === "assistant_message" ? [item.text] : [])).join(""), + ).toContain(text); + expect( + items.flatMap((item) => (item.type === "reasoning" ? [item.text] : [])).join(""), + ).not.toContain(text); + } +}, 280_000); diff --git a/packages/server/src/server/agent/providers/claude/agent.redesign.test.ts b/packages/server/src/server/agent/providers/claude/agent.redesign.test.ts index 3966de92da..6251526a5b 100644 --- a/packages/server/src/server/agent/providers/claude/agent.redesign.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.redesign.test.ts @@ -1,5 +1,6 @@ import { afterEach, beforeEach, expect, test, vi } from "vitest"; import type { Logger } from "pino"; +import type { Query } from "@anthropic-ai/claude-agent-sdk"; import { createTestLogger } from "../../../../test-utils/test-logger.js"; import { asInternals } from "../../../test-utils/class-mocks.js"; @@ -2097,46 +2098,41 @@ function blockSnapshot(input: { uuid: string; kind: "thinking" | "text"; text: s } async function runScriptedTurn(frames: Record[]): Promise { - sdkQueryFactory.mockImplementation(({ prompt }: { prompt: AsyncIterable }) => { - const readPromptUuid = createPromptUuidReader(prompt); - let step = 0; - return createBaseQueryMock( - vi.fn(async () => { - const index = step++; - if (index === 0) { - return { - done: false, - value: { ...claudeTurnInit(), session_id: MULTI_BLOCK_SESSION_ID }, - }; - } - if (index === 1) { - return { - done: false, - value: { - type: "user", - message: { role: "user", content: "multi-block prompt" }, - parent_tool_use_id: null, - uuid: (await readPromptUuid()) ?? "missing-prompt-uuid", - session_id: MULTI_BLOCK_SESSION_ID, - }, - }; - } - const frame = frames[index - 2]; - if (frame) { - return { done: false, value: frame }; - } - if (index === frames.length + 2) { - return { done: false, value: successResult() }; - } - return { done: true, value: undefined }; - }), - ); + // An injected SDK transport exercises the public session, without module mocks. + const client = new ClaudeAgentClient({ + logger: createTestLogger(), + resolveBinary: async () => "/test/claude/bin", + queryFactory: ({ prompt }) => { + const readPromptUuid = createPromptUuidReader(prompt as AsyncIterable); + const iterator = (async function* () { + yield { ...claudeTurnInit(), session_id: MULTI_BLOCK_SESSION_ID }; + yield { + type: "user", + message: { role: "user", content: "multi-block prompt" }, + parent_tool_use_id: null, + uuid: (await readPromptUuid()) ?? "missing-prompt-uuid", + session_id: MULTI_BLOCK_SESSION_ID, + }; + yield* frames; + yield successResult(); + })(); + return Object.assign(iterator, { + interrupt: async () => undefined, + close: () => undefined, + setPermissionMode: async () => undefined, + setModel: async () => undefined, + supportedModels: async () => [{ value: "opus", displayName: "Opus" }], + supportedCommands: async () => [], + }) as unknown as Query; + }, }); - - const session = await createSession(); - const events = await collectUntilTerminal(streamSession(session, "multi-block prompt")); - await session.close(); - return events.flatMap((event) => (event.type === "timeline" ? [event.item] : [])); + const session = await client.createSession({ provider: "claude", cwd: process.cwd() }); + try { + const events = await collectUntilTerminal(streamSession(session, "multi-block prompt")); + return events.flatMap((event) => (event.type === "timeline" ? [event.item] : [])); + } finally { + await session.close(); + } } // Each chunk as emitted, so a test sees duplicates and lost chunks, not just the final text. @@ -2175,8 +2171,7 @@ test("shows each thinking block of a multi-block message once", async () => { ]); expect(emittedChunks({ items, type: "reasoning" })).toEqual([ - "First paragraph.\n\n", - "Second paragraph.\n\n", + "First paragraph.\n\nSecond paragraph.\n\n", "Last paragraph.\n\n", ]); }); @@ -2212,7 +2207,7 @@ test("fills in text a block snapshot has beyond what streamed", async () => { MESSAGE_STOP, ]); - expect(emittedChunks({ items, type: "reasoning" })).toEqual(["Partial", " thought.", "Next."]); + expect(emittedChunks({ items, type: "reasoning" })).toEqual(["Partial thought.", "Next."]); }); test("keeps identical blocks that arrive only as snapshots", async () => { @@ -2271,3 +2266,81 @@ test("does not hold a snapshot slot for an empty thinking block", async () => { expect(emittedChunks({ items, type: "reasoning" })).toEqual(["Visible."]); }); + +test("shows a streamed narration block as assistant text before its question", async () => { + const comparison = "Plan B is fastest. Plan C costs least."; + const items = await runScriptedTurn([ + MESSAGE_START, + ...streamedBlock({ index: 0, kind: "thinking", deltas: ["Compare cost", " and time."] }), + blockSnapshot({ uuid: "reason", kind: "thinking", text: "Compare cost and time." }), + blockStop(0), + ...streamedBlock({ index: 1, kind: "thinking", deltas: [comparison] }), + { + ...blockSnapshot({ uuid: "narration", kind: "thinking", text: comparison }), + narration_block_indexes: [0], + }, + blockStop(1), + { + type: "assistant", + parent_tool_use_id: null, + message: { + id: "msg-multi", + content: [{ type: "tool_use", id: "ask-plan", name: "AskUserQuestion", input: {} }], + }, + }, + MESSAGE_STOP, + ]); + + expect(emittedChunks({ items, type: "reasoning" }).join("")).toBe("Compare cost and time."); + expect(emittedChunks({ items, type: "assistant_message" })).toEqual([comparison]); + const comparisonIndex = items.findIndex((item) => item.type === "assistant_message"); + const questionIndex = items.findIndex((item) => item.type === "tool_call"); + expect(questionIndex).toBeGreaterThan(comparisonIndex); +}); + +test("classifies only the marked thinking block of a snapshot-only message", async () => { + const items = await runScriptedTurn([ + { + type: "assistant", + uuid: "snapshot-only", + parent_tool_use_id: null, + narration_block_indexes: [2], + message: { + id: "msg-multi", + content: [ + { type: "text", text: "Reading plans." }, + { type: "thinking", thinking: "Compare the options." }, + { type: "thinking", thinking: "Plan B is fastest." }, + ], + }, + }, + ]); + expect(emittedChunks({ items, type: "assistant_message" })).toEqual([ + "Reading plans.", + "Plan B is fastest.", + ]); + expect(emittedChunks({ items, type: "reasoning" })).toEqual(["Compare the options."]); +}); + +test("ignores malformed narration markers and retains unmarked reasoning", async () => { + for (const marker of [undefined, null, "0", ["0"], [-1], [1.5], [3]]) { + const items = await runScriptedTurn([ + { + ...blockSnapshot({ uuid: "unmarked", kind: "thinking", text: "A thought." }), + narration_block_indexes: marker, + }, + ]); + expect(emittedChunks({ items, type: "reasoning" })).toEqual(["A thought."]); + expect(emittedChunks({ items, type: "assistant_message" })).toEqual([]); + } +}); + +test("retains thinking at message stop when no completed block snapshot arrives", async () => { + const items = await runScriptedTurn([ + MESSAGE_START, + ...streamedBlock({ index: 0, kind: "thinking", deltas: ["Partial", " thought."] }), + blockStop(0), + MESSAGE_STOP, + ]); + expect(emittedChunks({ items, type: "reasoning" }).join("")).toBe("Partial thought."); +}); diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index e529b3fd55..3a427e426a 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -1144,6 +1144,7 @@ function buildClaudePlanPermissionActions( interface TimelineFragment { kind: "assistant" | "reasoning"; text: string; + narration: boolean; } // While streaming, the SDK emits one assistant message per completed non-empty content block, @@ -1212,7 +1213,16 @@ class TimelineAssembler { } state.appliedSnapshotUuids.add(snapshotUuid); } - const fragments = this.extractFragments(message.message?.content); + const marker = toObjectRecord(message)?.narration_block_indexes; + const narrationIndexes = new Set( + Array.isArray(marker) + ? marker.filter((index): index is number => Number.isInteger(index) && index >= 0) + : [], + ); + const fragments = this.extractFragments({ + content: message.message?.content, + narrationIndexes, + }); return this.applyBlockSnapshots({ state, fragments }); } @@ -1291,7 +1301,7 @@ class TimelineAssembler { messageIdHint: string | null; }): AgentTimelineItem[] { const { content, blockIndex, runId, messageIdHint } = input; - const fragments = this.extractFragments(content); + const fragments = this.extractFragments({ content }); if (fragments.length === 0) { return []; } @@ -1310,7 +1320,7 @@ class TimelineAssembler { const block = state.blocks.get(key) ?? this.addBlock({ state, key, kind: fragment.kind }); block.text += fragment.text; } - return this.emitNewContent(state); + return this.emitNewContent({ state }); } private resolveStreamBlockKey(input: { @@ -1343,22 +1353,27 @@ class TimelineAssembler { if (!match) { const key = `snapshot:${++state.syntheticBlockCount}`; const snapshotOnly = this.addBlock({ state, key, kind: fragment.kind }); + snapshotOnly.kind = fragment.narration ? "assistant" : fragment.kind; snapshotOnly.text = fragment.text; snapshotOnly.snapshotApplied = true; continue; } const [key, block] = match; block.snapshotApplied = true; + if (fragment.narration) { + block.kind = "assistant"; + } // A snapshot completes its block, so later deltas without an index start a new one. if (state.openStreamBlockKey === key) { state.openStreamBlockKey = null; } - // Emitted text cannot be retracted, so a snapshot that rewrites streamed text is ignored. - if (fragment.text.startsWith(block.text)) { + // Buffered thinking has not been emitted, so its completed text is authoritative. + // Plain text has streamed already and can only be extended. + if (block.emittedLength === 0 || fragment.text.startsWith(block.text)) { block.text = fragment.text; } } - return this.emitNewContent(state); + return this.emitNewContent({ state }); } private addBlock(input: { @@ -1378,7 +1393,7 @@ class TimelineAssembler { return []; } state.stopped = true; - const items = this.emitNewContent(state); + const items = this.emitNewContent({ state, finalizing: true }); if (runId && this.activeMessageByRun.get(runId) === messageId) { this.activeMessageByRun.delete(runId); } @@ -1387,9 +1402,18 @@ class TimelineAssembler { return items; } - private emitNewContent(state: TimelineMessageState): AgentTimelineItem[] { + private emitNewContent(input: { + state: TimelineMessageState; + finalizing?: boolean; + }): AgentTimelineItem[] { + const { state, finalizing = false } = input; const items: AgentTimelineItem[] = []; for (const block of state.blocks.values()) { + // Narration is identified only by the completed snapshot. Hold thinking until then, + // so a progress update never reaches clients as reasoning. Plain text still streams. + if (block.kind === "reasoning" && !block.snapshotApplied && !finalizing) { + continue; + } const nextText = block.text.slice(block.emittedLength); if (nextText.length === 0) { continue; @@ -1456,16 +1480,20 @@ class TimelineAssembler { return synthetic; } - private extractFragments(content: unknown): TimelineFragment[] { + private extractFragments(input: { + content: unknown; + narrationIndexes?: ReadonlySet; + }): TimelineFragment[] { + const { content, narrationIndexes = new Set() } = input; if (typeof content === "string") { if (content.length === 0) { return []; } - return [{ kind: "assistant", text: content }]; + return [{ kind: "assistant", text: content, narration: false }]; } const blocks = Array.isArray(content) ? content : [content]; const fragments: TimelineFragment[] = []; - for (const rawBlock of blocks) { + for (const [index, rawBlock] of blocks.entries()) { if (!isClaudeContentChunk(rawBlock)) { continue; } @@ -1474,14 +1502,18 @@ class TimelineAssembler { typeof rawBlock.text === "string" && rawBlock.text.length > 0 ) { - fragments.push({ kind: "assistant", text: rawBlock.text }); + fragments.push({ kind: "assistant", text: rawBlock.text, narration: false }); } if ( (rawBlock.type === "thinking" || rawBlock.type === "thinking_delta") && typeof rawBlock.thinking === "string" && rawBlock.thinking.length > 0 ) { - fragments.push({ kind: "reasoning", text: rawBlock.thinking }); + fragments.push({ + kind: "reasoning", + text: rawBlock.thinking, + narration: narrationIndexes.has(index), + }); } } return fragments;