mirror of
https://github.com/NanmiCoder/cc-haha
synced 2026-07-16 13:03:31 +08:00
fix(trace): record aborted API calls instead of leaving them pending (#766)
When an upstream request was aborted mid-stream (SDK client timeout, stream idle watchdog, non-streaming fallback timeout, or user cancellation), the trace fetch hook waited on a clone of the response body that could hang forever, so the call never left "pending" in the trace panel — exactly the silent stall that misled the #766 report. - captureResponseTraceSnapshot reads the body with abort awareness: reader.cancel() on abort keeps the partial body, with a 2s grace backstop for runtimes where cancel cannot wake a hung read. - The fetch hook now records an error-state call on abort with the abort reason (e.g. the watchdog's stream idle timeout), duration, partial response body, and an api_call_aborted event; non-abort capture failures also record an error instead of inferring ok, and pre-response fetch rejections carry an aborted flag. - The trace detail panel shows an "Aborted" badge plus guidance for aborted calls, and labels the new api_call_aborted phase in all locales. Tested: bun test src/server/__tests__/trace-capture.test.ts Tested: bun run check:server Tested: cd desktop && bun run test -- --run && bun run lint
This commit is contained in:
程序员阿江(Relakkes)
parent
67660ab4e5
commit
ea82c6ec70
@ -160,6 +160,7 @@ export function traceEventPhaseLabel(phase: string, t: TraceTranslator): string
|
||||
case 'api_call_started': return t('trace.event.apiCallStarted')
|
||||
case 'api_call_completed': return t('trace.event.apiCallCompleted')
|
||||
case 'api_call_failed': return t('trace.event.apiCallFailed')
|
||||
case 'api_call_aborted': return t('trace.event.apiCallAborted')
|
||||
case 'response_capture_failed': return t('trace.event.responseCaptureFailed')
|
||||
case 'upstream_fetch_started': return t('trace.event.upstreamFetchStarted')
|
||||
case 'upstream_fetch_completed': return t('trace.event.upstreamFetchCompleted')
|
||||
|
||||
@ -142,6 +142,12 @@ export function LlmCallDetail({ sessionId, span }: { sessionId: string; span: Tr
|
||||
)
|
||||
}
|
||||
|
||||
export function isAbortedTraceCall(call: TraceCallRecord): boolean {
|
||||
if (call.metadata?.aborted === true) return true
|
||||
const name = call.error?.name
|
||||
return name === 'AbortError' || name === 'TimeoutError'
|
||||
}
|
||||
|
||||
function ResponseContent({
|
||||
call,
|
||||
pending,
|
||||
@ -155,10 +161,29 @@ function ResponseContent({
|
||||
}) {
|
||||
const t = useTranslation()
|
||||
if (call.error) {
|
||||
const aborted = isAbortedTraceCall(call)
|
||||
return (
|
||||
<div className="rounded-[var(--radius-md)] border border-[var(--color-error)]/25 bg-[var(--color-error-container)]/40 px-3 py-2">
|
||||
<div className="text-xs font-semibold text-[var(--color-error)]">{call.error.name}</div>
|
||||
<div
|
||||
className="rounded-[var(--radius-md)] border border-[var(--color-error)]/25 bg-[var(--color-error-container)]/40 px-3 py-2"
|
||||
data-testid="trace-call-error"
|
||||
>
|
||||
<div className="flex min-w-0 items-center gap-2">
|
||||
<div className="text-xs font-semibold text-[var(--color-error)]">{call.error.name}</div>
|
||||
{aborted ? (
|
||||
<span
|
||||
className="inline-flex shrink-0 items-center rounded-[var(--radius-sm)] bg-[var(--color-error)]/10 px-1.5 py-0.5 text-[10px] font-semibold text-[var(--color-error)]"
|
||||
data-testid="trace-call-aborted-badge"
|
||||
>
|
||||
{t('trace.status.aborted')}
|
||||
</span>
|
||||
) : null}
|
||||
</div>
|
||||
<div className="mt-1 text-xs leading-5 text-[var(--color-text-secondary)]">{call.error.message}</div>
|
||||
{aborted ? (
|
||||
<div className="mt-1 text-[11px] leading-4 text-[var(--color-text-tertiary)]">
|
||||
{t('trace.detail.aborted')}
|
||||
</div>
|
||||
) : null}
|
||||
{call.error.stack ? (
|
||||
<details className="mt-1.5">
|
||||
<summary className="cursor-pointer text-[10px] uppercase tracking-[0.08em] text-[var(--color-text-tertiary)]">
|
||||
|
||||
@ -1774,6 +1774,7 @@ export const en = {
|
||||
'trace.event.apiCallStarted': 'API call started',
|
||||
'trace.event.apiCallCompleted': 'API call completed',
|
||||
'trace.event.apiCallFailed': 'API call failed',
|
||||
'trace.event.apiCallAborted': 'API call aborted',
|
||||
'trace.event.responseCaptureFailed': 'Response capture failed',
|
||||
'trace.event.upstreamFetchStarted': 'Upstream fetch started',
|
||||
'trace.event.upstreamFetchCompleted': 'Upstream fetch completed',
|
||||
@ -1792,6 +1793,8 @@ export const en = {
|
||||
'trace.section.metadata': 'Metadata',
|
||||
'trace.detail.stopReason': 'Stop reason',
|
||||
'trace.detail.streaming': 'Waiting for the model response...',
|
||||
'trace.detail.aborted': 'The request was aborted before the response completed (timeout or cancellation).',
|
||||
'trace.status.aborted': 'Aborted',
|
||||
'trace.detail.legacyTruncated': 'Legacy truncated record; the semantic view is unavailable. See Raw below.',
|
||||
'trace.detail.fetchFailed': 'Failed to load the full record; showing the truncated preview.',
|
||||
'trace.detail.earlierMessages': 'Show {count} earlier messages',
|
||||
|
||||
@ -1776,6 +1776,7 @@ export const jp: Record<TranslationKey, string> = {
|
||||
'trace.event.apiCallStarted': 'API 呼び出し開始',
|
||||
'trace.event.apiCallCompleted': 'API 呼び出し完了',
|
||||
'trace.event.apiCallFailed': 'API 呼び出し失敗',
|
||||
'trace.event.apiCallAborted': 'API 呼び出し中断',
|
||||
'trace.event.responseCaptureFailed': 'レスポンスキャプチャ失敗',
|
||||
'trace.event.upstreamFetchStarted': 'アップストリーム要求開始',
|
||||
'trace.event.upstreamFetchCompleted': 'アップストリーム要求完了',
|
||||
@ -1794,6 +1795,8 @@ export const jp: Record<TranslationKey, string> = {
|
||||
'trace.section.metadata': 'メタデータ',
|
||||
'trace.detail.stopReason': '停止理由',
|
||||
'trace.detail.streaming': 'モデルの応答を待機中...',
|
||||
'trace.detail.aborted': '応答が完了する前にリクエストが中断されました（タイムアウトまたはキャンセル）。',
|
||||
'trace.status.aborted': '中断',
|
||||
'trace.detail.legacyTruncated': 'このレコードは旧形式の切り詰められたデータのため、セマンティック表示はできません。下の生データを確認してください。',
|
||||
'trace.detail.fetchFailed': '完全なレコードの読み込みに失敗したため、切り詰められたプレビューを表示しています。',
|
||||
'trace.detail.earlierMessages': '以前の {count} 件のメッセージを表示',
|
||||
|
||||
@ -1776,6 +1776,7 @@ export const kr: Record<TranslationKey, string> = {
|
||||
'trace.event.apiCallStarted': 'API 호출 시작',
|
||||
'trace.event.apiCallCompleted': 'API 호출 완료',
|
||||
'trace.event.apiCallFailed': 'API 호출 실패',
|
||||
'trace.event.apiCallAborted': 'API 호출 중단',
|
||||
'trace.event.responseCaptureFailed': '응답 캡처 실패',
|
||||
'trace.event.upstreamFetchStarted': '업스트림 요청 시작',
|
||||
'trace.event.upstreamFetchCompleted': '업스트림 요청 완료',
|
||||
@ -1794,6 +1795,8 @@ export const kr: Record<TranslationKey, string> = {
|
||||
'trace.section.metadata': '메타데이터',
|
||||
'trace.detail.stopReason': '중지 사유',
|
||||
'trace.detail.streaming': '모델 응답을 기다리는 중...',
|
||||
'trace.detail.aborted': '응답이 완료되기 전에 요청이 중단되었습니다(시간 초과 또는 취소).',
|
||||
'trace.status.aborted': '중단됨',
|
||||
'trace.detail.legacyTruncated': '이 레코드는 구형식의 잘린 데이터여서 시맨틱 보기를 사용할 수 없습니다. 아래 원본 데이터를 확인하세요.',
|
||||
'trace.detail.fetchFailed': '전체 레코드를 불러오지 못해 잘린 미리보기를 표시합니다.',
|
||||
'trace.detail.earlierMessages': '이전 메시지 {count}개 보기',
|
||||
|
||||
@ -1776,6 +1776,7 @@ export const zh: Record<TranslationKey, string> = {
|
||||
'trace.event.apiCallStarted': 'API 呼叫開始',
|
||||
'trace.event.apiCallCompleted': 'API 呼叫完成',
|
||||
'trace.event.apiCallFailed': 'API 呼叫失敗',
|
||||
'trace.event.apiCallAborted': 'API 呼叫中斷',
|
||||
'trace.event.responseCaptureFailed': '回應捕獲失敗',
|
||||
'trace.event.upstreamFetchStarted': '上游請求開始',
|
||||
'trace.event.upstreamFetchCompleted': '上游請求完成',
|
||||
@ -1794,6 +1795,8 @@ export const zh: Record<TranslationKey, string> = {
|
||||
'trace.section.metadata': '中繼資料',
|
||||
'trace.detail.stopReason': '停止原因',
|
||||
'trace.detail.streaming': '等待模型回應...',
|
||||
'trace.detail.aborted': '請求在回應完成前被中斷（逾時或取消）。',
|
||||
'trace.status.aborted': '已中斷',
|
||||
'trace.detail.legacyTruncated': '此記錄為舊格式截斷資料，無法解析語意檢視，請查看下方原始資料。',
|
||||
'trace.detail.fetchFailed': '完整記錄載入失敗，以下顯示截斷預覽。',
|
||||
'trace.detail.earlierMessages': '展開更早的 {count} 則訊息',
|
||||
|
||||
@ -1776,6 +1776,7 @@ export const zh: Record<TranslationKey, string> = {
|
||||
'trace.event.apiCallStarted': 'API 调用开始',
|
||||
'trace.event.apiCallCompleted': 'API 调用完成',
|
||||
'trace.event.apiCallFailed': 'API 调用失败',
|
||||
'trace.event.apiCallAborted': 'API 调用中断',
|
||||
'trace.event.responseCaptureFailed': '响应捕获失败',
|
||||
'trace.event.upstreamFetchStarted': '上游请求开始',
|
||||
'trace.event.upstreamFetchCompleted': '上游请求完成',
|
||||
@ -1794,6 +1795,8 @@ export const zh: Record<TranslationKey, string> = {
|
||||
'trace.section.metadata': '元数据',
|
||||
'trace.detail.stopReason': '停止原因',
|
||||
'trace.detail.streaming': '等待模型响应...',
|
||||
'trace.detail.aborted': '请求在响应完成前被中断（超时或取消）。',
|
||||
'trace.status.aborted': '已中断',
|
||||
'trace.detail.legacyTruncated': '该记录为旧格式截断数据，无法解析语义视图，请查看下方原始数据。',
|
||||
'trace.detail.fetchFailed': '完整记录加载失败，以下展示截断预览。',
|
||||
'trace.detail.earlierMessages': '展开更早的 {count} 条消息',
|
||||
|
||||
@ -240,4 +240,65 @@ describe('traceViewModel', () => {
|
||||
focusSpanId: 'event:event-failed',
|
||||
})
|
||||
})
|
||||
|
||||
it('surfaces aborted calls as model errors instead of pending', () => {
|
||||
const viewModel = buildTraceViewModel({
|
||||
sessionId: 'session-aborted',
|
||||
session: null,
|
||||
summary: {
|
||||
apiCalls: 1,
|
||||
failedCalls: 1,
|
||||
totalDurationMs: 240_000,
|
||||
totalInputTokens: 0,
|
||||
totalOutputTokens: 0,
|
||||
models: [{ model: 'gpt-5.5', calls: 1 }],
|
||||
updatedAt: '2026-06-09T10:04:01.000Z',
|
||||
},
|
||||
calls: [{
|
||||
id: 'call-aborted',
|
||||
sessionId: 'session-aborted',
|
||||
source: 'anthropic',
|
||||
model: 'gpt-5.5',
|
||||
status: 'error',
|
||||
startedAt: '2026-06-09T10:00:01.000Z',
|
||||
completedAt: '2026-06-09T10:04:01.000Z',
|
||||
durationMs: 240_000,
|
||||
metadata: { phase: 'api_call_aborted', aborted: true },
|
||||
request: {
|
||||
method: 'POST',
|
||||
url: 'https://sub2api.example/v1/messages',
|
||||
headers: {},
|
||||
body: { contentType: 'json', bytes: 20, sha256: 'a', preview: '{"model":"gpt-5.5"}', truncated: false },
|
||||
},
|
||||
response: {
|
||||
status: 200,
|
||||
headers: {},
|
||||
body: { contentType: 'text', bytes: 30, sha256: 'b', preview: 'data: {"type":"message_start"}', truncated: true },
|
||||
},
|
||||
error: {
|
||||
name: 'AbortError',
|
||||
message: 'Stream idle timeout: no chunks received for 240s',
|
||||
},
|
||||
}],
|
||||
events: [{
|
||||
id: 'event-aborted',
|
||||
sessionId: 'session-aborted',
|
||||
callId: 'call-aborted',
|
||||
timestamp: '2026-06-09T10:04:01.000Z',
|
||||
phase: 'api_call_aborted',
|
||||
severity: 'error',
|
||||
message: 'Stream idle timeout: no chunks received for 240s',
|
||||
}],
|
||||
}, [])
|
||||
|
||||
const llmSpan = viewModel.spansById.get('llm:call-aborted')
|
||||
expect(llmSpan).toMatchObject({ kind: 'llm', status: 'error', durationMs: 240_000 })
|
||||
expect(llmSpan?.completedAt).toBe('2026-06-09T10:04:01.000Z')
|
||||
expect(viewModel.diagnosis).toMatchObject({
|
||||
status: 'blocked',
|
||||
reason: 'model_error',
|
||||
focusSpanId: 'llm:call-aborted',
|
||||
pendingModelCalls: 0,
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@ -312,6 +312,49 @@ describe('TraceSession', () => {
|
||||
expect(detail.getByText('1.2k → 847')).toBeInTheDocument()
|
||||
})
|
||||
|
||||
it('shows the aborted badge and abort guidance for an aborted call', async () => {
|
||||
const abortedCall = makeCall({
|
||||
id: 'call-aborted',
|
||||
status: 'error',
|
||||
completedAt: '2026-06-09T10:04:01.000Z',
|
||||
durationMs: 240_000,
|
||||
metadata: { phase: 'api_call_aborted', aborted: true },
|
||||
error: { name: 'AbortError', message: 'Stream idle timeout: no chunks received for 240s' },
|
||||
response: {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'text/event-stream' },
|
||||
body: {
|
||||
contentType: 'text',
|
||||
bytes: 30,
|
||||
sha256: 'c'.repeat(64),
|
||||
preview: 'data: {"type":"message_start"}',
|
||||
truncated: true,
|
||||
},
|
||||
},
|
||||
})
|
||||
vi.mocked(sessionsApi.getTrace).mockResolvedValue({
|
||||
...baseTrace,
|
||||
summary: { ...baseTrace.summary, failedCalls: 1 },
|
||||
calls: [abortedCall],
|
||||
})
|
||||
vi.mocked(sessionsApi.getTraceCall).mockResolvedValue({ call: abortedCall })
|
||||
await renderReady()
|
||||
|
||||
fireEvent.click(within(screen.getByTestId('trace-tree')).getByText('claude-sonnet-4-5'))
|
||||
|
||||
const detail = within(screen.getByTestId('trace-detail'))
|
||||
expect(await detail.findByTestId('trace-call-error')).toBeInTheDocument()
|
||||
expect(detail.getByTestId('trace-call-aborted-badge')).toHaveTextContent('Aborted')
|
||||
expect(detail.getByText('AbortError')).toBeInTheDocument()
|
||||
expect(detail.getByText('Stream idle timeout: no chunks received for 240s')).toBeInTheDocument()
|
||||
expect(
|
||||
detail.getByText('The request was aborted before the response completed (timeout or cancellation).'),
|
||||
).toBeInTheDocument()
|
||||
// The header pill shows the terminal error state, not pending.
|
||||
expect(detail.getByText('error')).toBeInTheDocument()
|
||||
expect(detail.queryByText('pending')).not.toBeInTheDocument()
|
||||
})
|
||||
|
||||
it('falls back to Raw with a legacy notice when the body cannot be parsed', async () => {
|
||||
const legacyCall = makeCall({
|
||||
request: {
|
||||
|
||||
@ -5,6 +5,7 @@ import * as os from 'os'
|
||||
import * as path from 'path'
|
||||
import { handleApiRequest } from '../router.js'
|
||||
import {
|
||||
captureResponseTraceSnapshot,
|
||||
clearTraceCaptureStateForTests,
|
||||
createTraceCallId,
|
||||
createTraceBodySnapshot,
|
||||
@ -638,6 +639,300 @@ describe('trace capture service', () => {
|
||||
else process.env.CC_HAHA_TRACE_API_CALLS = originalTraceEnv
|
||||
}
|
||||
})
|
||||
|
||||
test('records an aborted error call when the request is aborted mid-stream', async () => {
|
||||
const originalFetch = globalThis.fetch
|
||||
const originalTraceEnv = process.env.CC_HAHA_TRACE_API_CALLS
|
||||
process.env.CC_HAHA_TRACE_API_CALLS = '1'
|
||||
try {
|
||||
// A stream that sends one chunk then goes silent forever, like the
|
||||
// wedged upstream in #766. The mock ignores the abort signal, so the
|
||||
// trace capture must end the read itself.
|
||||
globalThis.fetch = (async () => {
|
||||
const stream = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
controller.enqueue(new TextEncoder().encode('data: {"type":"message_start"}\n\n'))
|
||||
},
|
||||
})
|
||||
return new Response(stream, {
|
||||
status: 200,
|
||||
headers: { 'Content-Type': 'text/event-stream' },
|
||||
})
|
||||
}) as typeof fetch
|
||||
|
||||
const abortController = new AbortController()
|
||||
const traceFetch = createDumpPromptsFetch('agent-direct-abort', {
|
||||
traceSessionId: 'session-direct-abort',
|
||||
querySource: 'test_query',
|
||||
})
|
||||
await traceFetch('https://sub2api.example.test/v1/messages', {
|
||||
method: 'POST',
|
||||
body: JSON.stringify({ model: 'gpt-5.5', messages: [{ role: 'user', content: 'abort me' }] }),
|
||||
signal: abortController.signal,
|
||||
})
|
||||
|
||||
// Let the capture loop start reading, then abort like the stream idle
|
||||
// watchdog (stream.controller.abort()) or SDK client timeout would.
|
||||
await new Promise((resolve) => setTimeout(resolve, 20))
|
||||
abortController.abort(new Error('Stream idle timeout: no chunks received for 240s'))
|
||||
|
||||
const trace = await waitForTrace(
|
||||
'session-direct-abort',
|
||||
(snapshot) => snapshot.calls[0]?.status === 'error'
|
||||
&& snapshot.events.some((event) => event.phase === 'api_call_aborted'),
|
||||
)
|
||||
expect(trace.summary.apiCalls).toBe(1)
|
||||
expect(trace.summary.failedCalls).toBe(1)
|
||||
expect(trace.calls[0]).toMatchObject({
|
||||
source: 'anthropic',
|
||||
model: 'gpt-5.5',
|
||||
status: 'error',
|
||||
metadata: { phase: 'api_call_aborted', aborted: true },
|
||||
})
|
||||
expect(trace.calls[0].error?.message).toContain('Stream idle timeout')
|
||||
expect(typeof trace.calls[0].durationMs).toBe('number')
|
||||
expect(trace.calls[0].response?.status).toBe(200)
|
||||
expect(trace.calls[0].response?.body.preview).toContain('message_start')
|
||||
expect(trace.calls[0].response?.body.truncated).toBe(true)
|
||||
expect(trace.events.map((event) => event.phase)).toEqual(['api_call_started', 'api_call_aborted'])
|
||||
expect(trace.events.at(-1)?.severity).toBe('error')
|
||||
} finally {
|
||||
globalThis.fetch = originalFetch
|
||||
if (originalTraceEnv === undefined) delete process.env.CC_HAHA_TRACE_API_CALLS
|
||||
else process.env.CC_HAHA_TRACE_API_CALLS = originalTraceEnv
|
||||
}
|
||||
})
|
||||
|
||||
test('synthesizes an AbortError when the abort signal carries no reason', async () => {
|
||||
const originalFetch = globalThis.fetch
|
||||
const originalTraceEnv = process.env.CC_HAHA_TRACE_API_CALLS
|
||||
process.env.CC_HAHA_TRACE_API_CALLS = '1'
|
||||
try {
|
||||
globalThis.fetch = (async () => {
|
||||
const stream = new ReadableStream<Uint8Array>({
|
||||
start() {
|
||||
// No chunks at all: headers arrived, body never produces bytes.
|
||||
},
|
||||
})
|
||||
return new Response(stream, {
|
||||
status: 200,
|
||||
headers: { 'Content-Type': 'text/event-stream' },
|
||||
})
|
||||
}) as typeof fetch
|
||||
|
||||
const abortController = new AbortController()
|
||||
const traceFetch = createDumpPromptsFetch('agent-direct-abort-bare', {
|
||||
traceSessionId: 'session-direct-abort-bare',
|
||||
querySource: 'test_query',
|
||||
})
|
||||
await traceFetch('https://sub2api.example.test/v1/messages', {
|
||||
method: 'POST',
|
||||
body: JSON.stringify({ model: 'gpt-5.5', messages: [{ role: 'user', content: 'abort bare' }] }),
|
||||
signal: abortController.signal,
|
||||
})
|
||||
await new Promise((resolve) => setTimeout(resolve, 20))
|
||||
abortController.abort()
|
||||
|
||||
const trace = await waitForTrace(
|
||||
'session-direct-abort-bare',
|
||||
(snapshot) => snapshot.calls[0]?.status === 'error',
|
||||
)
|
||||
expect(trace.calls[0].status).toBe('error')
|
||||
expect(trace.calls[0].error?.name).toBe('AbortError')
|
||||
expect(trace.calls[0].metadata).toMatchObject({ phase: 'api_call_aborted', aborted: true })
|
||||
} finally {
|
||||
globalThis.fetch = originalFetch
|
||||
if (originalTraceEnv === undefined) delete process.env.CC_HAHA_TRACE_API_CALLS
|
||||
else process.env.CC_HAHA_TRACE_API_CALLS = originalTraceEnv
|
||||
}
|
||||
})
|
||||
|
||||
test('keeps a completed call ok when the signal aborts after the response finished', async () => {
|
||||
const originalFetch = globalThis.fetch
|
||||
const originalTraceEnv = process.env.CC_HAHA_TRACE_API_CALLS
|
||||
process.env.CC_HAHA_TRACE_API_CALLS = '1'
|
||||
try {
|
||||
globalThis.fetch = (async () => new Response(
|
||||
JSON.stringify({ id: 'msg-late-abort', content: [{ type: 'text', text: 'ok' }] }),
|
||||
{
|
||||
status: 200,
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
},
|
||||
)) as typeof fetch
|
||||
|
||||
const abortController = new AbortController()
|
||||
const traceFetch = createDumpPromptsFetch('agent-late-abort', {
|
||||
traceSessionId: 'session-late-abort',
|
||||
querySource: 'test_query',
|
||||
})
|
||||
await traceFetch('https://sub2api.example.test/v1/messages', {
|
||||
method: 'POST',
|
||||
body: JSON.stringify({ model: 'gpt-5.5', messages: [{ role: 'user', content: 'late abort' }] }),
|
||||
signal: abortController.signal,
|
||||
})
|
||||
|
||||
const completed = await waitForTrace(
|
||||
'session-late-abort',
|
||||
(snapshot) => snapshot.events.some((event) => event.phase === 'api_call_completed'),
|
||||
)
|
||||
expect(completed.calls[0].status).not.toBe('error')
|
||||
|
||||
// Aborting after completion (e.g. the user cancels the next tool step)
|
||||
// must not rewrite the finished call into an error.
|
||||
abortController.abort()
|
||||
await new Promise((resolve) => setTimeout(resolve, 50))
|
||||
const trace = await traceCaptureService.getSessionTrace('session-late-abort')
|
||||
expect(trace.calls).toHaveLength(1)
|
||||
expect(trace.calls[0].status).not.toBe('error')
|
||||
expect(trace.calls[0].error).toBeUndefined()
|
||||
expect(trace.summary.failedCalls).toBe(0)
|
||||
expect(trace.events.map((event) => event.phase)).toEqual(['api_call_started', 'api_call_completed'])
|
||||
} finally {
|
||||
globalThis.fetch = originalFetch
|
||||
if (originalTraceEnv === undefined) delete process.env.CC_HAHA_TRACE_API_CALLS
|
||||
else process.env.CC_HAHA_TRACE_API_CALLS = originalTraceEnv
|
||||
}
|
||||
})
|
||||
|
||||
test('marks fetch rejections from an aborted signal with abort metadata', async () => {
|
||||
const originalFetch = globalThis.fetch
|
||||
const originalTraceEnv = process.env.CC_HAHA_TRACE_API_CALLS
|
||||
process.env.CC_HAHA_TRACE_API_CALLS = '1'
|
||||
try {
|
||||
const abortController = new AbortController()
|
||||
globalThis.fetch = (async () => {
|
||||
// Mirror undici: reject with an AbortError once the signal aborts
|
||||
// before headers arrive (SDK client timeout during prefill).
|
||||
abortController.abort()
|
||||
const error = new Error('This operation was aborted')
|
||||
error.name = 'AbortError'
|
||||
throw error
|
||||
}) as typeof fetch
|
||||
|
||||
const traceFetch = createDumpPromptsFetch('agent-fetch-abort', {
|
||||
traceSessionId: 'session-fetch-abort',
|
||||
querySource: 'test_query',
|
||||
})
|
||||
await expect(traceFetch('https://sub2api.example.test/v1/messages', {
|
||||
method: 'POST',
|
||||
body: JSON.stringify({ model: 'gpt-5.5', messages: [{ role: 'user', content: 'pre-headers abort' }] }),
|
||||
signal: abortController.signal,
|
||||
})).rejects.toThrow('This operation was aborted')
|
||||
|
||||
const trace = await waitForTrace(
|
||||
'session-fetch-abort',
|
||||
(snapshot) => snapshot.calls[0]?.status === 'error'
|
||||
&& snapshot.events.some((event) => event.phase === 'api_call_failed'),
|
||||
)
|
||||
expect(trace.calls[0]).toMatchObject({
|
||||
status: 'error',
|
||||
error: { name: 'AbortError' },
|
||||
metadata: { phase: 'api_call_failed', aborted: true },
|
||||
})
|
||||
expect(trace.events.map((event) => event.phase)).toEqual(['api_call_started', 'api_call_failed'])
|
||||
} finally {
|
||||
globalThis.fetch = originalFetch
|
||||
if (originalTraceEnv === undefined) delete process.env.CC_HAHA_TRACE_API_CALLS
|
||||
else process.env.CC_HAHA_TRACE_API_CALLS = originalTraceEnv
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
describe('captureResponseTraceSnapshot', () => {
|
||||
test('returns the full body without abort involvement', async () => {
|
||||
const response = new Response(JSON.stringify({ ok: true }), {
|
||||
status: 200,
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
})
|
||||
const capture = await captureResponseTraceSnapshot(response, { signal: new AbortController().signal })
|
||||
expect(capture.aborted).toBe(false)
|
||||
expect(capture.snapshot.preview).toContain('"ok"')
|
||||
expect(capture.snapshot.truncated).toBe(false)
|
||||
})
|
||||
|
||||
test('finishes with partial data when aborted mid-stream', async () => {
|
||||
const stream = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
controller.enqueue(new TextEncoder().encode('partial sse data'))
|
||||
},
|
||||
})
|
||||
const response = new Response(stream, {
|
||||
status: 200,
|
||||
headers: { 'Content-Type': 'text/event-stream' },
|
||||
})
|
||||
|
||||
const controller = new AbortController()
|
||||
const capturePromise = captureResponseTraceSnapshot(response, { signal: controller.signal })
|
||||
await new Promise((resolve) => setTimeout(resolve, 10))
|
||||
controller.abort(new Error('client timeout'))
|
||||
|
||||
const capture = await capturePromise
|
||||
expect(capture.aborted).toBe(true)
|
||||
expect((capture.abortReason as Error).message).toBe('client timeout')
|
||||
expect(capture.snapshot.preview).toContain('partial sse data')
|
||||
expect(capture.snapshot.truncated).toBe(true)
|
||||
})
|
||||
|
||||
test('force-finishes after the grace period when cancel cannot wake a hung read', async () => {
|
||||
const encoder = new TextEncoder()
|
||||
let reads = 0
|
||||
let cancelled = false
|
||||
const fakeReader = {
|
||||
read() {
|
||||
reads += 1
|
||||
if (reads === 1) {
|
||||
return Promise.resolve({ done: false, value: encoder.encode('stuck partial body') })
|
||||
}
|
||||
// Hangs forever even after cancel(): models a runtime where
|
||||
// reader.cancel() does not settle a pending read.
|
||||
return new Promise(() => {})
|
||||
},
|
||||
cancel() {
|
||||
cancelled = true
|
||||
return Promise.resolve()
|
||||
},
|
||||
releaseLock() {},
|
||||
}
|
||||
const fakeResponse = {
|
||||
headers: new Headers({ 'content-type': 'text/event-stream' }),
|
||||
body: { getReader: () => fakeReader },
|
||||
} as unknown as Response
|
||||
|
||||
const controller = new AbortController()
|
||||
const capturePromise = captureResponseTraceSnapshot(fakeResponse, {
|
||||
signal: controller.signal,
|
||||
abortGraceMs: 20,
|
||||
})
|
||||
await new Promise((resolve) => setTimeout(resolve, 10))
|
||||
controller.abort()
|
||||
|
||||
const capture = await capturePromise
|
||||
expect(cancelled).toBe(true)
|
||||
expect(capture.aborted).toBe(true)
|
||||
expect(capture.snapshot.preview).toContain('stuck partial body')
|
||||
expect(capture.snapshot.truncated).toBe(true)
|
||||
})
|
||||
|
||||
test('treats an already-aborted signal as an immediate abort', async () => {
|
||||
const stream = new ReadableStream<Uint8Array>({
|
||||
start() {
|
||||
// Never produces data.
|
||||
},
|
||||
})
|
||||
const response = new Response(stream, {
|
||||
status: 200,
|
||||
headers: { 'Content-Type': 'text/event-stream' },
|
||||
})
|
||||
const controller = new AbortController()
|
||||
controller.abort()
|
||||
|
||||
const capture = await captureResponseTraceSnapshot(response, {
|
||||
signal: controller.signal,
|
||||
abortGraceMs: 50,
|
||||
})
|
||||
expect(capture.aborted).toBe(true)
|
||||
expect(capture.snapshot.preview).toBe('')
|
||||
})
|
||||
})
|
||||
|
||||
describe('session trace API', () => {
|
||||
|
||||
@ -1,4 +1,5 @@
|
||||
export {
|
||||
captureResponseTraceSnapshot,
|
||||
clearTraceCaptureStateForTests,
|
||||
createTraceCallId,
|
||||
createTraceBodySnapshot,
|
||||
@ -7,6 +8,7 @@ export {
|
||||
readTraceCaptureSettings,
|
||||
readResponseTraceSnapshot,
|
||||
shouldCaptureApiTrace,
|
||||
TRACE_ABORT_CAPTURE_GRACE_MS,
|
||||
TRACE_LIST_PREVIEW_CHARS,
|
||||
TRACE_STREAM_CAPTURE_BYTES,
|
||||
traceCaptureService,
|
||||
@ -24,6 +26,7 @@ export type {
|
||||
TraceEventRecord,
|
||||
TraceEventSeverity,
|
||||
TraceProviderInfo,
|
||||
TraceResponseCapture,
|
||||
TraceSession,
|
||||
TraceSessionFileItem,
|
||||
TraceSessionFileList,
|
||||
|
||||
@ -6,13 +6,13 @@ import { getSessionId } from 'src/bootstrap/state.js'
|
||||
import { getClaudeConfigHomeDir } from '../../utils/envUtils.js'
|
||||
import { jsonParse, jsonStringify } from '../../utils/slowOperations.js'
|
||||
import {
|
||||
captureResponseTraceSnapshot,
|
||||
createTraceCallId,
|
||||
createTraceBodySnapshot,
|
||||
readResponseTraceSnapshot,
|
||||
shouldCaptureApiTrace,
|
||||
traceCaptureService,
|
||||
} from './traceCapture.js'
|
||||
import type { TraceBodySnapshot, TraceProviderInfo } from './traceCapture.js'
|
||||
import type { TraceBodySnapshot, TraceProviderInfo, TraceResponseCapture } from './traceCapture.js'
|
||||
|
||||
const TRACE_SESSION_HEADER = 'x-claude-code-session-id'
|
||||
|
||||
@ -194,6 +194,29 @@ function createRequestPendingSnapshot(body: unknown): TraceBodySnapshot {
|
||||
})
|
||||
}
|
||||
|
||||
function isAbortLikeError(error: unknown): boolean {
|
||||
return error instanceof Error && (error.name === 'AbortError' || error.name === 'TimeoutError')
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds the error recorded for an aborted call. The fetch layer cannot see
|
||||
* which mechanism aborted the request (SDK client timeout surfaces upstream
|
||||
* as APIConnectionTimeoutError, the stream idle watchdog calls
|
||||
* stream.controller.abort(), user cancellation aborts the query signal), so
|
||||
* prefer the signal's abort reason when one was provided and otherwise name
|
||||
* the likely candidates.
|
||||
*/
|
||||
function normalizeAbortError(abortReason: unknown, fallback?: unknown): Error {
|
||||
if (abortReason instanceof Error) return abortReason
|
||||
if (abortReason !== undefined) return new Error(String(abortReason))
|
||||
if (fallback instanceof Error && isAbortLikeError(fallback)) return fallback
|
||||
const error = new Error(
|
||||
'Request aborted before the response completed (client timeout, stream idle watchdog, or user cancellation)',
|
||||
)
|
||||
error.name = 'AbortError'
|
||||
return error
|
||||
}
|
||||
|
||||
export function createDumpPromptsFetch(
|
||||
agentIdOrSessionId: string,
|
||||
options?: {
|
||||
@ -292,6 +315,7 @@ export function createDumpPromptsFetch(
|
||||
} catch (err) {
|
||||
if (timestamp && traceCallId) {
|
||||
const completedAt = new Date().toISOString()
|
||||
const aborted = Boolean(init?.signal?.aborted) || isAbortLikeError(err)
|
||||
void traceCaptureService.recordCall({
|
||||
id: traceCallId,
|
||||
sessionId: traceSessionId,
|
||||
@ -312,6 +336,7 @@ export function createDumpPromptsFetch(
|
||||
error: err,
|
||||
metadata: {
|
||||
phase: 'api_call_failed',
|
||||
...(aborted ? { aborted: true } : {}),
|
||||
},
|
||||
})
|
||||
void traceCaptureService.recordEvent({
|
||||
@ -327,6 +352,7 @@ export function createDumpPromptsFetch(
|
||||
message: err instanceof Error ? err.message : String(err),
|
||||
metadata: {
|
||||
url: requestUrl,
|
||||
...(aborted ? { aborted: true } : {}),
|
||||
},
|
||||
})
|
||||
}
|
||||
@ -335,40 +361,60 @@ export function createDumpPromptsFetch(
|
||||
|
||||
if (timestamp && traceCallId) {
|
||||
const cloned = response.clone()
|
||||
const abortSignal = init?.signal ?? undefined
|
||||
void (async () => {
|
||||
const callBase = {
|
||||
id: traceCallId,
|
||||
sessionId: traceSessionId,
|
||||
source: 'anthropic' as const,
|
||||
querySource: options?.querySource,
|
||||
provider: traceProvider,
|
||||
model: traceModel,
|
||||
startedAt: timestamp,
|
||||
request: {
|
||||
method: init?.method ?? 'POST',
|
||||
url: requestUrl,
|
||||
headers: requestInit?.headers as Headers | Record<string, string> | undefined,
|
||||
body: traceRequestBody,
|
||||
},
|
||||
}
|
||||
const eventBase = {
|
||||
sessionId: traceSessionId,
|
||||
callId: traceCallId,
|
||||
source: 'anthropic' as const,
|
||||
provider: traceProvider,
|
||||
model: traceModel,
|
||||
}
|
||||
|
||||
// captureResponseTraceSnapshot ends promptly when the request is
|
||||
// aborted mid-body (SDK client timeout, stream idle watchdog,
|
||||
// non-streaming fallback timeout). Waiting on the clone alone could
|
||||
// hang forever on a wedged stream, leaving this call pending in the
|
||||
// trace panel with no completion or error record (#766).
|
||||
let capture: TraceResponseCapture | undefined
|
||||
let captureFailure: unknown
|
||||
try {
|
||||
const bodySnapshot = await readResponseTraceSnapshot(cloned)
|
||||
capture = await captureResponseTraceSnapshot(cloned, { signal: abortSignal })
|
||||
} catch (err) {
|
||||
captureFailure = err
|
||||
}
|
||||
|
||||
if (capture && !capture.aborted) {
|
||||
await traceCaptureService.recordCall({
|
||||
id: traceCallId,
|
||||
sessionId: traceSessionId,
|
||||
source: 'anthropic',
|
||||
querySource: options?.querySource,
|
||||
provider: traceProvider,
|
||||
model: traceModel,
|
||||
startedAt: timestamp,
|
||||
...callBase,
|
||||
completedAt: new Date().toISOString(),
|
||||
durationMs: Date.now() - traceStartedAtMs,
|
||||
request: {
|
||||
method: init?.method ?? 'POST',
|
||||
url: requestUrl,
|
||||
headers: requestInit?.headers as Headers | Record<string, string> | undefined,
|
||||
body: traceRequestBody,
|
||||
},
|
||||
response: {
|
||||
status: response.status,
|
||||
headers: response.headers,
|
||||
bodySnapshot,
|
||||
bodySnapshot: capture.snapshot,
|
||||
},
|
||||
metadata: {
|
||||
phase: 'api_call_completed',
|
||||
},
|
||||
})
|
||||
await traceCaptureService.recordEvent({
|
||||
sessionId: traceSessionId,
|
||||
callId: traceCallId,
|
||||
source: 'anthropic',
|
||||
provider: traceProvider,
|
||||
model: traceModel,
|
||||
...eventBase,
|
||||
phase: 'api_call_completed',
|
||||
severity: response.ok ? 'info' : 'warning',
|
||||
title: 'API call completed',
|
||||
@ -377,48 +423,76 @@ export function createDumpPromptsFetch(
|
||||
url: requestUrl,
|
||||
},
|
||||
})
|
||||
} catch (err) {
|
||||
await traceCaptureService.recordEvent({
|
||||
sessionId: traceSessionId,
|
||||
callId: traceCallId,
|
||||
source: 'anthropic',
|
||||
provider: traceProvider,
|
||||
model: traceModel,
|
||||
phase: 'response_capture_failed',
|
||||
severity: 'warning',
|
||||
title: 'Response capture failed',
|
||||
message: err instanceof Error ? err.message : String(err),
|
||||
metadata: {
|
||||
status: response.status,
|
||||
url: requestUrl,
|
||||
},
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
const completedAt = new Date().toISOString()
|
||||
const durationMs = Date.now() - traceStartedAtMs
|
||||
if (capture?.aborted) {
|
||||
const abortError = normalizeAbortError(capture.abortReason, captureFailure)
|
||||
await traceCaptureService.recordCall({
|
||||
id: traceCallId,
|
||||
sessionId: traceSessionId,
|
||||
source: 'anthropic',
|
||||
querySource: options?.querySource,
|
||||
provider: traceProvider,
|
||||
model: traceModel,
|
||||
startedAt: timestamp,
|
||||
completedAt: new Date().toISOString(),
|
||||
durationMs: Date.now() - traceStartedAtMs,
|
||||
request: {
|
||||
method: init?.method ?? 'POST',
|
||||
url: requestUrl,
|
||||
headers: requestInit?.headers as Headers | Record<string, string> | undefined,
|
||||
body: traceRequestBody,
|
||||
},
|
||||
...callBase,
|
||||
status: 'error',
|
||||
completedAt,
|
||||
durationMs,
|
||||
response: {
|
||||
status: response.status,
|
||||
headers: response.headers,
|
||||
bodySnapshot: capture.snapshot,
|
||||
},
|
||||
error: abortError,
|
||||
metadata: {
|
||||
phase: 'api_call_completed',
|
||||
responseCaptureFailed: true,
|
||||
phase: 'api_call_aborted',
|
||||
aborted: true,
|
||||
},
|
||||
})
|
||||
await traceCaptureService.recordEvent({
|
||||
...eventBase,
|
||||
timestamp: completedAt,
|
||||
phase: 'api_call_aborted',
|
||||
severity: 'error',
|
||||
title: 'API call aborted',
|
||||
message: abortError.message,
|
||||
metadata: {
|
||||
status: response.status,
|
||||
url: requestUrl,
|
||||
durationMs,
|
||||
aborted: true,
|
||||
},
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
await traceCaptureService.recordEvent({
|
||||
...eventBase,
|
||||
timestamp: completedAt,
|
||||
phase: 'response_capture_failed',
|
||||
severity: 'warning',
|
||||
title: 'Response capture failed',
|
||||
message: captureFailure instanceof Error ? captureFailure.message : String(captureFailure),
|
||||
metadata: {
|
||||
status: response.status,
|
||||
url: requestUrl,
|
||||
},
|
||||
})
|
||||
// The clone shares the upstream source with the SDK's branch, so a
|
||||
// read failure here almost certainly failed the real request too.
|
||||
// Record an error state instead of letting the call sit pending.
|
||||
await traceCaptureService.recordCall({
|
||||
...callBase,
|
||||
status: 'error',
|
||||
completedAt,
|
||||
durationMs,
|
||||
response: {
|
||||
status: response.status,
|
||||
headers: response.headers,
|
||||
},
|
||||
error: captureFailure,
|
||||
metadata: {
|
||||
phase: 'response_capture_failed',
|
||||
responseCaptureFailed: true,
|
||||
},
|
||||
})
|
||||
})()
|
||||
}
|
||||
|
||||
|
||||
@ -449,10 +449,33 @@ class TraceCaptureService {
|
||||
|
||||
export const traceCaptureService = new TraceCaptureService()
|
||||
|
||||
export type TraceResponseCapture = {
|
||||
snapshot: TraceBodySnapshot
|
||||
aborted: boolean
|
||||
abortReason?: unknown
|
||||
}
|
||||
|
||||
export const TRACE_ABORT_CAPTURE_GRACE_MS = 2000
|
||||
|
||||
export async function readResponseTraceSnapshot(response: Response): Promise<TraceBodySnapshot> {
|
||||
return (await captureResponseTraceSnapshot(response)).snapshot
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads a response body into a trace snapshot, ending promptly when `signal`
|
||||
* aborts (SDK client timeout, stream idle watchdog, user cancellation).
|
||||
* Without this, an aborted upstream stream can leave the read pending forever
|
||||
* and the trace call stuck in `pending` (#766). On abort the partial body is
|
||||
* returned with `aborted: true` so callers can record an error-state call.
|
||||
*/
|
||||
export async function captureResponseTraceSnapshot(
|
||||
response: Response,
|
||||
options?: { signal?: AbortSignal; abortGraceMs?: number },
|
||||
): Promise<TraceResponseCapture> {
|
||||
const signal = options?.signal
|
||||
const contentType = response.headers.get('content-type') ?? ''
|
||||
if (!response.body) {
|
||||
return createTraceBodySnapshot(null)
|
||||
return { snapshot: createTraceBodySnapshot(null), aborted: false }
|
||||
}
|
||||
|
||||
const reader = response.body.getReader()
|
||||
@ -460,11 +483,18 @@ export async function readResponseTraceSnapshot(response: Response): Promise<Tra
|
||||
let text = ''
|
||||
let bytes = 0
|
||||
let truncated = false
|
||||
let completed = false
|
||||
let interrupted = false
|
||||
let onAbort: (() => void) | undefined
|
||||
let graceTimer: ReturnType<typeof setTimeout> | undefined
|
||||
|
||||
try {
|
||||
const readAll = async (): Promise<'done'> => {
|
||||
while (true) {
|
||||
const { done, value } = await reader.read()
|
||||
if (done) break
|
||||
if (done) {
|
||||
completed = true
|
||||
break
|
||||
}
|
||||
bytes += value.byteLength
|
||||
if (text.length < TRACE_STREAM_CAPTURE_BYTES) {
|
||||
text += decoder.decode(value, { stream: true })
|
||||
@ -475,15 +505,54 @@ export async function readResponseTraceSnapshot(response: Response): Promise<Tra
|
||||
truncated = true
|
||||
}
|
||||
}
|
||||
text += decoder.decode()
|
||||
} finally {
|
||||
reader.releaseLock()
|
||||
return 'done'
|
||||
}
|
||||
|
||||
return createTraceBodySnapshot(
|
||||
// Resolves only when the abort grace period expires with the read still
|
||||
// hung. Spec-compliant runtimes resolve the pending read() with done after
|
||||
// reader.cancel(), letting readAll() win the race; the timer is the
|
||||
// backstop for runtimes where cancel() does not wake a pending read.
|
||||
const forcedAbort = new Promise<'forced'>((resolve) => {
|
||||
if (!signal) return
|
||||
onAbort = () => {
|
||||
if (completed) return
|
||||
interrupted = true
|
||||
void reader.cancel().catch(() => {})
|
||||
graceTimer = setTimeout(() => resolve('forced'), options?.abortGraceMs ?? TRACE_ABORT_CAPTURE_GRACE_MS)
|
||||
graceTimer.unref?.()
|
||||
}
|
||||
if (signal.aborted) onAbort()
|
||||
else signal.addEventListener('abort', onAbort, { once: true })
|
||||
})
|
||||
|
||||
try {
|
||||
await Promise.race([readAll(), forcedAbort])
|
||||
} catch (err) {
|
||||
// Some runtimes reject the pending read() on abort instead of resolving
|
||||
// it after cancel(); fold that into the aborted outcome. Genuine read
|
||||
// failures (no abort in flight) propagate to the caller.
|
||||
if (!interrupted && !signal?.aborted) throw err
|
||||
interrupted = true
|
||||
} finally {
|
||||
if (signal && onAbort) signal.removeEventListener('abort', onAbort)
|
||||
if (graceTimer) clearTimeout(graceTimer)
|
||||
try {
|
||||
reader.releaseLock()
|
||||
} catch {
|
||||
// A still-pending read keeps the lock; the reader is abandoned with it.
|
||||
}
|
||||
}
|
||||
|
||||
text += decoder.decode()
|
||||
const snapshot = createTraceBodySnapshot(
|
||||
contentType.includes('application/json') ? parseJsonOrText(text) : text,
|
||||
{ alreadyTruncated: truncated },
|
||||
{ alreadyTruncated: truncated || interrupted },
|
||||
)
|
||||
return {
|
||||
snapshot,
|
||||
aborted: interrupted,
|
||||
...(interrupted && signal?.reason !== undefined ? { abortReason: signal.reason } : {}),
|
||||
}
|
||||
}
|
||||
|
||||
function serializeTraceBody(body: unknown): { serialized: string; contentType: TraceBodySnapshot['contentType'] } {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user