fix: 检测网关 SSE 空闲超时导致的流中途截断，避免误判为 end_turn

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)
+  })
+})

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<string, unknown> = {
+            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,20 +2512,70 @@ 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', {
           model:
             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

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
+  )
+}