Skip to content

Commit ab98e51

Browse files
Marvintheboredclaude
authored and
Peetiegonzalez
committed
feat(cli): render claude CLI native thinking with /reasoning gating
The claude-cli (claude-stdio) live-session backend emits native reasoning on stream-json (`thinking_delta` + a snapshot `thinking` block), but the CLI parser had no thinking handling at all, so it was silently dropped on every surface in every /reasoning mode. - cli-output.ts: parse thinking_delta/snapshot into stream:"thinking" with replace-style per-content-block-index dedupe (assembleThinkingTextByIndex), robust to tool-interleaved multi-block turns (snapshot replaces per-index rather than assuming a streamed prefix). signature_delta/redacted_thinking stay skipped. - agent-runner-cli-dispatch.ts: bridge stream:"thinking" into onReasoningStream via a new createReasoningTextBridge (mirrors the existing assistant-text bridge), replacing the old assistant-text-as-reasoning heuristic. Track the final reasoning snapshot and, when present, prepend a durable `{text, isReasoning: true}` payload before the answer — mirroring the embedded-runner durable reasoning payload (embedded-agent-runner/run/payloads.ts). - agent-runner-execution.ts: thread isReasoningSnapshot through to onReasoningStream, still passing requiresReasoningProgressOptIn:true so CLI reasoning rides the exact same channel gates as embedded reasoning (Discord/Telegram): /reasoning off -> nothing on any surface, stream -> window preview only, on -> durable payload too. The bridge is suppressed under the same conditions as the assistant bridge (e.g. silentExpected follow-ups), so no durable payload is synthesized for turns that are expected to stay silent. Supersedes this branch's prior commit (a straight port of af7eb48e05's emit-always model, no gating and no durable mode) with the above gated/durable design per product decision. Also, in cli-output.ts: reset the per-content-block-index thinking tracker on every new Anthropic message (message_start, or a fresh top-level "assistant" snapshot's message.id) instead of leaving it scoped to the whole CLI turn. Content-block indices restart at 0 per Anthropic message, so a tool round-trip within one turn (message A: thinking + tool_use; tool result fed back; message B: a fresh message_start) was letting message A's stale index-0 thinking text bleed into message B's index-0 delta. And in agent-runner-cli-dispatch.ts: attach the durable reasoning payload whenever any thinking was bridged, not only when the CLI also produced a visible final answer, so a thinking-only turn (no visible reply) doesn't silently drop its reasoning. Round 2: the durable `{isReasoning: true}` payload synthesized above was also reaching the QUEUED FOLLOW-UP delivery path (followup-runner.ts -> resolveFollowupDeliveryPayloads) with no gate at all, so a claude-cli queued follow-up with /reasoning off would leak internal reasoning to the channel as an ordinary visible message. resolveFollowupDeliveryPayloads (followup-delivery.ts) now takes a reasoningPayloadsEnabled flag and drops isReasoning payloads unless it is true; followup-runner.ts threads opts?.reasoningPayloadsEnabled through at both call sites, from the same GetReplyOptions the direct path already reads (dispatch-from-config.ts:2054). This makes the queued-follow-up path apply the identical isReasoning/ reasoningPayloadsEnabled gate that direct dispatch already enforces (dispatch-from-config.ts:3347, :3537) — no new mechanism, CLI follow-ups now go through the exact same gate embedded follow-ups do. Scope note: routeReply's own unconditional suppression of isReasoning payloads on the origin-routing branch (route-reply.ts:131, shouldSuppressReasoningPayload) is pre-existing, shared with the embedded runner, and unchanged by this commit — that gate governs whether a kept-and-routed reasoning payload is actually rendered on a cross-channel queued follow-up, and is out of scope here. This commit closes the leak (off -> never eligible on any path); it does not change what happens to an eligible payload once it reaches routeReply. Co-Authored-By: Claude <[email protected]>
1 parent 4e6933c commit ab98e51

12 files changed

Lines changed: 918 additions & 30 deletions

‎src/agents/cli-output.test.ts‎

Lines changed: 395 additions & 0 deletions
Large diffs are not rendered by default.

‎src/agents/cli-output.ts‎

Lines changed: 146 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,13 @@ export type CliStreamJsonOutputLimits = {
7676
maxTurnLines: number;
7777
};
7878

79+
/** Incremental thinking text emitted while parsing a streaming CLI response. */
80+
export type CliThinkingDelta = {
81+
text: string;
82+
delta: string;
83+
isReasoningSnapshot?: boolean;
84+
};
85+
7986
/** Tool-call start event reconstructed from CLI stream output. */
8087
export type CliToolUseStartDelta = {
8188
toolCallId: string;
@@ -778,6 +785,133 @@ function dispatchClaudeCliStreamingToolEvent(params: {
778785
}
779786
}
780787

788+
type ThinkingTracker = {
789+
currentMessageId?: string;
790+
// Thinking text already streamed via thinking_delta, keyed by the Anthropic
791+
// content-block index. Snapshot frames repeat streamed thinking, so each block
792+
// is deduped against its own index; a single global concatenation misfires
793+
// once a message carries more than one thinking block (re-emits or reorders).
794+
streamedByIndex: Map<number, string>;
795+
// Full thinking already emitted for the message in block order. The callback
796+
// contract exposes this as the running snapshot text for downstream coalescing,
797+
// so it stays a message-level concatenation, not a per-index value.
798+
emittedText: string;
799+
};
800+
801+
function createThinkingTracker(): ThinkingTracker {
802+
return { streamedByIndex: new Map(), emittedText: "" };
803+
}
804+
805+
function resetThinkingTrackerForMessage(
806+
tracker: ThinkingTracker,
807+
messageId: string | undefined,
808+
): void {
809+
if (messageId && messageId === tracker.currentMessageId) {
810+
return;
811+
}
812+
if (messageId && tracker.currentMessageId === undefined) {
813+
tracker.currentMessageId = messageId;
814+
return;
815+
}
816+
// Anthropic content-block indexes restart at 0 for each message, so a prior
817+
// tool-round message's per-index thinking must not bleed into the next one.
818+
tracker.streamedByIndex.clear();
819+
tracker.emittedText = "";
820+
tracker.currentMessageId = messageId;
821+
}
822+
823+
function assembleThinkingTextByIndex(streamedByIndex: Map<number, string>): string {
824+
return [...streamedByIndex.entries()]
825+
.toSorted(([left], [right]) => left - right)
826+
.map(([, text]) => text)
827+
.join("");
828+
}
829+
830+
function emitClaudeThinking(
831+
tracker: ThinkingTracker,
832+
index: number,
833+
streamed: string,
834+
delta: string,
835+
onThinkingDelta: (delta: CliThinkingDelta) => void,
836+
): void {
837+
tracker.streamedByIndex.set(index, `${streamed}${delta}`);
838+
tracker.emittedText = assembleThinkingTextByIndex(tracker.streamedByIndex);
839+
onThinkingDelta({ text: tracker.emittedText, delta, isReasoningSnapshot: true });
840+
}
841+
842+
function dispatchClaudeCliThinking(params: {
843+
backend: CliBackendConfig;
844+
providerId: string;
845+
parsed: Record<string, unknown>;
846+
tracker: ThinkingTracker;
847+
onThinkingDelta: (delta: CliThinkingDelta) => void;
848+
}): void {
849+
if (!supportsCliJsonlToolEvents(params)) {
850+
return;
851+
}
852+
const tracker = params.tracker;
853+
854+
if (params.parsed.type === "stream_event" && isRecord(params.parsed.event)) {
855+
const event = params.parsed.event;
856+
if (event.type === "message_start") {
857+
const message = isRecord(event.message) ? event.message : undefined;
858+
resetThinkingTrackerForMessage(
859+
tracker,
860+
typeof message?.id === "string" ? message.id : undefined,
861+
);
862+
return;
863+
}
864+
if (event.type !== "content_block_delta" || !isRecord(event.delta)) {
865+
return;
866+
}
867+
// Thinking state is per content-block; a delta without a numeric index cannot
868+
// be attributed to a block, so it is dropped rather than mixed into another.
869+
if (typeof event.index !== "number") {
870+
return;
871+
}
872+
// signature_delta carries opaque continuation material; the Claude CLI owns
873+
// its own session transcript, so it never enters the thinking text lane.
874+
if (event.delta.type !== "thinking_delta" || typeof event.delta.thinking !== "string") {
875+
return;
876+
}
877+
if (!event.delta.thinking) {
878+
return;
879+
}
880+
const streamed = tracker.streamedByIndex.get(event.index) ?? "";
881+
emitClaudeThinking(
882+
tracker,
883+
event.index,
884+
streamed,
885+
event.delta.thinking,
886+
params.onThinkingDelta,
887+
);
888+
return;
889+
}
890+
891+
if (params.parsed.type === "assistant" && isRecord(params.parsed.message)) {
892+
resetThinkingTrackerForMessage(
893+
tracker,
894+
typeof params.parsed.message.id === "string" ? params.parsed.message.id : undefined,
895+
);
896+
const content = Array.isArray(params.parsed.message.content)
897+
? params.parsed.message.content
898+
: [];
899+
for (const [index, block] of content.entries()) {
900+
// redacted_thinking blocks are opaque provider material with no text lane.
901+
if (!isRecord(block) || block.type !== "thinking" || typeof block.thinking !== "string") {
902+
continue;
903+
}
904+
tracker.streamedByIndex.set(index, block.thinking);
905+
const text = assembleThinkingTextByIndex(tracker.streamedByIndex);
906+
if (text === tracker.emittedText) {
907+
continue;
908+
}
909+
tracker.emittedText = text;
910+
params.onThinkingDelta({ text, delta: block.thinking, isReasoningSnapshot: true });
911+
}
912+
}
913+
}
914+
781915
function dispatchGeminiCliStreamingToolEvent(params: {
782916
backend: CliBackendConfig;
783917
providerId: string;
@@ -854,6 +988,7 @@ export function createCliJsonlStreamingParser(params: {
854988
backend: CliBackendConfig;
855989
providerId: string;
856990
onAssistantDelta: (delta: CliStreamingDelta) => void;
991+
onThinkingDelta?: (delta: CliThinkingDelta) => void;
857992
onToolUseStart?: (delta: CliToolUseStartDelta) => void;
858993
onToolResult?: (delta: CliToolResultDelta) => void;
859994
onCommentaryText?: (text: string) => void;
@@ -874,6 +1009,7 @@ export function createCliJsonlStreamingParser(params: {
8741009
// always has a destination; a separate enable flag let it be dropped (#92092).
8751010
const classifyClaudeCommentary =
8761011
Boolean(params.onCommentaryText) && supportsCliJsonlToolEvents(params);
1012+
const thinkingTracker = createThinkingTracker();
8771013

8781014
const flushPendingClaudeAssistantText = () => {
8791015
if (!pendingClaudeText) {
@@ -969,6 +1105,16 @@ export function createCliJsonlStreamingParser(params: {
9691105
}
9701106
}
9711107

1108+
if (params.onThinkingDelta) {
1109+
dispatchClaudeCliThinking({
1110+
backend: params.backend,
1111+
providerId: params.providerId,
1112+
parsed,
1113+
tracker: thinkingTracker,
1114+
onThinkingDelta: params.onThinkingDelta,
1115+
});
1116+
}
1117+
9721118
if (params.onToolUseStart || params.onToolResult) {
9731119
dispatchGeminiCliStreamingToolEvent({
9741120
backend: params.backend,

‎src/agents/cli-runner/claude-live-session.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@ import {
3131
type CliOutput,
3232
type CliStreamJsonOutputLimits,
3333
type CliStreamingDelta,
34+
type CliThinkingDelta,
3435
type CliToolResultDelta,
3536
type CliToolUseStartDelta,
3637
resolveCliStreamJsonOutputLimits,
@@ -1098,6 +1099,7 @@ function createTurn(params: {
10981099
context: PreparedCliRunContext;
10991100
noOutputTimeoutMs: number;
11001101
onAssistantDelta: (delta: CliStreamingDelta) => void;
1102+
onThinkingDelta?: (delta: CliThinkingDelta) => void;
11011103
onToolUseStart?: (delta: CliToolUseStartDelta) => void;
11021104
onToolResult?: (delta: CliToolResultDelta) => void;
11031105
onCommentaryText?: (text: string) => void;
@@ -1126,6 +1128,7 @@ function createTurn(params: {
11261128
backend: params.context.preparedBackend.backend,
11271129
providerId: params.context.backendResolved.id,
11281130
onAssistantDelta: params.onAssistantDelta,
1131+
onThinkingDelta: params.onThinkingDelta,
11291132
onToolUseStart: params.onToolUseStart,
11301133
onToolResult: params.onToolResult,
11311134
onCommentaryText: params.onCommentaryText,
@@ -1197,6 +1200,7 @@ export async function runClaudeLiveSessionTurn(params: {
11971200
noOutputTimeoutMs: number;
11981201
getProcessSupervisor: () => ProcessSupervisor;
11991202
onAssistantDelta: (delta: CliStreamingDelta) => void;
1203+
onThinkingDelta?: (delta: CliThinkingDelta) => void;
12001204
onToolUseStart?: (delta: CliToolUseStartDelta) => void;
12011205
onToolResult?: (delta: CliToolResultDelta) => void;
12021206
onCommentaryText?: (text: string) => void;
@@ -1320,6 +1324,7 @@ export async function runClaudeLiveSessionTurn(params: {
13201324
context: params.context,
13211325
noOutputTimeoutMs: params.noOutputTimeoutMs,
13221326
onAssistantDelta: params.onAssistantDelta,
1327+
onThinkingDelta: params.onThinkingDelta,
13231328
onToolUseStart: params.onToolUseStart,
13241329
onToolResult: params.onToolResult,
13251330
onCommentaryText: params.onCommentaryText,

‎src/agents/cli-runner/execute.ts‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@ import {
3434
parseCliOutput,
3535
type CliOutput,
3636
type CliStreamingDelta,
37+
type CliThinkingDelta,
3738
} from "../cli-output.js";
3839
import { classifyFailoverReason } from "../embedded-agent-helpers.js";
3940
import {
@@ -1017,6 +1018,28 @@ export async function executePreparedCliRun(
10171018
},
10181019
});
10191020
};
1021+
// Emit-always: thinking reaches the agent-event bus and session archive
1022+
// like the embedded reasoning stream; /reasoning and /verbose gate only
1023+
// presentation. Text stays raw here to match the thinking-stream contract
1024+
// shared with embedded-agent-subscribe, which archives untransformed
1025+
// reasoning regardless of source.
1026+
const emitCliThinkingDelta = ({
1027+
text,
1028+
delta,
1029+
isReasoningSnapshot,
1030+
}: CliThinkingDelta) => {
1031+
if (text || delta) {
1032+
observedCliActivity = true;
1033+
}
1034+
if (!emitLiveEvents) {
1035+
return;
1036+
}
1037+
emitAgentEvent({
1038+
runId: params.runId,
1039+
stream: "thinking",
1040+
data: { text, delta, ...(isReasoningSnapshot ? { isReasoningSnapshot } : {}) },
1041+
});
1042+
};
10201043
if (shouldUseClaudeLiveSession(context)) {
10211044
if (!hasJsonlOutput) {
10221045
throw new Error("Claude live session requires JSONL streaming parser");
@@ -1037,6 +1060,7 @@ export async function executePreparedCliRun(
10371060
noOutputTimeoutMs,
10381061
getProcessSupervisor: executeDeps.getProcessSupervisor,
10391062
onAssistantDelta: emitCliAssistantDelta,
1063+
onThinkingDelta: emitCliThinkingDelta,
10401064
onToolUseStart: emitCliToolUseStart,
10411065
onToolResult: emitCliToolResult,
10421066
onCommentaryText:
@@ -1064,6 +1088,7 @@ export async function executePreparedCliRun(
10641088
backend,
10651089
providerId: context.backendResolved.id,
10661090
onAssistantDelta: emitCliAssistantDelta,
1091+
onThinkingDelta: emitCliThinkingDelta,
10671092
onToolUseStart: emitCliToolUseStart,
10681093
onToolResult: emitCliToolResult,
10691094
onCommentaryText:

0 commit comments

Comments
 (0)