From 6850eb2fb9b1c1270b322565f45a21ef59ef884f Mon Sep 17 00:00:00 2001 From: xiaozh Date: Thu, 3 Sep 2026 02:34:59 +0800 Subject: [PATCH] fix(runtime): recover once from output-free streams closed before completion Generated-by: pi (gpt-5.6-sol) --- packages/core/src/events.ts | 1 + .../src/__tests__/protocol.test.ts | 24 ++ packages/runtime-host/src/protocol/index.ts | 4 +- packages/runtime-host/src/protocol/turn.ts | 1 + .../src/__tests__/ai-sdk-backend.test.ts | 347 +++++++++++++++++- .../__tests__/model-adapter-onerror.test.ts | 38 ++ .../provider-error-classification.test.ts | 64 ++++ packages/runtime/src/ai-sdk-turn.ts | 21 +- packages/runtime/src/model-adapter.ts | 1 + packages/runtime/src/model-protocol.ts | 2 + .../src/provider-error-classification.ts | 58 ++- .../__tests__/live-turn-projection.test.ts | 15 + packages/ui/src/conversation-copy.ts | 6 +- 13 files changed, 558 insertions(+), 24 deletions(-) diff --git a/packages/core/src/events.ts b/packages/core/src/events.ts index a30d3999d4..64d53f555f 100644 --- a/packages/core/src/events.ts +++ b/packages/core/src/events.ts @@ -1208,6 +1208,7 @@ export type ProviderRetryReason = | 'provider_unavailable' | 'rate_limit' | 'timeout' + | 'incomplete_stream' | 'unknown'; /** diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index 9ac1cfde4f..7931e53e88 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -60,10 +60,34 @@ import { TURN_MESSAGE_QUOTE_TEXT_MAX_LENGTH, TURN_FAILURE_MESSAGE_MAX_BYTES, decodeMessageContent, + decodeTurnProviderRetry, TURN_SKILL_ID_MAX_COUNT, TURN_SKILL_ID_MAX_LENGTH, } from '../protocol/turn.js'; +test('decodes the protocol-incomplete stream retry reason', () => { + assert.deepEqual( + decodeTurnProviderRetry({ + phase: 'scheduled', + attempt: 2, + maxAttempts: 2, + delayMs: 1_000, + reason: 'incomplete_stream', + }), + { + phase: 'scheduled', + attempt: 2, + maxAttempts: 2, + delayMs: 1_000, + reason: 'incomplete_stream', + }, + ); +}); + +test('publishes a new compatibility epoch for protocol-incomplete retry progress', () => { + assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 109); +}); + describe('Runtime Host bootstrap protocol', () => { test('accepts only authenticated-listener registration endpoints on IPv4 loopback', () => { const registration = { diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index b58467c6ee..6c1669c829 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -101,7 +101,9 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 112 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 113 as const; +// 113: Turn provider-retry progress may carry the `incomplete_stream` reason; +// older peers reject that closed-union value. // 112: Owners can query the Host execution environment through an extensible, // bounded resource-envelope contract. Older Hosts do not implement the query. // 111: Client Capability tool schemas may use draft-07 tuple additionalItems. diff --git a/packages/runtime-host/src/protocol/turn.ts b/packages/runtime-host/src/protocol/turn.ts index dafa428649..80e0df6078 100644 --- a/packages/runtime-host/src/protocol/turn.ts +++ b/packages/runtime-host/src/protocol/turn.ts @@ -792,6 +792,7 @@ function requireProviderRetryReason(value: unknown): ProviderRetryReason { value === 'provider_unavailable' || value === 'rate_limit' || value === 'timeout' || + value === 'incomplete_stream' || value === 'unknown' ) { return value; diff --git a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts index 055875101e..93b04e3f69 100644 --- a/packages/runtime/src/__tests__/ai-sdk-backend.test.ts +++ b/packages/runtime/src/__tests__/ai-sdk-backend.test.ts @@ -23,7 +23,12 @@ import { Buffer } from 'node:buffer'; import { createHash } from 'node:crypto'; import { join, resolve } from 'node:path'; import { describe, test } from 'node:test'; -import type { ModelMessage, ModelStreamResult } from '../model-protocol.js'; +import type { + ModelFailure, + ModelMessage, + ModelStreamEvent, + ModelStreamResult, +} from '../model-protocol.js'; import { MockLanguageModelV4, simulateReadableStream } from 'ai/test'; import { APICallError, type LanguageModelV4StreamPart } from '@ai-sdk/provider'; import type { RuntimeInvocationRootAuthority } from '@maka/core/runtime-event'; @@ -7312,6 +7317,323 @@ describe('AiSdkBackend usage telemetry', () => { assert.equal(events.find((event) => event.type === 'complete')?.stopReason, 'end_turn'); }); + test('retries an output-free protocol-incomplete stream once with a distinct reason', async () => { + const durable = durableTurnHarness('turn-incomplete-retry', 'finish the report'); + let calls = 0; + const model = new MockLanguageModelV4({ + doStream: async () => { + calls += 1; + const chunks: LanguageModelV4StreamPart[] = + calls === 1 + ? [{ type: 'stream-start', warnings: [] }, incompleteStreamErrorPart()] + : [ + { type: 'stream-start', warnings: [] }, + { type: 'text-start', id: 'text-1' }, + { type: 'text-delta', id: 'text-1', delta: 'Recovered' }, + { type: 'text-end', id: 'text-1' }, + { + type: 'finish', + finishReason: { unified: 'stop', raw: 'stop' }, + usage: emptyUsage(), + }, + ]; + return { + stream: simulateReadableStream({ + chunks, + initialDelayInMs: null, + chunkDelayInMs: null, + }), + }; + }, + }); + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + loadTurnRuntimeEvents: durable.loadTurnRuntimeEvents, + newId: idGenerator(), + now: monotonicClock(), + providerRetrySleep: async () => {}, + }); + + const events = await drainDurably(backend.send(durable.input()), durable); + + assert.equal(calls, 2); + assert.deepEqual( + events + .filter((event) => event.type === 'provider_retry') + .map(({ phase, attempt, maxAttempts, reason }) => ({ + phase, + attempt, + maxAttempts, + reason, + })), + [ + { phase: 'scheduled', attempt: 2, maxAttempts: 2, reason: 'incomplete_stream' }, + { phase: 'started', attempt: 2, maxAttempts: 2, reason: 'incomplete_stream' }, + ], + ); + assert.equal( + events.some((event) => event.type === 'error'), + false, + ); + assert.equal(events.find((event) => event.type === 'complete')?.stopReason, 'end_turn'); + }); + + test('stops after one protocol-incomplete stream recovery and preserves the provider error', async () => { + const durable = durableTurnHarness('turn-incomplete-exhausted', 'finish the report'); + let calls = 0; + const model = new MockLanguageModelV4({ + doStream: async () => { + calls += 1; + return { + stream: simulateReadableStream({ + chunks: [{ type: 'stream-start', warnings: [] }, incompleteStreamErrorPart()], + initialDelayInMs: null, + chunkDelayInMs: null, + }), + }; + }, + }); + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + loadTurnRuntimeEvents: durable.loadTurnRuntimeEvents, + newId: idGenerator(), + now: monotonicClock(), + providerRetrySleep: async () => {}, + }); + + const events = await drainDurably(backend.send(durable.input()), durable); + const error = events.find( + (event): event is Extract => event.type === 'error', + ); + + assert.equal(calls, 2); + assert.equal( + events.filter((event) => event.type === 'provider_retry' && event.phase === 'scheduled') + .length, + 1, + ); + assert.equal(error?.code, 'invalid_request_error'); + assert.match(error?.message ?? '', /stream closed before response\.completed/); + assert.equal(events.find((event) => event.type === 'complete')?.stopReason, 'error'); + }); + + test('fails closed on protocol-incomplete streams after observable output or a step boundary', async () => { + const cases: Array<{ name: string; events: ModelStreamEvent[] }> = [ + { name: 'text', events: [{ kind: 'text', text: 'partial answer' }] }, + { name: 'thinking', events: [{ kind: 'thinking', text: 'partial thought' }] }, + { name: 'tool activity', events: [{ kind: 'provider-tool-input' }] }, + { + name: 'continuation metadata', + events: [{ kind: 'text-end', providerOptions: { openai: { itemId: 'item-1' } } }], + }, + { name: 'completed step', events: [{ kind: 'step-finish', finishReason: 'stop' }] }, + ]; + + for (const candidate of cases) { + let calls = 0; + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => completionModel(), + tools: [], + newId: idGenerator(), + now: monotonicClock(), + providerRetrySleep: async () => {}, + }); + const failure = incompleteStreamFailure(); + ( + backend as unknown as { + modelAdapter: { startStream: () => Promise }; + } + ).modelAdapter.startStream = async () => { + calls += 1; + return { + events: (async function* () { + for (const event of candidate.events) yield event; + yield { kind: 'error' as const, failure }; + })(), + outcome: Promise.resolve({ + kind: 'terminal-failure', + failure, + request: { messages: [] }, + continuation: 'none', + }), + }; + }; + + const events: SessionEvent[] = []; + for await (const event of backend.send({ + turnId: `turn-${candidate.name}`, + text: 'hi', + context: [], + })) { + events.push(event); + } + + assert.equal(calls, 1, candidate.name); + assert.equal( + events.some((event) => event.type === 'provider_retry'), + false, + candidate.name, + ); + assert.equal( + events.find((event) => event.type === 'complete')?.stopReason, + 'error', + candidate.name, + ); + } + }); + + test('retries a post-tool protocol-incomplete stream without re-running the durable tool', async () => { + const durable = durableTurnHarness('turn-incomplete-after-tool', 'read notes'); + let providerCalls = 0; + let toolCalls = 0; + const model = new MockLanguageModelV4({ + doStream: async () => { + providerCalls += 1; + const chunks: LanguageModelV4StreamPart[] = + providerCalls === 1 + ? [ + { type: 'stream-start', warnings: [] }, + { + type: 'tool-call', + toolCallId: 'read-1', + toolName: 'Read', + input: JSON.stringify({ path: 'notes.md' }), + }, + { + type: 'finish', + finishReason: { unified: 'tool-calls', raw: 'tool_calls' }, + usage: emptyUsage(), + }, + ] + : providerCalls === 2 + ? [{ type: 'stream-start', warnings: [] }, incompleteStreamErrorPart()] + : [ + { type: 'stream-start', warnings: [] }, + { type: 'text-start', id: 'text-final' }, + { type: 'text-delta', id: 'text-final', delta: 'done' }, + { type: 'text-end', id: 'text-final' }, + { + type: 'finish', + finishReason: { unified: 'stop', raw: 'stop' }, + usage: emptyUsage(), + }, + ]; + return { + stream: simulateReadableStream({ + chunks, + initialDelayInMs: null, + chunkDelayInMs: null, + }), + }; + }, + }); + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [ + { + name: 'Read', + description: 'Read notes', + parameters: z.object({ path: z.string() }), + impl: async () => { + toolCalls += 1; + return 'notes contents'; + }, + }, + ], + loadTurnRuntimeEvents: durable.loadTurnRuntimeEvents, + newId: idGenerator(), + now: monotonicClock(), + providerRetrySleep: async () => {}, + }); + + const events = await drainDurably(backend.send(durable.input()), durable); + + assert.equal(providerCalls, 3); + assert.equal(toolCalls, 1); + assert.equal(events.filter((event) => event.type === 'tool_result').length, 1); + assert.equal( + durable.ledger.filter((event) => event.content?.kind === 'function_response').length, + 1, + ); + assert.equal(events.find((event) => event.type === 'complete')?.stopReason, 'end_turn'); + }); + + test('Stop aborts the turn while protocol-incomplete recovery is waiting', async () => { + let calls = 0; + const model = new MockLanguageModelV4({ + doStream: async () => { + calls += 1; + return { + stream: simulateReadableStream({ + chunks: [{ type: 'stream-start', warnings: [] }, incompleteStreamErrorPart()], + initialDelayInMs: null, + chunkDelayInMs: null, + }), + }; + }, + }); + const backend = createTestAiSdkBackend({ + sessionId: 'session-1', + header: header(), + appendMessage: async () => {}, + connection: connection(), + apiKey: 'sk-test', + modelId: 'mock-model-id', + modelFactory: () => model, + tools: [], + newId: idGenerator(), + now: monotonicClock(), + providerRetrySleep: async (_delayMs, signal) => + await new Promise((_resolve, reject) => { + const abort = () => + reject(signal.reason ?? Object.assign(new Error('aborted'), { name: 'AbortError' })); + if (signal.aborted) abort(); + else signal.addEventListener('abort', abort, { once: true }); + }), + }); + + const events: SessionEvent[] = []; + for await (const event of backend.send({ turnId: 'turn-1', text: 'hi', context: [] })) { + events.push(event); + if (event.type === 'provider_retry' && event.phase === 'scheduled') { + await backend.stop('user_stop'); + } + } + + assert.equal(calls, 1); + assert.equal( + events.some((event) => event.type === 'provider_retry' && event.phase === 'started'), + false, + ); + assert.equal(events.find((event) => event.type === 'complete')?.stopReason, 'user_stop'); + }); + test('classifies an exhausted output-free truncated stream as provider unavailable', async () => { const durable = durableTurnHarness('turn-truncated-exhausted', 'analyse the image'); let calls = 0; @@ -15660,6 +15982,29 @@ function emptyUsage() { }; } +function incompleteStreamErrorPart(): LanguageModelV4StreamPart { + return { + type: 'error', + error: { + code: 'invalid_request_error', + message: + 'stream error: stream disconnected before completion: stream closed before response.completed', + }, + }; +} + +function incompleteStreamFailure(): ModelFailure & { recoveryReason: 'incomplete_stream' } { + return { + type: 'model_failure', + kind: 'unknown', + message: + 'stream error: stream disconnected before completion: stream closed before response.completed (code=invalid_request_error)', + retryable: false, + code: 'invalid_request_error', + recoveryReason: 'incomplete_stream', + }; +} + function imageReplayBackend( model: MockLanguageModelV4, options: { supportsVision: boolean; readAttachmentBytes: AttachmentByteReader }, diff --git a/packages/runtime/src/__tests__/model-adapter-onerror.test.ts b/packages/runtime/src/__tests__/model-adapter-onerror.test.ts index c789c6548a..83641913eb 100644 --- a/packages/runtime/src/__tests__/model-adapter-onerror.test.ts +++ b/packages/runtime/src/__tests__/model-adapter-onerror.test.ts @@ -298,6 +298,44 @@ describe('ModelAdapter.startStream onError', () => { }); }); + test('marks a protocol-incomplete stream error without changing retryability or provider code', async () => { + const message = + 'stream error: stream disconnected before completion: stream closed before response.completed'; + const model = new MockLanguageModelV4({ + doStream: async () => ({ + stream: convertArrayToReadableStream([ + { type: 'stream-start', warnings: [] }, + { type: 'error', error: { code: 'invalid_request_error', message } }, + ]), + }), + }); + const result = await newAdapter().startStream({ + model, + messages: [{ role: 'user', content: 'hi' }], + tools: {}, + activeTools: [], + onStreamActivity: () => {}, + abortSignal: new AbortController().signal, + repairToolCall: async () => null, + }); + + const failures = []; + for await (const event of result.events) { + if (event.kind === 'error') failures.push(event.failure); + } + + assert.equal(failures.length, 1); + assert.equal(failures[0]?.retryable, false); + assert.equal(failures[0]?.code, 'invalid_request_error'); + assert.equal( + (failures[0] as (typeof failures)[number] & { recoveryReason?: string })?.recoveryReason, + 'incomplete_stream', + ); + assert.match(failures[0]?.message ?? '', /stream closed before response\.completed/); + const outcome = await result.outcome; + assert.equal(outcome.kind, 'terminal-failure'); + }); + test('does not hide provider retries inside one adapter call', async () => { let providerCalls = 0; const model = new MockLanguageModelV4({ diff --git a/packages/runtime/src/__tests__/provider-error-classification.test.ts b/packages/runtime/src/__tests__/provider-error-classification.test.ts index bf0507a18c..9d0bcfde9f 100644 --- a/packages/runtime/src/__tests__/provider-error-classification.test.ts +++ b/packages/runtime/src/__tests__/provider-error-classification.test.ts @@ -275,6 +275,70 @@ describe('Provider error classification', () => { }); }); + test('marks a protocol-incomplete stream for bounded recovery without widening retryability', () => { + const message = + 'stream error: stream disconnected before completion: stream closed before response.completed'; + const typedCause = Object.assign(new Error('provider stream failed'), { + name: 'AI_APICallError', + cause: { code: 'invalid_request_error', message }, + }); + const typedMetadata = providerRetryMetadata(typedCause) as ReturnType< + typeof providerRetryMetadata + > & { recoveryReason?: string }; + + assert.deepEqual(typedMetadata, { + retryable: false, + recoveryReason: 'incomplete_stream', + }); + assert.deepEqual(providerFailureDiagnostic(typedCause), { + errorClass: 'Other', + providerCode: 'invalid_request_error', + retryable: false, + }); + + const textFallback = providerRetryMetadata(new Error(message)) as ReturnType< + typeof providerRetryMetadata + > & { recoveryReason?: string }; + assert.deepEqual(textFallback, { + retryable: false, + recoveryReason: 'incomplete_stream', + }); + }); + + test('requires both protocol-incomplete stream phrases for text fallback recovery', () => { + for (const message of [ + 'stream disconnected before completion', + 'stream closed before response.completed', + 'stream disconnected after response.completed', + ]) { + assert.deepEqual(providerRetryMetadata(new Error(message)), { retryable: false }, message); + } + }); + + test('does not let text fallback override a contradictory structured provider code', () => { + const message = + 'stream disconnected before completion: stream closed before response.completed'; + + assert.deepEqual(providerRetryMetadata({ code: 'insufficient_quota', message }), { + retryable: false, + }); + assert.deepEqual(providerRetryMetadata({ type: 'authentication_error', message }), { + retryable: false, + }); + assert.deepEqual( + providerRetryMetadata( + Object.assign(new Error(message), { cause: { code: 'insufficient_quota' } }), + ), + { retryable: false }, + ); + assert.deepEqual( + providerRetryMetadata( + Object.assign(new Error(message), { cause: { type: 'authentication_error' } }), + ), + { retryable: false }, + ); + }); + test('retries an AI SDK transport failure without an HTTP response', () => { const failure = Object.assign( new Error( diff --git a/packages/runtime/src/ai-sdk-turn.ts b/packages/runtime/src/ai-sdk-turn.ts index c501a66ac7..5c6d2bd10e 100644 --- a/packages/runtime/src/ai-sdk-turn.ts +++ b/packages/runtime/src/ai-sdk-turn.ts @@ -1991,12 +1991,6 @@ export class AiSdkTurn { const settledWatchdogTimeout = consumeWatchdogTimeout(); providerOutcome = await result.outcome; const incompleteStreamTerminal = providerOutcome.kind === 'truncated'; - const incompleteStreamHasNoObservableOutput = - incompleteStreamTerminal && - !attemptSawText && - !attemptSawThinking && - !attemptSawToolActivity && - !attemptSawContinuationMetadata; const attemptFailure = settledWatchdogTimeout?.error ?? (providerOutcome.kind === 'completed' ? undefined : providerOutcome.failure); @@ -2116,12 +2110,17 @@ export class AiSdkTurn { }; await this.deps.backend.appendMessage(note).catch(() => {}); } + const protocolIncompleteStreamFailure = + failure.recoveryReason === 'incomplete_stream'; + const incompleteStreamHasNoObservableOutput = + (incompleteStreamTerminal || protocolIncompleteStreamFailure) && + attemptHasNoObservableOutput(); const idleWatchdogRecovery = settledWatchdogTimeout?.phase === 'idle' && idleWatchdogRetryCount < MAX_IDLE_WATCHDOG_RETRIES_PER_STEP && attemptCanRecoverWithSealedThinking(); const incompleteStreamRecovery = - incompleteStreamTerminal && + (incompleteStreamTerminal || protocolIncompleteStreamFailure) && incompleteStreamRetryCount < MAX_INCOMPLETE_STREAM_RETRIES_PER_STEP && incompleteStreamHasNoObservableOutput; // Same seal-and-retry contract as the watchdog path, entered when @@ -2136,8 +2135,9 @@ export class AiSdkTurn { sealedThinkingRetryCount < MAX_SEALED_THINKING_RETRIES_PER_STEP && attemptCanRecoverWithSealedThinking() && !attemptHasNoObservableOutput(); + const ordinaryProviderRetry = failure.retryable && !protocolIncompleteStreamFailure; if ( - (failure.retryable || idleWatchdogRecovery || incompleteStreamRecovery) && + (ordinaryProviderRetry || idleWatchdogRecovery || incompleteStreamRecovery) && failure.kind !== 'context_overflow' && providerAttempt < MAX_PROVIDER_ATTEMPTS_PER_STEP && stepBudgetRemains && @@ -2165,7 +2165,10 @@ export class AiSdkTurn { idleWatchdogRecovery || incompleteStreamRecovery || sealedThinkingRecovery ? nextAttempt : MAX_PROVIDER_ATTEMPTS_PER_STEP; - const reason = providerRetryReason(failure.kind); + const reason: ProviderRetryReason = + protocolIncompleteStreamFailure && incompleteStreamRecovery + ? 'incomplete_stream' + : providerRetryReason(failure.kind); queue.push({ type: 'provider_retry', id: this.deps.newId(), diff --git a/packages/runtime/src/model-adapter.ts b/packages/runtime/src/model-adapter.ts index 83a62a01b0..dece1b7ae8 100644 --- a/packages/runtime/src/model-adapter.ts +++ b/packages/runtime/src/model-adapter.ts @@ -1280,6 +1280,7 @@ function normalizeModelFailure(error: unknown): ModelFailure { kind: modelFailureKind(errorClass), retryable: retry.retryable, ...(retry.retryAfterMs !== undefined ? { retryAfterMs: retry.retryAfterMs } : {}), + ...(retry.recoveryReason !== undefined ? { recoveryReason: retry.recoveryReason } : {}), ...(code !== undefined ? { code } : {}), message: presentation.message ?? generalizedErrorMessage(error), }; diff --git a/packages/runtime/src/model-protocol.ts b/packages/runtime/src/model-protocol.ts index 68e8a32c85..e885555efc 100644 --- a/packages/runtime/src/model-protocol.ts +++ b/packages/runtime/src/model-protocol.ts @@ -349,6 +349,8 @@ export interface ModelFailure { /** Provider-requested delay for the next physical attempt, in milliseconds. */ retryAfterMs?: number; code?: string; + /** Narrow Runtime-owned recovery path that remains separate from transport retryability. */ + recoveryReason?: 'incomplete_stream'; } /** diff --git a/packages/runtime/src/provider-error-classification.ts b/packages/runtime/src/provider-error-classification.ts index aaa73bbe9b..7a3eedd6bb 100644 --- a/packages/runtime/src/provider-error-classification.ts +++ b/packages/runtime/src/provider-error-classification.ts @@ -123,6 +123,7 @@ interface ProviderErrorFacts { export interface ProviderRetryMetadata { retryable: boolean; retryAfterMs?: number; + recoveryReason?: 'incomplete_stream'; } /** Bounded, allowlisted provider failure facts safe for durable telemetry. */ @@ -215,6 +216,36 @@ function parseRetryAfterMs(headers: Record): number | null | und return Math.ceil(delayMs); } +const INCOMPLETE_STREAM_PHRASES = [ + 'stream disconnected before completion', + 'stream closed before response.completed', +] as const; + +function incompleteStreamRecoveryReason(error: unknown): 'incomplete_stream' | undefined { + let current: unknown = providerErrorTarget(error); + let typedMatch = false; + let textFallback = false; + let sawStructuredEvidence = false; + const seen = new Set(); + for (let depth = 0; depth < 5 && current !== undefined && !seen.has(current); depth += 1) { + seen.add(current); + const facts = normalizeProviderError(current); + if (facts) { + const codes = [facts.evidence.code.toLowerCase(), ...facts.evidence.structuredCodes].filter( + Boolean, + ); + sawStructuredEvidence ||= codes.length > 0; + if (INCOMPLETE_STREAM_PHRASES.every((phrase) => facts.evidence.text.includes(phrase))) { + typedMatch ||= codes.includes('invalid_request_error'); + if (codes.length === 0) textFallback = true; + } + } + const record = objectRecord(current); + current = record ? safeField(record, 'cause') : undefined; + } + return typedMatch || (textFallback && !sawStructuredEvidence) ? 'incomplete_stream' : undefined; +} + /** * Normalizes provider retry facts without leaking SDK error objects or raw * response headers across the ModelAdapter boundary. @@ -223,11 +254,16 @@ export function providerRetryMetadata(error: unknown): ProviderRetryMetadata { const facts = normalizeProviderError(error); if (!facts) return { retryable: false }; const { evidence } = facts; + const recoveryReason = incompleteStreamRecoveryReason(error); + const withRecoveryReason = (metadata: ProviderRetryMetadata): ProviderRetryMetadata => + recoveryReason ? { ...metadata, recoveryReason } : metadata; - if (RUNTIME_RETRYABLE_ERROR_CODES.has(evidence.code)) return { retryable: true }; + if (RUNTIME_RETRYABLE_ERROR_CODES.has(evidence.code)) { + return withRecoveryReason({ retryable: true }); + } // The Codex transport already spent its complete 2/10/30-second budget. // Do not let the outer model loop restart that same transport budget. - if (isTrustedCodexEdgeRejection(facts)) return { retryable: false }; + if (isTrustedCodexEdgeRejection(facts)) return withRecoveryReason({ retryable: false }); const status = Number(evidence.statusCode || evidence.code); const errorClass = classifyProviderFacts(facts); @@ -235,14 +271,16 @@ export function providerRetryMetadata(error: unknown): ProviderRetryMetadata { if (errorClass === 'ProviderCapacity') { // Capacity is transient even when the provider sends a malformed delay; // fall back to the adapter's bounded local backoff in that case. - return { + return withRecoveryReason({ retryable: true, ...(retryAfterMs !== undefined && retryAfterMs !== null ? { retryAfterMs } : {}), - }; + }); } if (errorClass === 'RateLimit' || status === 429) { - if (retryAfterMs === undefined || retryAfterMs === null) return { retryable: false }; - return { retryable: true, retryAfterMs }; + if (retryAfterMs === undefined || retryAfterMs === null) { + return withRecoveryReason({ retryable: false }); + } + return withRecoveryReason({ retryable: true, retryAfterMs }); } const retryable = errorClass === 'Network' || @@ -250,12 +288,12 @@ export function providerRetryMetadata(error: unknown): ProviderRetryMetadata { status === 408 || status === 409 || (status >= 500 && status <= 599); - if (!retryable) return { retryable: false }; - if (retryAfterMs === null) return { retryable: false }; - return { + if (!retryable) return withRecoveryReason({ retryable: false }); + if (retryAfterMs === null) return withRecoveryReason({ retryable: false }); + return withRecoveryReason({ retryable: true, ...(retryAfterMs !== undefined ? { retryAfterMs } : {}), - }; + }); } /** Collects `code`/`type` strings from a payload and from its `error` wrapper. */ diff --git a/packages/ui/src/__tests__/live-turn-projection.test.ts b/packages/ui/src/__tests__/live-turn-projection.test.ts index 29fc03fc7b..0e70e5a39c 100644 --- a/packages/ui/src/__tests__/live-turn-projection.test.ts +++ b/packages/ui/src/__tests__/live-turn-projection.test.ts @@ -76,6 +76,21 @@ describe('provider retry copy', () => { /capacity/, ); }); + + it('describes protocol-incomplete stream recovery independently', () => { + assert.match( + getConversationCopy('zh-CN').messages.providerRetryReason.incomplete_stream, + /提前结束/, + ); + assert.match( + getConversationCopy('zh-TW').messages.providerRetryReason.incomplete_stream, + /提前結束/, + ); + assert.match( + getConversationCopy('en').messages.providerRetryReason.incomplete_stream, + /ended early/, + ); + }); }); describe('applyLiveTurnEvent', () => { diff --git a/packages/ui/src/conversation-copy.ts b/packages/ui/src/conversation-copy.ts index 12ed2c76d6..307789baae 100644 --- a/packages/ui/src/conversation-copy.ts +++ b/packages/ui/src/conversation-copy.ts @@ -548,7 +548,7 @@ const CONVERSATION_COPY = { chooseAriaLabel: (label, branch) => branch ? `选择项目：${label}，当前分支 ${branch}` : `选择项目：${label}`, }, messages: { - you: '你', assistant: 'Maka', processing: '正在处理…', continuing: '继续中…', awaitingModelOutput: '等待模型输出…', providerRetryScheduled: (seconds, attempt, maxAttempts) => `${formatRetryDelay(seconds, 'zh-CN')}后重试（${attempt}/${maxAttempts}）`, providerRetryStarted: (attempt, maxAttempts) => `正在重试（${attempt}/${maxAttempts}）`, providerRetryWaiting: (attempt, maxAttempts) => `等待重试（${attempt}/${maxAttempts}）`, providerRetryReason: { network: '网络中断', provider_capacity: '模型服务暂时满载', provider_unavailable: '模型服务暂时不可用', rate_limit: '触发模型速率限制', timeout: '请求超时', unknown: '模型请求失败' }, safeResumePending: '正在检查…', safeResume: '继续这一轮', thinking: '深度思考', truncated: '已截断', copied: '已复制', copying: '复制中', copyFailed: '复制失败', copy: '复制', editMessage: '编辑并重发', editMessageDisabledRunning: '当前回答仍在进行中，结束后再编辑', editMessageDisabledAttachments: '包含附件的历史消息暂不支持编辑并重发', editMessageDisabledQuotes: '包含引用的历史消息暂不支持编辑并重发', editMessageDisabledTransformedText: '包含已展开上下文的历史消息暂不支持编辑并重发', + you: '你', assistant: 'Maka', processing: '正在处理…', continuing: '继续中…', awaitingModelOutput: '等待模型输出…', providerRetryScheduled: (seconds, attempt, maxAttempts) => `${formatRetryDelay(seconds, 'zh-CN')}后重试（${attempt}/${maxAttempts}）`, providerRetryStarted: (attempt, maxAttempts) => `正在重试（${attempt}/${maxAttempts}）`, providerRetryWaiting: (attempt, maxAttempts) => `等待重试（${attempt}/${maxAttempts}）`, providerRetryReason: { network: '网络中断', provider_capacity: '模型服务暂时满载', provider_unavailable: '模型服务暂时不可用', rate_limit: '触发模型速率限制', timeout: '请求超时', incomplete_stream: '模型流提前结束', unknown: '模型请求失败' }, safeResumePending: '正在检查…', safeResume: '继续这一轮', thinking: '深度思考', truncated: '已截断', copied: '已复制', copying: '复制中', copyFailed: '复制失败', copy: '复制', editMessage: '编辑并重发', editMessageDisabledRunning: '当前回答仍在进行中，结束后再编辑', editMessageDisabledAttachments: '包含附件的历史消息暂不支持编辑并重发', editMessageDisabledQuotes: '包含引用的历史消息暂不支持编辑并重发', editMessageDisabledTransformedText: '包含已展开上下文的历史消息暂不支持编辑并重发', editMessageDisabledDirectoryReferences: '包含文件夹引用的历史消息暂不支持编辑并重发', userAriaLabel: '你发送的消息', systemAriaLabel: '系统消息', assistantAriaLabel: 'Maka 的回答', answerActionsAriaLabel: (context) => `回答操作${context ? `：${context}` : ''}`, answerActionAriaLabel: (action, context) => `${action}回答${context ? `：${context}` : ''}`, messageActionAriaLabel: (action, context) => `${action}消息${context ? `：${context}` : ''}`, sourceAriaLabel: '本轮回答的来源', derivativesAriaLabel: '本轮回答的衍生', scheduledTaskTriggered: '定时任务触发', scheduledTaskTitle: (id) => `由定时任务触发 · ${id}`, legacyAutomationTriggered: '旧版自动化（仅历史）', legacyAutomationTitle: (id) => `由旧版自动化触发 · ${id} · 仅保留历史，不会再次执行`, goalContinued: 'Goal 自动继续', goalTitle: (id) => `由 Goal 继续执行 · ${id}`, agentGraphTriggered: 'Agent Graph 自动继续', agentGraphTitle: (graphId) => `由 Agent Graph 调度器触发 · ${graphId}`, thinkingTruncatedTitle: '部分 reasoning 已截断；显示的是最近的内容', outputTruncatedTitle: '助手输出已超过单次回合上限，超出部分未渲染。如需完整内容请重新生成或查看持久化的任务日志。', removeAttachmentAriaLabel: (name) => `移除 ${name}`, quoteLabel: '引用', quoteExpandAriaLabel: '展开引用全文', quoteCollapseAriaLabel: '收起引用', removeQuoteAriaLabel: '移除引用', aborted: '已中断', abortedByStop: '已中断 · 由停止按钮触发', @@ -700,7 +700,7 @@ const CONVERSATION_COPY = { chooseAriaLabel: (label, branch) => branch ? `選擇專案：${label}，目前分支 ${branch}` : `選擇專案：${label}`, }, messages: { - you: '你', assistant: 'Maka', processing: '正在處理…', continuing: '繼續中…', awaitingModelOutput: '等待模型輸出…', providerRetryScheduled: (seconds, attempt, maxAttempts) => `${formatRetryDelay(seconds, 'zh-TW')}後重試（${attempt}/${maxAttempts}）`, providerRetryStarted: (attempt, maxAttempts) => `正在重試（${attempt}/${maxAttempts}）`, providerRetryWaiting: (attempt, maxAttempts) => `等待重試（${attempt}/${maxAttempts}）`, providerRetryReason: { network: '網路中斷', provider_capacity: '模型服務暫時滿載', provider_unavailable: '模型服務暫時不可用', rate_limit: '觸發模型速率限制', timeout: '請求超時', unknown: '模型請求失敗' }, safeResumePending: '正在檢查…', safeResume: '繼續這一輪', thinking: '深度思考', truncated: '已截斷', copied: '已複製', copying: '複製中', copyFailed: '複製失敗', copy: '複製', editMessage: '編輯並重發', editMessageDisabledRunning: '目前回答仍在進行中，結束後再編輯', editMessageDisabledAttachments: '包含附件的歷史訊息暫不支援編輯並重發', editMessageDisabledQuotes: '包含引用的歷史訊息暫不支援編輯並重發', editMessageDisabledTransformedText: '包含已展開上下文的歷史訊息暫不支援編輯並重發', + you: '你', assistant: 'Maka', processing: '正在處理…', continuing: '繼續中…', awaitingModelOutput: '等待模型輸出…', providerRetryScheduled: (seconds, attempt, maxAttempts) => `${formatRetryDelay(seconds, 'zh-TW')}後重試（${attempt}/${maxAttempts}）`, providerRetryStarted: (attempt, maxAttempts) => `正在重試（${attempt}/${maxAttempts}）`, providerRetryWaiting: (attempt, maxAttempts) => `等待重試（${attempt}/${maxAttempts}）`, providerRetryReason: { network: '網路中斷', provider_capacity: '模型服務暫時滿載', provider_unavailable: '模型服務暫時不可用', rate_limit: '觸發模型速率限制', timeout: '請求超時', incomplete_stream: '模型串流提前結束', unknown: '模型請求失敗' }, safeResumePending: '正在檢查…', safeResume: '繼續這一輪', thinking: '深度思考', truncated: '已截斷', copied: '已複製', copying: '複製中', copyFailed: '複製失敗', copy: '複製', editMessage: '編輯並重發', editMessageDisabledRunning: '目前回答仍在進行中，結束後再編輯', editMessageDisabledAttachments: '包含附件的歷史訊息暫不支援編輯並重發', editMessageDisabledQuotes: '包含引用的歷史訊息暫不支援編輯並重發', editMessageDisabledTransformedText: '包含已展開上下文的歷史訊息暫不支援編輯並重發', editMessageDisabledDirectoryReferences: '包含資料夾引用的歷史訊息暫不支援編輯並重發', userAriaLabel: '你傳送的訊息', systemAriaLabel: '系統訊息', assistantAriaLabel: 'Maka 的回答', answerActionsAriaLabel: (context) => `回答操作${context ? `：${context}` : ''}`, answerActionAriaLabel: (action, context) => `${action}回答${context ? `：${context}` : ''}`, messageActionAriaLabel: (action, context) => `${action}訊息${context ? `：${context}` : ''}`, sourceAriaLabel: '本輪迴答的來源', derivativesAriaLabel: '本輪迴答的衍生', scheduledTaskTriggered: '定時任務觸發', scheduledTaskTitle: (id) => `由定時任務觸發 · ${id}`, legacyAutomationTriggered: '舊版自動化（僅歷史）', legacyAutomationTitle: (id) => `由舊版自動化觸發 · ${id} · 僅保留歷史，不會再次執行`, goalContinued: 'Goal 自動繼續', goalTitle: (id) => `由 Goal 繼續執行 · ${id}`, agentGraphTriggered: 'Agent Graph 自動繼續', agentGraphTitle: (graphId) => `由 Agent Graph 排程器觸發 · ${graphId}`, thinkingTruncatedTitle: '部分 reasoning 已截斷；顯示的是最近的內容', outputTruncatedTitle: '助手輸出已超過單次回合上限，超出部分未渲染。如需完整內容請重新生成或檢視持久化的任務記錄。', removeAttachmentAriaLabel: (name) => `移除 ${name}`, quoteLabel: '引用', quoteExpandAriaLabel: '展開引用全文', quoteCollapseAriaLabel: '收起引用', removeQuoteAriaLabel: '移除引用', aborted: '(已中斷)', abortedByStop: '(已中斷 · 由停止按鈕觸發)', @@ -878,7 +878,7 @@ const CONVERSATION_COPY = { chooseAriaLabel: (label, branch) => branch ? `Choose project: ${label}, current branch ${branch}` : `Choose project: ${label}`, }, messages: { - you: 'You', assistant: 'Maka', processing: 'Working…', continuing: 'Continuing…', awaitingModelOutput: 'Waiting for model output…', providerRetryScheduled: (seconds, attempt, maxAttempts) => `Retrying in ${formatRetryDelay(seconds, 'en')} (${attempt}/${maxAttempts})`, providerRetryStarted: (attempt, maxAttempts) => `Retrying (${attempt}/${maxAttempts})`, providerRetryWaiting: (attempt, maxAttempts) => `Waiting to retry (${attempt}/${maxAttempts})`, providerRetryReason: { network: 'Network interrupted', provider_capacity: 'The model service is temporarily at capacity', provider_unavailable: 'Model service temporarily unavailable', rate_limit: 'Model rate limit reached', timeout: 'Request timed out', unknown: 'Model request failed' }, safeResumePending: 'Checking…', safeResume: 'Continue this turn', thinking: 'Thinking', truncated: 'Truncated', copied: 'Copied', copying: 'Copying', copyFailed: 'Copy failed', copy: 'Copy', editMessage: 'Edit & resend', editMessageDisabledRunning: 'Wait for this answer to finish before editing', editMessageDisabledAttachments: 'Edit & resend does not yet support messages with attachments', editMessageDisabledQuotes: 'Edit & resend does not yet support messages with quotes', editMessageDisabledTransformedText: 'Edit & resend does not yet support messages with expanded context', + you: 'You', assistant: 'Maka', processing: 'Working…', continuing: 'Continuing…', awaitingModelOutput: 'Waiting for model output…', providerRetryScheduled: (seconds, attempt, maxAttempts) => `Retrying in ${formatRetryDelay(seconds, 'en')} (${attempt}/${maxAttempts})`, providerRetryStarted: (attempt, maxAttempts) => `Retrying (${attempt}/${maxAttempts})`, providerRetryWaiting: (attempt, maxAttempts) => `Waiting to retry (${attempt}/${maxAttempts})`, providerRetryReason: { network: 'Network interrupted', provider_capacity: 'The model service is temporarily at capacity', provider_unavailable: 'Model service temporarily unavailable', rate_limit: 'Model rate limit reached', timeout: 'Request timed out', incomplete_stream: 'Model stream ended early', unknown: 'Model request failed' }, safeResumePending: 'Checking…', safeResume: 'Continue this turn', thinking: 'Thinking', truncated: 'Truncated', copied: 'Copied', copying: 'Copying', copyFailed: 'Copy failed', copy: 'Copy', editMessage: 'Edit & resend', editMessageDisabledRunning: 'Wait for this answer to finish before editing', editMessageDisabledAttachments: 'Edit & resend does not yet support messages with attachments', editMessageDisabledQuotes: 'Edit & resend does not yet support messages with quotes', editMessageDisabledTransformedText: 'Edit & resend does not yet support messages with expanded context', editMessageDisabledDirectoryReferences: 'Edit & resend does not yet support messages with folder references', userAriaLabel: 'Your message', systemAriaLabel: 'System message', assistantAriaLabel: "Maka's response", answerActionsAriaLabel: (context) => `Response actions${context ? `: ${context}` : ''}`, answerActionAriaLabel: (action, context) => `${action} response${context ? `: ${context}` : ''}`, messageActionAriaLabel: (action, context) => `${action} message${context ? `: ${context}` : ''}`, sourceAriaLabel: 'Source of this response', derivativesAriaLabel: 'Responses derived from this one', scheduledTaskTriggered: 'Triggered by scheduled task', scheduledTaskTitle: (id) => `Triggered by scheduled task · ${id}`, legacyAutomationTriggered: 'Legacy Automation (history only)', legacyAutomationTitle: (id) => `Triggered by legacy Automation · ${id} · Historical only; it will not run again`, goalContinued: 'Continued by Goal', goalTitle: (id) => `Continued by Goal · ${id}`, agentGraphTriggered: 'Continued by Agent Graph', agentGraphTitle: (graphId) => `Triggered by the Agent Graph scheduler · ${graphId}`, thinkingTruncatedTitle: 'Some reasoning was truncated; showing the most recent content', outputTruncatedTitle: 'The assistant output exceeded the per-turn limit. Regenerate it or inspect the persisted task log for the complete content.', removeAttachmentAriaLabel: (name) => `Remove ${name}`, quoteLabel: 'Quote', quoteExpandAriaLabel: 'Show the full quoted excerpt', quoteCollapseAriaLabel: 'Collapse the quoted excerpt', removeQuoteAriaLabel: 'Remove quote', aborted: 'Interrupted', abortedByStop: 'Interrupted · Stop button',