From b34c188f8f0d27f003ea7c2d4bac422370eef231 Mon Sep 17 00:00:00 2001 From: Ayaan Zaidi Date: Sat, 4 Jul 2026 19:37:36 -0700 Subject: [PATCH] fix(agents): surface Claude CLI thinking token progress --- .../telegram/src/bot-message-dispatch.test.ts | 26 ++++ .../telegram/src/bot-message-dispatch.ts | 22 +++ src/agents/cli-output.test.ts | 127 +++++++++++++++++- src/agents/cli-output.ts | 117 +++++++++++++--- src/agents/cli-runner/claude-live-session.ts | 7 +- src/agents/cli-runner/execute.ts | 20 ++- src/auto-reply/get-reply-options.types.ts | 5 + .../reply/agent-runner-cli-dispatch.test.ts | 52 ++++++- .../reply/agent-runner-cli-dispatch.ts | 40 ++++++ .../reply/agent-runner-execution.ts | 3 + src/auto-reply/reply/followup-runner.ts | 3 + .../cli/claude-2.1-thinking-progress.jsonl | 20 +++ 12 files changed, 417 insertions(+), 25 deletions(-) create mode 100644 test/fixtures/cli/claude-2.1-thinking-progress.jsonl diff --git a/extensions/telegram/src/bot-message-dispatch.test.ts b/extensions/telegram/src/bot-message-dispatch.test.ts index 5968ef4ca81..a59de0a1d8b 100644 --- a/extensions/telegram/src/bot-message-dispatch.test.ts +++ b/extensions/telegram/src/bot-message-dispatch.test.ts @@ -4137,6 +4137,32 @@ describe("dispatchTelegramMessage draft streaming", () => { ); }); + it("renders CLI thinking token progress in the Telegram progress draft", async () => { + const draftStream = createSequencedDraftStream(2001); + createTelegramDraftStream.mockReturnValue(draftStream); + dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async ({ replyOptions }) => { + await replyOptions?.onReplyStart?.(); + await replyOptions?.onAssistantMessageStart?.(); + await replyOptions?.onReasoningProgress?.({ progressTokens: 50 }); + await replyOptions?.onReasoningProgress?.({ progressTokens: 200 }); + return { queuedFinal: false }; + }); + + await dispatchWithContext({ + context: createContext(), + streamMode: "progress", + telegramCfg: { streaming: { mode: "progress", progress: { label: "Shelling" } } }, + }); + + expect(createTelegramDraftStream).toHaveBeenCalledTimes(1); + expect(draftStream.updatePreview).toHaveBeenLastCalledWith( + telegramProgressPreview( + "Shelling\n\n🧠 Thinking… (~200 tokens)", + "Shelling\n🧠 Thinking… (~200 tokens)", + ), + ); + }); + it("renders model markdown in streamed reasoning and commentary lanes", async () => { const draftStream = createSequencedDraftStream(2001); createTelegramDraftStream.mockReturnValue(draftStream); diff --git a/extensions/telegram/src/bot-message-dispatch.ts b/extensions/telegram/src/bot-message-dispatch.ts index c8c62d0a2c4..f7c5a822997 100644 --- a/extensions/telegram/src/bot-message-dispatch.ts +++ b/extensions/telegram/src/bot-message-dispatch.ts @@ -394,6 +394,19 @@ function formatTelegramProgressLine(text: string): string { : formatProgressAsMarkdownCode(text); } +function buildTelegramThinkingProgressLine(progressTokens: number): ChannelProgressDraftLine { + const label = `Thinking… (~${Math.round(progressTokens)} tokens)`; + const text = `🧠 ${label}`; + return { + id: "reasoning:token-progress", + kind: "item", + icon: "🧠", + label, + text, + prefix: false, + }; +} + function escapeTelegramProgressHtml(text: string): string { return text .replaceAll("&", "&") @@ -2631,6 +2644,15 @@ export const dispatchTelegramMessage = async ({ await pushStreamReasoningProgress(payload); }) : undefined, + onReasoningProgress: answerLane.stream + ? (payload) => + enqueueDraftLaneEvent(async () => { + await pushStreamToolProgress( + buildTelegramThinkingProgressLine(payload.progressTokens), + { startImmediately: true }, + ); + }) + : undefined, onAssistantMessageStart: answerLane.stream ? () => enqueueDraftLaneEvent(async () => { diff --git a/src/agents/cli-output.test.ts b/src/agents/cli-output.test.ts index b304698b6ff..6ad1f5ae01f 100644 --- a/src/agents/cli-output.test.ts +++ b/src/agents/cli-output.test.ts @@ -1,4 +1,5 @@ /** Tests CLI JSON/JSONL output parsing, streamed deltas, and error extraction. */ +import { readFileSync } from "node:fs"; import { describe, expect, it } from "vitest"; import { createCliJsonlStreamingParser, @@ -7,6 +8,7 @@ import { parseCliJsonl, parseCliOutput, supportsCliJsonlToolEvents, + type CliThinkingProgress, type CliToolResultDelta, type CliToolUseStartDelta, } from "./cli-output.js"; @@ -1286,7 +1288,7 @@ describe("createCliJsonlStreamingParser", () => { event: { type: "content_block_delta", index: 1, - delta: { type: "input_json_delta", partial_json: "{\"file_path\":\"x\"}" }, + delta: { type: "input_json_delta", partial_json: '{"file_path":"x"}' }, }, }), JSON.stringify({ @@ -1322,6 +1324,127 @@ describe("createCliJsonlStreamingParser", () => { ]); }); + it("streams indexless thinking deltas from content block framing", () => { + const thinking: Array<{ text: string; delta: string; isReasoningSnapshot?: boolean }> = []; + const parser = createCliJsonlStreamingParser({ + backend: { + command: "local-cli", + output: "jsonl", + jsonlDialect: "claude-stream-json", + sessionIdFields: ["session_id"], + }, + providerId: "local-cli", + onAssistantDelta: () => {}, + onThinkingDelta: (delta) => thinking.push(delta), + }); + + parser.push( + [ + JSON.stringify({ + type: "stream_event", + event: { + type: "message_start", + message: { id: "msg-1" }, + }, + }), + JSON.stringify({ + type: "stream_event", + event: { + type: "content_block_start", + content_block: { type: "thinking" }, + }, + }), + JSON.stringify({ + type: "stream_event", + event: { + type: "content_block_delta", + delta: { type: "thinking_delta", thinking: "A" }, + }, + }), + JSON.stringify({ + type: "stream_event", + event: { type: "content_block_stop" }, + }), + JSON.stringify({ + type: "stream_event", + event: { + type: "content_block_start", + content_block: { type: "tool_use", id: "tool-1", name: "Read" }, + }, + }), + JSON.stringify({ + type: "stream_event", + event: { + type: "content_block_delta", + delta: { type: "input_json_delta", partial_json: '{"file_path":"x"}' }, + }, + }), + JSON.stringify({ + type: "stream_event", + event: { type: "content_block_stop" }, + }), + JSON.stringify({ + type: "stream_event", + event: { + type: "content_block_start", + content_block: { type: "thinking" }, + }, + }), + JSON.stringify({ + type: "stream_event", + event: { + type: "content_block_delta", + delta: { type: "thinking_delta", thinking: "B" }, + }, + }), + JSON.stringify({ + type: "stream_event", + event: { type: "content_block_stop" }, + }), + JSON.stringify({ + type: "assistant", + message: { + id: "msg-1", + content: [{ type: "text", text: "Answer." }], + }, + }), + ].join("\n"), + ); + parser.finish(); + + expect(thinking).toEqual([ + { text: "A", delta: "A", isReasoningSnapshot: true }, + { text: "AB", delta: "B", isReasoningSnapshot: true }, + ]); + }); + + it("emits token progress for Claude CLI 2.1 empty thinking deltas", () => { + const thinking: Array<{ text: string; delta: string; isReasoningSnapshot?: boolean }> = []; + const progress: CliThinkingProgress[] = []; + const parser = createCliJsonlStreamingParser({ + backend: { + command: "claude", + output: "jsonl", + jsonlDialect: "claude-stream-json", + sessionIdFields: ["session_id"], + }, + providerId: "claude-cli", + onAssistantDelta: () => {}, + onThinkingDelta: (delta) => thinking.push(delta), + onThinkingProgress: (payload) => progress.push(payload), + }); + + parser.push(readFileSync("test/fixtures/cli/claude-2.1-thinking-progress.jsonl", "utf8")); + parser.finish(); + + expect(thinking).toEqual([]); + expect(progress).toEqual([ + { progressTokens: 50 }, + { progressTokens: 200 }, + { progressTokens: 300 }, + ]); + }); + it("resets per-index thinking state on a new message within the same turn (tool round-trip)", () => { const thinking: Array<{ text: string; delta: string; isReasoningSnapshot?: boolean }> = []; const parser = createCliJsonlStreamingParser({ @@ -1410,7 +1533,7 @@ describe("createCliJsonlStreamingParser", () => { ]); }); - it("ignores thinking deltas without a numeric content-block index", () => { + it("ignores indexless thinking deltas without content block framing", () => { const thinking: Array<{ text: string; delta: string; isReasoningSnapshot?: boolean }> = []; const parser = createCliJsonlStreamingParser({ backend: { diff --git a/src/agents/cli-output.ts b/src/agents/cli-output.ts index 7fc84212a9c..b28798211ec 100644 --- a/src/agents/cli-output.ts +++ b/src/agents/cli-output.ts @@ -83,6 +83,10 @@ export type CliThinkingDelta = { isReasoningSnapshot?: boolean; }; +export type CliThinkingProgress = { + progressTokens: number; +}; + /** Tool-call start event reconstructed from CLI stream output. */ export type CliToolUseStartDelta = { toolCallId: string; @@ -796,10 +800,26 @@ type ThinkingTracker = { // contract exposes this as the running snapshot text for downstream coalescing, // so it stays a message-level concatenation, not a per-index value. emittedText: string; + currentSyntheticBlockIndex?: number; + nextSyntheticBlockIndex: number; + progressTokens: number; }; function createThinkingTracker(): ThinkingTracker { - return { streamedByIndex: new Map(), emittedText: "" }; + return { + streamedByIndex: new Map(), + emittedText: "", + nextSyntheticBlockIndex: 0, + progressTokens: 0, + }; +} + +function resetThinkingBlockState(tracker: ThinkingTracker): void { + tracker.streamedByIndex.clear(); + tracker.emittedText = ""; + tracker.currentSyntheticBlockIndex = undefined; + tracker.nextSyntheticBlockIndex = 0; + tracker.progressTokens = 0; } function resetThinkingTrackerForMessage( @@ -815,11 +835,39 @@ function resetThinkingTrackerForMessage( } // Anthropic content-block indexes restart at 0 for each message, so a prior // tool-round message's per-index thinking must not bleed into the next one. - tracker.streamedByIndex.clear(); - tracker.emittedText = ""; + resetThinkingBlockState(tracker); tracker.currentMessageId = messageId; } +function beginClaudeContentBlock(tracker: ThinkingTracker, index: unknown): void { + if (typeof index === "number") { + tracker.currentSyntheticBlockIndex = index; + tracker.nextSyntheticBlockIndex = Math.max(tracker.nextSyntheticBlockIndex, index + 1); + return; + } + if (index !== undefined) { + tracker.currentSyntheticBlockIndex = undefined; + return; + } + tracker.currentSyntheticBlockIndex = tracker.nextSyntheticBlockIndex; + tracker.nextSyntheticBlockIndex += 1; +} + +function stopClaudeContentBlock(tracker: ThinkingTracker): void { + tracker.currentSyntheticBlockIndex = undefined; +} + +function resolveClaudeContentBlockIndex(tracker: ThinkingTracker, index: unknown): number | null { + if (typeof index === "number") { + tracker.nextSyntheticBlockIndex = Math.max(tracker.nextSyntheticBlockIndex, index + 1); + return index; + } + if (index !== undefined) { + return null; + } + return tracker.currentSyntheticBlockIndex ?? null; +} + function assembleThinkingTextByIndex(streamedByIndex: Map): string { return [...streamedByIndex.entries()] .toSorted(([left], [right]) => left - right) @@ -839,12 +887,33 @@ function emitClaudeThinking( onThinkingDelta({ text: tracker.emittedText, delta, isReasoningSnapshot: true }); } +function readThinkingProgressTokens(delta: Record): number | undefined { + if (delta.type !== "thinking_delta" || delta.thinking !== "") { + return undefined; + } + const estimatedTokens = delta.estimated_tokens; + if (typeof estimatedTokens !== "number" || !Number.isFinite(estimatedTokens)) { + return undefined; + } + return estimatedTokens > 0 ? estimatedTokens : undefined; +} + +function emitClaudeThinkingProgress( + tracker: ThinkingTracker, + progressTokensDelta: number, + onThinkingProgress: (progress: CliThinkingProgress) => void, +): void { + tracker.progressTokens += progressTokensDelta; + onThinkingProgress({ progressTokens: tracker.progressTokens }); +} + function dispatchClaudeCliThinking(params: { backend: CliBackendConfig; providerId: string; parsed: Record; tracker: ThinkingTracker; - onThinkingDelta: (delta: CliThinkingDelta) => void; + onThinkingDelta?: (delta: CliThinkingDelta) => void; + onThinkingProgress?: (progress: CliThinkingProgress) => void; }): void { if (!supportsCliJsonlToolEvents(params)) { return; @@ -861,12 +930,26 @@ function dispatchClaudeCliThinking(params: { ); return; } + if (event.type === "content_block_start") { + beginClaudeContentBlock(tracker, event.index); + return; + } + if (event.type === "content_block_stop") { + stopClaudeContentBlock(tracker); + return; + } if (event.type !== "content_block_delta" || !isRecord(event.delta)) { return; } - // Thinking state is per content-block; a delta without a numeric index cannot - // be attributed to a block, so it is dropped rather than mixed into another. - if (typeof event.index !== "number") { + // Thinking state is per content-block; when the CLI omits indexes, the + // surrounding block start/stop stream supplies the ordering slot. + const blockIndex = resolveClaudeContentBlockIndex(tracker, event.index); + if (blockIndex === null) { + return; + } + const progressTokensDelta = readThinkingProgressTokens(event.delta); + if (progressTokensDelta !== undefined && params.onThinkingProgress) { + emitClaudeThinkingProgress(tracker, progressTokensDelta, params.onThinkingProgress); return; } // signature_delta carries opaque continuation material; the Claude CLI owns @@ -877,14 +960,11 @@ function dispatchClaudeCliThinking(params: { if (!event.delta.thinking) { return; } - const streamed = tracker.streamedByIndex.get(event.index) ?? ""; - emitClaudeThinking( - tracker, - event.index, - streamed, - event.delta.thinking, - params.onThinkingDelta, - ); + if (!params.onThinkingDelta) { + return; + } + const streamed = tracker.streamedByIndex.get(blockIndex) ?? ""; + emitClaudeThinking(tracker, blockIndex, streamed, event.delta.thinking, params.onThinkingDelta); return; } @@ -901,6 +981,9 @@ function dispatchClaudeCliThinking(params: { if (!isRecord(block) || block.type !== "thinking" || typeof block.thinking !== "string") { continue; } + if (!params.onThinkingDelta) { + continue; + } tracker.streamedByIndex.set(index, block.thinking); const text = assembleThinkingTextByIndex(tracker.streamedByIndex); if (text === tracker.emittedText) { @@ -989,6 +1072,7 @@ export function createCliJsonlStreamingParser(params: { providerId: string; onAssistantDelta: (delta: CliStreamingDelta) => void; onThinkingDelta?: (delta: CliThinkingDelta) => void; + onThinkingProgress?: (progress: CliThinkingProgress) => void; onToolUseStart?: (delta: CliToolUseStartDelta) => void; onToolResult?: (delta: CliToolResultDelta) => void; onCommentaryText?: (text: string) => void; @@ -1105,13 +1189,14 @@ export function createCliJsonlStreamingParser(params: { } } - if (params.onThinkingDelta) { + if (params.onThinkingDelta || params.onThinkingProgress) { dispatchClaudeCliThinking({ backend: params.backend, providerId: params.providerId, parsed, tracker: thinkingTracker, onThinkingDelta: params.onThinkingDelta, + onThinkingProgress: params.onThinkingProgress, }); } diff --git a/src/agents/cli-runner/claude-live-session.ts b/src/agents/cli-runner/claude-live-session.ts index 15d201bb582..7cea69b620d 100644 --- a/src/agents/cli-runner/claude-live-session.ts +++ b/src/agents/cli-runner/claude-live-session.ts @@ -3,9 +3,9 @@ */ import crypto from "node:crypto"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; -import { createAbortError as createNamedAbortError } from "../../infra/abort-signal.js"; import type { ReplyBackendHandle } from "../../auto-reply/reply/reply-run-registry.js"; import type { CliBackendConfig } from "../../config/types.js"; +import { createAbortError as createNamedAbortError } from "../../infra/abort-signal.js"; import { emitTrustedDiagnosticEvent, type DiagnosticToolParamsSummary, @@ -33,6 +33,7 @@ import { type CliStreamJsonOutputLimits, type CliStreamingDelta, type CliThinkingDelta, + type CliThinkingProgress, type CliToolResultDelta, type CliToolUseStartDelta, resolveCliStreamJsonOutputLimits, @@ -1099,6 +1100,7 @@ function createTurn(params: { noOutputTimeoutMs: number; onAssistantDelta: (delta: CliStreamingDelta) => void; onThinkingDelta?: (delta: CliThinkingDelta) => void; + onThinkingProgress?: (progress: CliThinkingProgress) => void; onToolUseStart?: (delta: CliToolUseStartDelta) => void; onToolResult?: (delta: CliToolResultDelta) => void; onCommentaryText?: (text: string) => void; @@ -1128,6 +1130,7 @@ function createTurn(params: { providerId: params.context.backendResolved.id, onAssistantDelta: params.onAssistantDelta, onThinkingDelta: params.onThinkingDelta, + onThinkingProgress: params.onThinkingProgress, onToolUseStart: params.onToolUseStart, onToolResult: params.onToolResult, onCommentaryText: params.onCommentaryText, @@ -1200,6 +1203,7 @@ export async function runClaudeLiveSessionTurn(params: { getProcessSupervisor: () => ProcessSupervisor; onAssistantDelta: (delta: CliStreamingDelta) => void; onThinkingDelta?: (delta: CliThinkingDelta) => void; + onThinkingProgress?: (progress: CliThinkingProgress) => void; onToolUseStart?: (delta: CliToolUseStartDelta) => void; onToolResult?: (delta: CliToolResultDelta) => void; onCommentaryText?: (text: string) => void; @@ -1324,6 +1328,7 @@ export async function runClaudeLiveSessionTurn(params: { noOutputTimeoutMs: params.noOutputTimeoutMs, onAssistantDelta: params.onAssistantDelta, onThinkingDelta: params.onThinkingDelta, + onThinkingProgress: params.onThinkingProgress, onToolUseStart: params.onToolUseStart, onToolResult: params.onToolResult, onCommentaryText: params.onCommentaryText, diff --git a/src/agents/cli-runner/execute.ts b/src/agents/cli-runner/execute.ts index ce89095d571..f377fb72d88 100644 --- a/src/agents/cli-runner/execute.ts +++ b/src/agents/cli-runner/execute.ts @@ -36,6 +36,7 @@ import { type CliOutput, type CliStreamingDelta, type CliThinkingDelta, + type CliThinkingProgress, } from "../cli-output.js"; import { classifyFailoverReason } from "../embedded-agent-helpers.js"; import { @@ -1035,11 +1036,7 @@ export async function executePreparedCliRun( // presentation. Text stays raw here to match the thinking-stream contract // shared with embedded-agent-subscribe, which archives untransformed // reasoning regardless of source. - const emitCliThinkingDelta = ({ - text, - delta, - isReasoningSnapshot, - }: CliThinkingDelta) => { + const emitCliThinkingDelta = ({ text, delta, isReasoningSnapshot }: CliThinkingDelta) => { if (text || delta) { observedCliActivity = true; } @@ -1052,6 +1049,17 @@ export async function executePreparedCliRun( data: { text, delta, ...(isReasoningSnapshot ? { isReasoningSnapshot } : {}) }, }); }; + const emitCliThinkingProgress = ({ progressTokens }: CliThinkingProgress) => { + observedCliActivity = true; + if (!emitLiveEvents) { + return; + } + emitAgentEvent({ + runId: params.runId, + stream: "thinking", + data: { progressTokens }, + }); + }; if (shouldUseClaudeLiveSession(context)) { if (!hasJsonlOutput) { throw new Error("Claude live session requires JSONL streaming parser"); @@ -1073,6 +1081,7 @@ export async function executePreparedCliRun( getProcessSupervisor: executeDeps.getProcessSupervisor, onAssistantDelta: emitCliAssistantDelta, onThinkingDelta: emitCliThinkingDelta, + onThinkingProgress: emitCliThinkingProgress, onToolUseStart: emitCliToolUseStart, onToolResult: emitCliToolResult, onCommentaryText: @@ -1101,6 +1110,7 @@ export async function executePreparedCliRun( providerId: context.backendResolved.id, onAssistantDelta: emitCliAssistantDelta, onThinkingDelta: emitCliThinkingDelta, + onThinkingProgress: emitCliThinkingProgress, onToolUseStart: emitCliToolUseStart, onToolResult: emitCliToolResult, onCommentaryText: diff --git a/src/auto-reply/get-reply-options.types.ts b/src/auto-reply/get-reply-options.types.ts index ed858a17952..1f73b15aa3e 100644 --- a/src/auto-reply/get-reply-options.types.ts +++ b/src/auto-reply/get-reply-options.types.ts @@ -67,6 +67,10 @@ type ReasoningStreamPayload = Pick< requiresReasoningProgressOptIn?: boolean; }; +type ReasoningProgressPayload = { + progressTokens: number; +}; + /** Reply generation options shared by auto-reply, webchat, channels, and tests. */ export type GetReplyOptions = { /** Override run id for agent events (defaults to random UUID). */ @@ -133,6 +137,7 @@ export type GetReplyOptions = { onVerboseProgressVisibility?: (isActive: () => boolean) => void; onPartialReply?: (payload: PartialReplyPayload) => Promise | void; onReasoningStream?: (payload: ReasoningStreamPayload) => Promise | void; + onReasoningProgress?: (payload: ReasoningProgressPayload) => Promise | void; streamReasoningInNonStreamModes?: boolean; /** Called when a thinking/reasoning block ends. */ onReasoningEnd?: () => Promise | void; diff --git a/src/auto-reply/reply/agent-runner-cli-dispatch.test.ts b/src/auto-reply/reply/agent-runner-cli-dispatch.test.ts index be003a61045..2f69d68f753 100644 --- a/src/auto-reply/reply/agent-runner-cli-dispatch.test.ts +++ b/src/auto-reply/reply/agent-runner-cli-dispatch.test.ts @@ -8,7 +8,10 @@ import { onAgentEvent, resetAgentEventsForTest, } from "../../infra/agent-events.js"; -import type { ReasoningTextPayload } from "./agent-runner-cli-dispatch.js"; +import type { + ReasoningProgressPayload, + ReasoningTextPayload, +} from "./agent-runner-cli-dispatch.js"; import { createCliToolSummaryTracker, keepCliSessionBindingOnlyWhenReused, @@ -120,6 +123,53 @@ describe("runCliAgentWithLifecycle", () => { expect(result.payloads).toEqual([{ text: "Only thinking more", isReasoning: true }]); }); + it("bridges thinking token progress without adding durable reasoning", async () => { + cliDispatchState.runCliAgentMock.mockImplementationOnce(async (params: { runId: string }) => { + emitAgentEvent({ + runId: params.runId, + stream: "thinking", + data: { progressTokens: 50 }, + }); + emitAgentEvent({ + runId: params.runId, + stream: "thinking", + data: { progressTokens: 50 }, + }); + emitAgentEvent({ + runId: params.runId, + stream: "thinking", + data: { progressTokens: 200 }, + }); + return { payloads: [{ text: "Visible answer" }], meta: { durationMs: 1 } }; + }); + const onReasoningProgress = vi.fn<(payload: ReasoningProgressPayload) => Promise>( + async () => undefined, + ); + + const result = await runCliAgentWithLifecycle({ + runId: "run-thinking-progress", + provider: "claude-cli", + onReasoningProgress, + runParams: { + sessionId: "session-1", + sessionFile: "/tmp/session.jsonl", + workspaceDir: "/tmp/workspace", + prompt: "hello", + provider: "claude-cli", + model: "claude", + thinkLevel: "high", + timeoutMs: 1_000, + runId: "run-thinking-progress", + }, + }); + + expect(onReasoningProgress.mock.calls.map((call) => call[0])).toEqual([ + { progressTokens: 50 }, + { progressTokens: 200 }, + ]); + expect(result.payloads).toEqual([{ text: "Visible answer" }]); + }); + it("does not add a durable reasoning payload when the CLI emits no thinking", async () => { cliDispatchState.runCliAgentMock.mockResolvedValueOnce({ payloads: [{ text: "Visible answer" }], diff --git a/src/auto-reply/reply/agent-runner-cli-dispatch.ts b/src/auto-reply/reply/agent-runner-cli-dispatch.ts index 06fb7287246..12b32767e87 100644 --- a/src/auto-reply/reply/agent-runner-cli-dispatch.ts +++ b/src/auto-reply/reply/agent-runner-cli-dispatch.ts @@ -115,6 +115,10 @@ export type ReasoningTextPayload = { isReasoningSnapshot?: boolean; }; +export type ReasoningProgressPayload = { + progressTokens: number; +}; + export function createCliReasoningStreamBridge( onReasoningStream: GetReplyOptions["onReasoningStream"] | undefined, ): ((payload: ReasoningTextPayload) => Promise) | undefined { @@ -157,6 +161,35 @@ function createReasoningTextBridge(params: { }); } +function createReasoningProgressBridge(params: { + runId: string; + suppressed?: boolean; + deliver?: (payload: ReasoningProgressPayload) => Promise; +}) { + let lastProgressTokens: number | undefined; + return createAgentEventBridge({ + runId: params.runId, + suppressed: params.suppressed, + deliver: params.deliver, + read: (evt) => { + if (evt.stream !== "thinking") { + return undefined; + } + const progressTokens = evt.data.progressTokens; + if ( + typeof progressTokens !== "number" || + !Number.isFinite(progressTokens) || + progressTokens <= 0 || + progressTokens === lastProgressTokens + ) { + return undefined; + } + lastProgressTokens = progressTokens; + return { progressTokens }; + }, + }); +} + type CommentaryTextPayload = { text: string; itemId?: string; @@ -380,6 +413,7 @@ type RunCliAgentWithLifecycleParams = { suppressAssistantBridge?: boolean; onAssistantText?: (text: string) => Promise; onReasoningText?: (payload: ReasoningTextPayload) => Promise; + onReasoningProgress?: (payload: ReasoningProgressPayload) => Promise; onToolEvent?: (payload: CliToolEventPayload) => Promise; onCommentaryText?: (payload: CommentaryTextPayload) => Promise; onFastModeAutoProgress?: (payload: ReplyPayload) => Promise; @@ -496,6 +530,11 @@ async function runCliAgentWithLifecycleInternal( await params.onReasoningText?.(payload); }, }); + const reasoningProgressBridge = createReasoningProgressBridge({ + runId: params.runId, + suppressed: params.suppressAssistantBridge, + deliver: params.onReasoningProgress, + }); const toolBridge = createToolEventBridge({ runId: params.runId, suppressed: params.suppressAssistantBridge, @@ -514,6 +553,7 @@ async function runCliAgentWithLifecycleInternal( const bridges = [ assistantBridge, reasoningBridge, + reasoningProgressBridge, toolBridge, commentaryBridge, toolBoundaryBridge, diff --git a/src/auto-reply/reply/agent-runner-execution.ts b/src/auto-reply/reply/agent-runner-execution.ts index 77da977d795..688be7a23c7 100644 --- a/src/auto-reply/reply/agent-runner-execution.ts +++ b/src/auto-reply/reply/agent-runner-execution.ts @@ -2370,6 +2370,9 @@ async function runAgentTurnWithFallbackInternal( await params.opts.onPartialReply({ text: textForTyping }); }, onReasoningText: createCliReasoningStreamBridge(params.opts?.onReasoningStream), + onReasoningProgress: async (payload) => { + await params.opts?.onReasoningProgress?.(payload); + }, onToolEvent: async (payload) => { await cliToolSummaryTracker.noteToolEvent(payload); if (payload.phase === "result") { diff --git a/src/auto-reply/reply/followup-runner.ts b/src/auto-reply/reply/followup-runner.ts index 4163d81bea5..fda57580894 100644 --- a/src/auto-reply/reply/followup-runner.ts +++ b/src/auto-reply/reply/followup-runner.ts @@ -1023,6 +1023,9 @@ export function createFollowupRunner(params: { onAgentRunStart: () => opts?.onAgentRunStart?.(runId), suppressAssistantBridge: run.silentExpected, onReasoningText: createCliReasoningStreamBridge(opts?.onReasoningStream), + onReasoningProgress: async (payload) => { + await opts?.onReasoningProgress?.(payload); + }, onToolEvent: async (payload) => { await cliToolSummaryTracker.noteToolEvent(payload); if (payload.phase === "result") { diff --git a/test/fixtures/cli/claude-2.1-thinking-progress.jsonl b/test/fixtures/cli/claude-2.1-thinking-progress.jsonl new file mode 100644 index 00000000000..28c81cbfd02 --- /dev/null +++ b/test/fixtures/cli/claude-2.1-thinking-progress.jsonl @@ -0,0 +1,20 @@ +{"type":"stream_event","event":{"type":"message_start","message":{"model":"claude-opus-4-8","id":"msg_01JsEVMBeuv1k63xPve8duB8","type":"message","role":"assistant","content":[],"stop_reason":null,"stop_sequence":null,"stop_details":null,"usage":{"input_tokens":4418,"cache_creation_input_tokens":36955,"cache_read_input_tokens":0,"cache_creation":{"ephemeral_5m_input_tokens":0,"ephemeral_1h_input_tokens":36955},"output_tokens":5,"service_tier":"standard","inference_geo":"not_available"},"diagnostics":null}},"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","parent_tool_use_id":null,"uuid":"3716b499-8d5b-46ef-bd50-f2334af08fb0","ttft_ms":1513} +{"type":"stream_event","event":{"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":"","signature":""}},"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","parent_tool_use_id":null,"uuid":"f09f5288-88c7-44d1-9303-e3a651ffc5e0"} +{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"","estimated_tokens":50}},"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","parent_tool_use_id":null,"uuid":"ab52bbf3-f11d-4220-bf0e-32699dc1be25"} +{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"","estimated_tokens":150}},"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","parent_tool_use_id":null,"uuid":"40e59622-cc0c-4fd6-b544-88aa76a546b6"} +{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"","estimated_tokens":100}},"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","parent_tool_use_id":null,"uuid":"e9ad6ce0-6f50-46bb-8bcd-9cf33d3c3e2b"} +{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"ErYHCmMIDxgCKkCifXPKYKmFFEibgywq1/nvaQBF0KOx1ici86UBjamFq/7bRsSKiXxnlSe7EHIAZEPU+KgLGwnwM1zyEmHT6OmqMg9jbGF1ZGUtb3B1cy00LTg4AEIIdGhpbmtpbmcSDP5ewS+HCtROMWDAGRoMzyp/NzcaFYKo2pnnIjC34OLtaXwrWGYUBj4/oBGtTSmwF+53jkYQ9IuBehdrq4RbeG8NYLzwgeekN4C2U8kqgAb9Rt08ksgOwVuOtirGx9jLDj33DHTFDuYi18+eNdXzwd6mxRz13hugTEwDxfZe19TWuwEgvsu+SRPTvsIinNFYVuIVl9SMfHeDwRdSYz3UuPmOogvLdJcIb+0DT1GlMnBmiU/SPBGoSlsXBxv+bp2n8IY1BS8Y2rtXR8knCssMIINGg+cvX6NZ+9gAcXwhA0UjNG81xW3kd+fenUCWWeaHocmUnLl/h11SiPlb5X5+9cHReAilONkN+DJx+5Mohbt+E3ss/6qgWFJ4Zb+BLFS50w72+Rc7vLlX6ityXtrasuKmiiO7/v4YkCEl4XBfh71ARg5o+COwkz9uyPJXfysjeAYlpOS3GPAsYlmyIjXgQHvFEJjXQJV4Fizy6Il0vrVgLY9RFVDuKArq+QhsE2Cm4fFA86kAGEI6q7I7uGGcsh3LsJZj2m0Mz3hCCZ7uHaN5JZpeUAJJ2L0osxS9q+VNEhLFTK9SXk13ZDv7NYl3WrlV7x4sG9MLjMNvTnm3TyCfNHWLpOvASaFdZvM6w8zBIiis46JALHdi1gQWON40JahXoHTDb/Jrem4JkEIGk37NwJWGeNmlUI1FMipEDcSpklOIKnFR1DCWG1GwtlqsZ1kxKlvzsYcsSvtuvIeWy7Mu+SmdamDr1WMfqmgmu0MWEVBFUg6ijfxGrE4Y0JRVMW3VhU87eYDACrKt59pMMLcWKgESeVmORlex7xv7hlu/9PZWnGs5pl0eDPiFuaHQmG/AZ1FkJ+cbzKoP7jV3xAL8zT4EFv7/l04jww+lUap3C2PNeDEzMf9nv2s7fo+0vPKrbmPByqVtoZb+0DPDqIiJH+UTK9ZCIg4lJEu7dRd5Xftk61n84sKopFnwqkhBeXij96DWIB8b30qN2+JUX29E/IOCG4tK2R3HY7bxMf0THh4bNSDg3ZLTqJ9Z/zhxKKcP/7kzXhhr7O4o30gFdv+8G+8nvEaJ7HlcN9sjOr+tK8N+KpPGJXtOmNx8GNuqKGJJ3jAWSJytZ5x7N+DH70IYAQ=="}},"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","parent_tool_use_id":null,"uuid":"a2818f52-45ae-493a-842f-90bed83facba"} +{"type":"assistant","message":{"model":"claude-opus-4-8","id":"msg_01JsEVMBeuv1k63xPve8duB8","type":"message","role":"assistant","content":[{"type":"thinking","thinking":"","signature":"ErYHCmMIDxgCKkCifXPKYKmFFEibgywq1/nvaQBF0KOx1ici86UBjamFq/7bRsSKiXxnlSe7EHIAZEPU+KgLGwnwM1zyEmHT6OmqMg9jbGF1ZGUtb3B1cy00LTg4AEIIdGhpbmtpbmcSDP5ewS+HCtROMWDAGRoMzyp/NzcaFYKo2pnnIjC34OLtaXwrWGYUBj4/oBGtTSmwF+53jkYQ9IuBehdrq4RbeG8NYLzwgeekN4C2U8kqgAb9Rt08ksgOwVuOtirGx9jLDj33DHTFDuYi18+eNdXzwd6mxRz13hugTEwDxfZe19TWuwEgvsu+SRPTvsIinNFYVuIVl9SMfHeDwRdSYz3UuPmOogvLdJcIb+0DT1GlMnBmiU/SPBGoSlsXBxv+bp2n8IY1BS8Y2rtXR8knCssMIINGg+cvX6NZ+9gAcXwhA0UjNG81xW3kd+fenUCWWeaHocmUnLl/h11SiPlb5X5+9cHReAilONkN+DJx+5Mohbt+E3ss/6qgWFJ4Zb+BLFS50w72+Rc7vLlX6ityXtrasuKmiiO7/v4YkCEl4XBfh71ARg5o+COwkz9uyPJXfysjeAYlpOS3GPAsYlmyIjXgQHvFEJjXQJV4Fizy6Il0vrVgLY9RFVDuKArq+QhsE2Cm4fFA86kAGEI6q7I7uGGcsh3LsJZj2m0Mz3hCCZ7uHaN5JZpeUAJJ2L0osxS9q+VNEhLFTK9SXk13ZDv7NYl3WrlV7x4sG9MLjMNvTnm3TyCfNHWLpOvASaFdZvM6w8zBIiis46JALHdi1gQWON40JahXoHTDb/Jrem4JkEIGk37NwJWGeNmlUI1FMipEDcSpklOIKnFR1DCWG1GwtlqsZ1kxKlvzsYcsSvtuvIeWy7Mu+SmdamDr1WMfqmgmu0MWEVBFUg6ijfxGrE4Y0JRVMW3VhU87eYDACrKt59pMMLcWKgESeVmORlex7xv7hlu/9PZWnGs5pl0eDPiFuaHQmG/AZ1FkJ+cbzKoP7jV3xAL8zT4EFv7/l04jww+lUap3C2PNeDEzMf9nv2s7fo+0vPKrbmPByqVtoZb+0DPDqIiJH+UTK9ZCIg4lJEu7dRd5Xftk61n84sKopFnwqkhBeXij96DWIB8b30qN2+JUX29E/IOCG4tK2R3HY7bxMf0THh4bNSDg3ZLTqJ9Z/zhxKKcP/7kzXhhr7O4o30gFdv+8G+8nvEaJ7HlcN9sjOr+tK8N+KpPGJXtOmNx8GNuqKGJJ3jAWSJytZ5x7N+DH70IYAQ=="}],"stop_reason":null,"stop_sequence":null,"stop_details":null,"usage":{"input_tokens":4418,"cache_creation_input_tokens":36955,"cache_read_input_tokens":0,"cache_creation":{"ephemeral_5m_input_tokens":0,"ephemeral_1h_input_tokens":36955},"output_tokens":5,"service_tier":"standard","inference_geo":"not_available"},"diagnostics":null,"context_management":null},"parent_tool_use_id":null,"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","uuid":"72385cdf-f707-406e-97b1-be9fcc83b7d7","request_id":"req_011Cci7ovWFi58ZBU9JpBwkc"} +{"type":"stream_event","event":{"type":"content_block_stop","index":0},"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","parent_tool_use_id":null,"uuid":"25739953-468a-45d2-9f06-02138cc2ee99"} +{"type":"stream_event","event":{"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}},"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","parent_tool_use_id":null,"uuid":"e4b71600-2ba8-4eb9-9d1e-a67f39db01e7"} +{"type":"stream_event","event":{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"**9 per barn.**\n\nStep by step:\n\n1. **\"All but 9 die\"** — the trick line. This means 9 survive ("}},"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","parent_tool_use_id":null,"uuid":"56f7aec5-bf98-4cec-b758-ed511b82b952"} +{"type":"stream_event","event":{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"all die *except* 9). Not 17−9. → **9 remain.**\n2. **Buys twice as many as remain** — 2 × 9 "}},"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","parent_tool_use_id":null,"uuid":"8f42a7a8-6c5d-4f7b-9216-688d35b6ee11"} +{"type":"stream_event","event":{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"= 18 bought. Total = 9 + 18 = **27.**\n3. **Sells a third of the total** — ⅓ × 27 = 9 sold. Left"}},"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","parent_tool_use_id":null,"uuid":"2c80d31b-db36-403a-9160-77bea2fbece7"} +{"type":"stream_event","event":{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":": 27 − 9 = **18.**\n4. **Split between 2 barns** — 18 ÷ 2 = **9 per barn.**\n\nThe 17"}},"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","parent_tool_use_id":null,"uuid":"d2213f2b-43ad-4699-adbe-9d78bf341f42"} +{"type":"stream_event","event":{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":" is a decoy — it never actually matters once you read \"all but 9\" corre"}},"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","parent_tool_use_id":null,"uuid":"b8ce2f7a-adc0-4a54-acf4-8334b2060043"} +{"type":"stream_event","event":{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"ctly."}},"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","parent_tool_use_id":null,"uuid":"e7ccba3e-c0e1-48ae-88af-c74e654c909c"} +{"type":"assistant","message":{"model":"claude-opus-4-8","id":"msg_01JsEVMBeuv1k63xPve8duB8","type":"message","role":"assistant","content":[{"type":"text","text":"**9 per barn.**\n\nStep by step:\n\n1. **\"All but 9 die\"** — the trick line. This means 9 survive (all die *except* 9). Not 17−9. → **9 remain.**\n2. **Buys twice as many as remain** — 2 × 9 = 18 bought. Total = 9 + 18 = **27.**\n3. **Sells a third of the total** — ⅓ × 27 = 9 sold. Left: 27 − 9 = **18.**\n4. **Split between 2 barns** — 18 ÷ 2 = **9 per barn.**\n\nThe 17 is a decoy — it never actually matters once you read \"all but 9\" correctly."}],"stop_reason":null,"stop_sequence":null,"stop_details":null,"usage":{"input_tokens":4418,"cache_creation_input_tokens":36955,"cache_read_input_tokens":0,"cache_creation":{"ephemeral_5m_input_tokens":0,"ephemeral_1h_input_tokens":36955},"output_tokens":5,"service_tier":"standard","inference_geo":"not_available"},"diagnostics":null,"context_management":null},"parent_tool_use_id":null,"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","uuid":"dcdf9c92-9171-46e2-833b-1111c7173de6","request_id":"req_011Cci7ovWFi58ZBU9JpBwkc"} +{"type":"stream_event","event":{"type":"content_block_stop","index":1},"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","parent_tool_use_id":null,"uuid":"2a1beb51-1812-4cfc-82d5-0c8151db1205"} +{"type":"stream_event","event":{"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null,"stop_details":null},"usage":{"input_tokens":4418,"cache_creation_input_tokens":36955,"cache_read_input_tokens":0,"output_tokens":534,"output_tokens_details":{"thinking_tokens":315},"iterations":[{"input_tokens":4418,"output_tokens":534,"cache_read_input_tokens":0,"cache_creation_input_tokens":36955,"cache_creation":{"ephemeral_5m_input_tokens":0,"ephemeral_1h_input_tokens":36955},"type":"message"}]},"context_management":{"applied_edits":[]}},"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","parent_tool_use_id":null,"uuid":"585b9722-efc4-41ea-b0d1-665e7d463a4c"} +{"type":"stream_event","event":{"type":"message_stop"},"session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","parent_tool_use_id":null,"uuid":"b8a17dc4-5a9e-4056-82e4-8c847d02a3c5"} +{"type":"result","subtype":"success","is_error":false,"api_error_status":null,"duration_ms":9800,"duration_api_ms":9423,"ttft_ms":6051,"ttft_stream_ms":1549,"time_to_request_ms":36,"num_turns":1,"result":"**9 per barn.**\n\nStep by step:\n\n1. **\"All but 9 die\"** — the trick line. This means 9 survive (all die *except* 9). Not 17−9. → **9 remain.**\n2. **Buys twice as many as remain** — 2 × 9 = 18 bought. Total = 9 + 18 = **27.**\n3. **Sells a third of the total** — ⅓ × 27 = 9 sold. Left: 27 − 9 = **18.**\n4. **Split between 2 barns** — 18 ÷ 2 = **9 per barn.**\n\nThe 17 is a decoy — it never actually matters once you read \"all but 9\" correctly.","stop_reason":"end_turn","session_id":"6f8fa5d2-520b-4d0c-8731-3439b3dcabbf","total_cost_usd":0.40563000000000005,"usage":{"input_tokens":4418,"cache_creation_input_tokens":36955,"cache_read_input_tokens":0,"output_tokens":534,"server_tool_use":{"web_search_requests":0,"web_fetch_requests":0},"service_tier":"standard","cache_creation":{"ephemeral_1h_input_tokens":36955,"ephemeral_5m_input_tokens":0},"inference_geo":"not_available","iterations":[{"input_tokens":4418,"output_tokens":534,"cache_read_input_tokens":0,"cache_creation_input_tokens":36955,"cache_creation":{"ephemeral_5m_input_tokens":0,"ephemeral_1h_input_tokens":36955},"type":"message"}],"speed":"standard"},"modelUsage":{"claude-haiku-4-5-20251001":{"inputTokens":570,"outputTokens":14,"cacheReadInputTokens":0,"cacheCreationInputTokens":0,"webSearchRequests":0,"costUSD":0.0006399999999999999,"contextWindow":200000,"maxOutputTokens":32000},"claude-opus-4-8":{"inputTokens":4418,"outputTokens":534,"cacheReadInputTokens":0,"cacheCreationInputTokens":36955,"webSearchRequests":0,"costUSD":0.40499000000000007,"contextWindow":1000000,"maxOutputTokens":64000}},"permission_denials":[],"terminal_reason":"completed","fast_mode_state":"off","uuid":"fa8d7d86-b09e-42d6-9085-6ec6aa024186"}