diff --git a/src/services/api/__tests__/streamCompletion.test.ts b/src/services/api/__tests__/streamCompletion.test.ts new file mode 100644 index 0000000000..fa9bd7f88e --- /dev/null +++ b/src/services/api/__tests__/streamCompletion.test.ts @@ -0,0 +1,130 @@ +import { describe, expect, test } from 'bun:test' +import { + hasCompletedToolUse, + isPrematureStreamTruncation, + type CompletionMessageLike, + type StreamTruncationInput, +} from '../streamCompletion' + +function msg(content: unknown): CompletionMessageLike { + return { message: { content } } +} + +describe('hasCompletedToolUse', () => { + test('returns false for an empty message list', () => { + expect(hasCompletedToolUse([])).toBe(false) + }) + + test('returns false when messages contain only text blocks', () => { + expect(hasCompletedToolUse([msg([{ type: 'text', text: 'hello' }])])).toBe( + false, + ) + }) + + test('returns false when a thinking block completed but no tool_use', () => { + expect( + hasCompletedToolUse([ + msg([{ type: 'thinking', thinking: '...' }]), + msg([{ type: 'text', text: 'done' }]), + ]), + ).toBe(false) + }) + + test('returns true when a tool_use block completed', () => { + expect( + hasCompletedToolUse([ + msg([{ type: 'text', text: 'ok' }]), + msg([{ type: 'tool_use', name: 'Write', input: {} }]), + ]), + ).toBe(true) + }) + + test('returns true for a server_tool_use block', () => { + expect( + hasCompletedToolUse([ + msg([{ type: 'server_tool_use', name: 'advisor' }]), + ]), + ).toBe(true) + }) + + test('tolerates missing/non-array content without throwing', () => { + expect(hasCompletedToolUse([{ message: {} }])).toBe(false) + expect(hasCompletedToolUse([{}])).toBe(false) + expect(hasCompletedToolUse([msg('not-an-array')])).toBe(false) + }) +}) + +describe('isPrematureStreamTruncation', () => { + const base: StreamTruncationInput = { + hasPartialMessage: true, + stopReason: null, + startedBlockCount: 2, + completedMessageCount: 1, + hasCompletedToolUse: false, + } + + test('detects the gateway idle-timeout signature (open block, no stop_reason)', () => { + // text block closed (completedMessageCount=1), tool_use opened but never + // closed (startedBlockCount=2), no message_delta (stopReason=null). + expect(isPrematureStreamTruncation(base)).toBe(true) + }) + + test('does not fire when a terminal stop_reason arrived', () => { + expect( + isPrematureStreamTruncation({ ...base, stopReason: 'end_turn' }), + ).toBe(false) + expect( + isPrematureStreamTruncation({ ...base, stopReason: 'tool_use' }), + ).toBe(false) + }) + + test('does not fire when every opened block was closed', () => { + // startedBlockCount === completedMessageCount → no dangling block. + expect( + isPrematureStreamTruncation({ + ...base, + startedBlockCount: 1, + completedMessageCount: 1, + }), + ).toBe(false) + }) + + test('does not fire when no message_start was received', () => { + expect( + isPrematureStreamTruncation({ ...base, hasPartialMessage: false }), + ).toBe(false) + }) + + test('does not fire when a tool_use already completed (inc-4258 guard)', () => { + // Even with an open trailing block and null stop_reason, a completed + // tool_use means the non-streaming fallback could double-execute it. + expect( + isPrematureStreamTruncation({ ...base, hasCompletedToolUse: true }), + ).toBe(false) + }) + + test('does not fire on a legitimate empty response (no blocks at all)', () => { + // Structured-output turn 2: end_turn with zero content blocks. Here + // stopReason would be set, but even with stopReason=null a zero-block + // response has startedBlockCount === completedMessageCount === 0. + expect( + isPrematureStreamTruncation({ + ...base, + startedBlockCount: 0, + completedMessageCount: 0, + }), + ).toBe(false) + }) + + test('fires when multiple blocks completed but a later block was cut', () => { + expect( + isPrematureStreamTruncation({ + hasPartialMessage: true, + stopReason: null, + startedBlockCount: 3, + completedMessageCount: 2, + hasCompletedToolUse: false, + }), + ).toBe(true) + }) +}) diff --git a/src/services/api/claude.ts b/src/services/api/claude.ts index 5c75e22221..e63157bc6b 100644 --- a/src/services/api/claude.ts +++ b/src/services/api/claude.ts @@ -181,6 +181,11 @@ import { headlessProfilerCheckpoint } from 'src/utils/headlessProfiler.js' import { isMcpInstructionsDeltaEnabled } from 'src/utils/mcpInstructionsDelta.js' import { calculateUSDCost } from 'src/utils/modelCost.js' import { endQueryProfile, queryCheckpoint } from 'src/utils/queryProfiler.js' +import { isSseTraceEnabled, traceSseEvent } from 'src/utils/sseTrace.js' +import { + hasCompletedToolUse, + isPrematureStreamTruncation, +} from 'src/services/api/streamCompletion.js' import { modelSupportsAdaptiveThinking, modelSupportsThinking, @@ -2037,6 +2042,40 @@ async function* queryModel( resetStreamIdleTimer() const now = Date.now() + // Raw SSE tracer (opt-in via CLAUDE_CODE_SSE_TRACE_FILE): capture every + // stream event's type — and for content_block_start, the block type / + // tool name — so we can see whether a tool_use block actually arrives + // after a long thinking block, or whether the gateway jumps straight to + // message_delta(stop_reason=end_turn) and drops the tool call. + if (isSseTraceEnabled()) { + const detail: Record = { + type: part.type, + elapsedMs: now - start, + requestId: streamRequestId ?? null, + } + if (part.type === 'content_block_start') { + detail.blockType = part.content_block.type + detail.index = part.index + if ( + part.content_block.type === 'tool_use' || + part.content_block.type === 'server_tool_use' + ) { + detail.toolName = (part.content_block as { name?: string }).name + } + } else if (part.type === 'content_block_delta') { + detail.deltaType = (part.delta as { type?: string }).type + detail.index = part.index + } else if (part.type === 'message_delta') { + detail.stopReason = part.delta.stop_reason + detail.outputTokens = ( + part.usage as { output_tokens?: number } + )?.output_tokens + } else if (part.type === 'content_block_stop') { + detail.index = part.index + } + traceSseEvent('stream_event', detail) + } + // Detect and log streaming stalls (only after first event to avoid counting TTFB) if (lastEventTime !== null) { const timeSinceLastEvent = now - lastEventTime @@ -2414,6 +2453,23 @@ async function* queryModel( // Clear the idle timeout watchdog now that the stream loop has exited clearStreamIdleTimers() + // Raw SSE tracer: stream-end summary. Records the final assembled block + // layout and stop_reason so we can confirm — for a premature end_turn — + // whether any tool_use block was ever assembled. If blockTypes shows only + // [thinking, text] with stopReason=end_turn right after the model said it + // would act, the gateway dropped the tool call. + if (isSseTraceEnabled()) { + traceSseEvent('stream_end', { + elapsedMs: Date.now() - start, + requestId: streamRequestId ?? null, + stopReason: stopReason ?? null, + blockTypes: contentBlocks + .filter(Boolean) + .map(b => (b as { type?: string }).type ?? 'unknown'), + blockCount: contentBlocks.filter(Boolean).length, + }) + } + // If the stream was aborted by our idle timeout watchdog, fall back to // non-streaming retry rather than treating it as a completed stream. if (streamIdleAborted) { @@ -2456,11 +2512,52 @@ async function* queryModel( // structured output (--json-schema), the model calls a StructuredOutput tool // on turn 1, then on turn 2 responds with end_turn and no content blocks. // That's a legitimate empty response, not an incomplete stream. - if (!partialMessage || (newMessages.length === 0 && !stopReason)) { + // + // Mode 3 (gateway SSE idle-timeout truncation): the stream completed SOME + // content blocks, but was then cut mid-turn — a content block was opened + // (content_block_start) and never closed (no matching content_block_stop), + // and no message_delta ever set a terminal stop_reason (stopReason===null). + // This is the gateway signature: during a long extended-thinking + // pause or while streaming a large tool_use input_json_delta (e.g. a big + // Write payload), the SSE goes idle past the gateway's proxy_read_timeout + // (~180s) and the connection is closed gracefully — the async iterator just + // ends, no exception. Without this branch the partial text is misreported + // as a completed end_turn and the turn silently stops half-done. + // + // The unclosed-block check (startedBlocks > newMessages.length) is the key + // discriminator: a provider that merely omits stop_reason still emits + // content_block_stop for every block it opened, so it won't false-positive + // here — only a genuinely truncated stream leaves a block open. We also + // require NO completed tool_use, because a tool_use that already closed may + // have started executing via the streaming tool executor; re-issuing it via + // the non-streaming fallback would double-execute it (inc-4258). When a + // tool_use already completed we leave the existing behavior untouched. + // + // Limitation: the non-streaming fallback re-buffers the whole response and + // is itself subject to the same gateway idle timeout for very long + // generations. The guaranteed win here is that truncation stops being + // silently reported as end_turn — it either recovers via non-streaming or + // surfaces a real error the caller (e.g. the ACP retry layer) can act on. + const startedBlockCount = contentBlocks.filter(Boolean).length + const streamHadCompletedToolUse = hasCompletedToolUse(newMessages) + const streamWasTruncatedMidTurn = isPrematureStreamTruncation({ + hasPartialMessage: Boolean(partialMessage), + stopReason, + startedBlockCount, + completedMessageCount: newMessages.length, + hasCompletedToolUse: streamHadCompletedToolUse, + }) + if ( + !partialMessage || + (newMessages.length === 0 && !stopReason) || + streamWasTruncatedMidTurn + ) { logForDebugging( !partialMessage ? 'Stream completed without receiving message_start event - triggering non-streaming fallback' - : 'Stream completed with message_start but no content blocks completed - triggering non-streaming fallback', + : streamWasTruncatedMidTurn + ? 'Stream truncated mid-turn (open content block, stop_reason=null) - triggering non-streaming fallback' + : 'Stream completed with message_start but no content blocks completed - triggering non-streaming fallback', { level: 'error' }, ) logEvent('tengu_stream_no_events', { @@ -2468,8 +2565,17 @@ async function* queryModel( options.model as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS, request_id: (streamRequestId ?? 'unknown') as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS, + truncation_kind: (streamWasTruncatedMidTurn + ? 'mid_block_stop_reason_null' + : !partialMessage + ? 'no_message_start' + : 'no_block_completed') as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS, }) - throw new Error('Stream ended without receiving any events') + throw new Error( + streamWasTruncatedMidTurn + ? 'Stream truncated mid-turn before completion' + : 'Stream ended without receiving any events', + ) } // Log summary if any stalls occurred during streaming diff --git a/src/services/api/streamCompletion.ts b/src/services/api/streamCompletion.ts new file mode 100644 index 0000000000..aca20f9c22 --- /dev/null +++ b/src/services/api/streamCompletion.ts @@ -0,0 +1,91 @@ +/** + * Stream-completion classification helpers for the streaming API client. + * + * Extracted as pure functions so the tricky "was this stream truncated?" + * decision can be unit-tested without standing up a full network stream. + * + * Background: on gateway-fronted providers a long + * extended-thinking pause or a large tool_use `input_json_delta` (e.g. a big + * Write payload) can leave the SSE idle past the gateway's proxy_read_timeout + * (~180s). The gateway then closes the connection gracefully — the async + * iterator just ends with no exception, no `content_block_stop` for the open + * block and no `message_delta` carrying a terminal `stop_reason`. Without + * explicit detection the partial output is misreported as a completed + * `end_turn` and the turn silently stops half-done. + */ + +/** + * Minimal shape needed to decide whether a completed assistant message already + * contains a tool_use block. Kept structural so callers can pass their richer + * AssistantMessage type without conversion. + */ +export interface CompletionMessageLike { + message?: { + content?: unknown + } +} + +/** + * True if any completed message already carries a (server_)tool_use block. + * + * Used to guard against inc-4258: a tool_use that already closed may have begun + * executing via the streaming tool executor, so re-issuing the request through + * the non-streaming fallback would double-execute it. When a tool_use has + * completed we must NOT treat a subsequent truncation as recoverable here. + */ +export function hasCompletedToolUse( + messages: readonly CompletionMessageLike[], +): boolean { + return messages.some(m => { + const content = m.message?.content + return ( + Array.isArray(content) && + content.some(b => { + const t = (b as { type?: string }).type + return t === 'tool_use' || t === 'server_tool_use' + }) + ) + }) +} + +/** Inputs describing the final state of a streamed response. */ +export interface StreamTruncationInput { + /** Whether a `message_start` event was ever received (partialMessage set). */ + hasPartialMessage: boolean + /** Terminal stop_reason from `message_delta`, or null if none arrived. */ + stopReason: string | null + /** Count of content blocks that were opened via `content_block_start`. */ + startedBlockCount: number + /** Count of assistant messages emitted (one per `content_block_stop`). */ + completedMessageCount: number + /** Whether any completed message already contains a tool_use block. */ + hasCompletedToolUse: boolean +} + +/** + * Detect a stream that was cut mid-turn by an intermediary (gateway idle + * timeout), as opposed to a legitimately completed or legitimately empty + * response. + * + * Signature of the truncation we recover from: + * - a `message_start` was received (`hasPartialMessage`), AND + * - no terminal `stop_reason` ever arrived (`stopReason === null`), AND + * - at least one content block was opened but never closed + * (`startedBlockCount > completedMessageCount`), AND + * - no tool_use block has already completed (avoids inc-4258 double execution). + * + * The unclosed-block check is the key discriminator: a well-behaved provider + * that merely omits `stop_reason` still emits `content_block_stop` for every + * block it opened, so it won't false-positive here — only a genuinely truncated + * stream leaves a block open. + */ +export function isPrematureStreamTruncation( + input: StreamTruncationInput, +): boolean { + return ( + input.hasPartialMessage && + input.stopReason === null && + input.startedBlockCount > input.completedMessageCount && + !input.hasCompletedToolUse + ) +} diff --git a/src/utils/sseTrace.ts b/src/utils/sseTrace.ts new file mode 100644 index 0000000000..a6580c2194 --- /dev/null +++ b/src/utils/sseTrace.ts @@ -0,0 +1,75 @@ +/** + * Raw SSE stream event tracer for diagnosing premature `end_turn` / dropped + * tool_use events on gateway-fronted providers. + * + * Background: on a gateway-fronted Anthropic endpoint, Opus extended-thinking turns + * sometimes end with `stop_reason=end_turn` right after the model announces an + * action ("Now I'll write the file...") WITHOUT emitting the tool_use content + * block. We need to see, at the raw stream level, exactly which SSE events the + * gateway forwards after a long thinking block — specifically whether a + * `content_block_start(tool_use)` ever arrives, or whether the stream jumps + * straight to `message_delta(stop_reason=end_turn)`. + * + * This tracer is fully opt-in via env and self-contained (no PII filtering, + * because we intentionally want to capture block types / stop reasons). It is + * NOT wired into the normal debug pipeline so it works even when the ACP child + * process runs without `--debug`. + * + * Enable by setting CLAUDE_CODE_SSE_TRACE_FILE=/absolute/path/to/sse-trace.log + * (Optionally CLAUDE_CODE_SSE_TRACE=1 alone logs to a default temp path.) + */ +import { appendFileSync, mkdirSync } from 'node:fs' +import { dirname, join } from 'node:path' +import { tmpdir } from 'node:os' + +let cachedPath: string | null | undefined + +function resolveTracePath(): string | null { + if (cachedPath !== undefined) { + return cachedPath + } + const explicit = process.env.CLAUDE_CODE_SSE_TRACE_FILE?.trim() + if (explicit) { + cachedPath = explicit + return cachedPath + } + const enabled = process.env.CLAUDE_CODE_SSE_TRACE?.trim() + if (enabled && !['0', 'false', 'no', 'off'].includes(enabled.toLowerCase())) { + cachedPath = join(tmpdir(), 'claude-code-sse-trace.log') + return cachedPath + } + cachedPath = null + return cachedPath +} + +export function isSseTraceEnabled(): boolean { + return resolveTracePath() !== null +} + +/** + * Append a single raw SSE trace line. No-op unless enabled via env. + * + * @param event Short event tag, e.g. 'stream_event', 'stream_end'. + * @param data Arbitrary structured data (block types, stop_reason, elapsed). + */ +export function traceSseEvent( + event: string, + data: Record, +): void { + const path = resolveTracePath() + if (!path) { + return + } + const line = + JSON.stringify({ ts: new Date().toISOString(), event, ...data }) + '\n' + try { + appendFileSync(path, line) + } catch { + try { + mkdirSync(dirname(path), { recursive: true }) + appendFileSync(path, line) + } catch { + // Best-effort: never crash the stream loop on a trace failure. + } + } +}