vrr/qwen-code
2
0
Fork 0
mirror of https://github.com/QwenLM/qwen-code.git synced 2026-07-09 17:19:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

fix(serve): classify interrupted model stream errors (#6422)

Browse source 
* fix(serve): classify interrupted model streams

* fix(serve): address interrupted stream review

* test(webui): cover legacy terminated turn error fallback

* fix(web-shell): preserve error message data shape

* test(daemon): cover turn error fallback boundaries

* fix(web-shell): preserve classified error data

---------

Co-authored-by: ytahdn <ytahdn@gmail.com>
This commit is contained in:
ytahdn 2026-07-07 21:38:56 +08:00 • committed by GitHub
parent d1d53d7122
commit 40340ef505
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 495 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

60
packages/acp-bridge/src/bridge.test.ts
View file

@ -40,7 +40,11 @@ import {
WorkspaceMismatchError,
} from './bridgeErrors.js';
import { MAX_WORKSPACE_PATH_LENGTH } from './workspacePaths.js';
import { extractErrorMessage, extractErrorCode } from './bridge.js';
import {
classifyTurnErrorKind,
extractErrorMessage,
extractErrorCode,
} from './bridge.js';
import { SERVE_STATUS_EXT_METHODS } from './status.js';
import type { ChannelFactory } from './channel.js';
import type { BridgeTelemetry } from './bridgeOptions.js';
@ -3376,6 +3380,42 @@ describe('createAcpSessionBridge', () => {
await bridge.shutdown();
});
it('adds structured errorKind when a turn ends with terminated', async () => {
const handle = makeChannel({
promptImpl: () => {
throw new RequestError(-32603, 'terminated');
},
});
const bridge = makeBridge({ channelFactory: async () => handle.channel });
const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
const abort = new AbortController();
const iter = bridge.subscribeEvents(session.sessionId, {
signal: abort.signal,
});
const turnError = (async () => {
for await (const event of iter) {
if (event.type === 'turn_error') return event;
}
throw new Error('turn_error was not published');
})();
await expect(
bridge.sendPrompt(session.sessionId, {
sessionId: session.sessionId,
prompt: [{ type: 'text', text: 'stream may fail' }],
}),
).rejects.toThrow('terminated');
const event = await turnError;
expect(event.data).toMatchObject({
message: 'terminated',
errorKind: 'model_stream_interrupted',
});
abort.abort();
await bridge.shutdown();
});
it('echoes user_message_chunk to ALL session subscribers (cross-client sync)', async () => {
// Cross-client sync fix: a prompt sent by client A must be visible
// to every SSE subscriber of the same session — not just the
@ -10287,6 +10327,24 @@ describe('extractErrorCode', () => {
});
});
describe('classifyTurnErrorKind', () => {
it('classifies bare terminated errors as model stream interruptions', () => {
expect(classifyTurnErrorKind('terminated')).toBe(
'model_stream_interrupted',
);
expect(classifyTurnErrorKind('Terminated')).toBe(
'model_stream_interrupted',
);
expect(classifyTurnErrorKind(' terminated ')).toBe(
'model_stream_interrupted',
);
});
it('leaves unrelated turn errors unclassified', () => {
expect(classifyTurnErrorKind('API rate limit exceeded')).toBeUndefined();
});
});
// ---------------------------------------------------------------------------
// §2.3 side-channel state layer: publish helpers + reconciliation + snapshot
// ---------------------------------------------------------------------------

19
packages/acp-bridge/src/bridge.ts
View file

@ -912,6 +912,14 @@ export function extractErrorCode(err: unknown): string | undefined {
return undefined;
}
export function classifyTurnErrorKind(
message: string,
): 'model_stream_interrupted' | undefined {
return message.trim().toLowerCase() === 'terminated'
? 'model_stream_interrupted'
: undefined;
}
function broadcastTurnError(
entry: SessionEntry,
sessionId: string,
@ -921,6 +929,16 @@ function broadcastTurnError(
): void {
const message = extractErrorMessage(err);
const code = extractErrorCode(err);
const errorKind = classifyTurnErrorKind(message);
if (errorKind) {
writeServeDebugLine(
`turn_error classified session=${JSON.stringify(sessionId)} ` +
`message=${JSON.stringify(message)} ` +
`errorKind=${JSON.stringify(errorKind)}` +
(code ? ` code=${JSON.stringify(code)}` : '') +
(promptId ? ` promptId=${JSON.stringify(promptId)}` : ''),
);
}
entry.retryAllowed = true;
try {
entry.events.publish({
@ -929,6 +947,7 @@ function broadcastTurnError(
sessionId,
message,
...(code ? { code } : {}),
...(errorKind ? { errorKind } : {}),
...(promptId ? { promptId } : {}),
},
...(originatorClientId ? { originatorClientId } : {}),

1
packages/sdk-typescript/src/daemon/events.ts
View file

@ -757,6 +757,7 @@ export interface DaemonTurnErrorData {
sessionId: string;
message: string;
code?: string;
errorKind?: DaemonErrorKind | (string & {});
promptId?: string;
[key: string]: unknown;
}

2
packages/sdk-typescript/src/daemon/types.ts
View file

@ -730,6 +730,8 @@ export const DAEMON_ERROR_KINDS = [
// An SSE writer's last successful flush was older than the daemon's
// writer-idle deadline.
'writer_idle_timeout',
// The model response stream ended before a complete turn could be read.
'model_stream_interrupted',
] as const;
export type DaemonErrorKind = (typeof DAEMON_ERROR_KINDS)[number];

2
packages/sdk-typescript/src/daemon/ui/normalizer.ts
View file

@ -159,6 +159,7 @@ export function normalizeDaemonEvent(
}
case 'turn_error': {
const code = getString(event.data, 'code');
const errorKind = asDaemonErrorKind(getString(event.data, 'errorKind'));
const promptId = getString(event.data, 'promptId');
return [
{
@ -167,6 +168,7 @@ export function normalizeDaemonEvent(
source: 'turn_error',
recoverable: true,
...(code ? { code } : {}),
...(errorKind ? { errorKind } : {}),
...(promptId ? { promptId } : {}),
text:
getString(event.data, 'message') ??

3
packages/sdk-typescript/src/daemon/ui/transcript.ts
View file

@ -1016,6 +1016,9 @@ function appendStatusBlock(
...(event?.type === 'error' && event.promptId
? { promptId: event.promptId }
: {}),
...(event?.type === 'error' && event.errorKind
? { errorKind: event.errorKind }
: {}),
...(event?.type === 'error' && event.source
? { source: event.source }
: {}),

1
packages/sdk-typescript/src/daemon/ui/types.ts
View file

@ -852,6 +852,7 @@ export interface DaemonStatusTranscriptBlock extends DaemonTranscriptBlockBase {
text: string;
code?: string;
promptId?: string;
errorKind?: DaemonErrorKind;
source?: string;
data?: unknown;
}

97
packages/sdk-typescript/test/unit/daemonUi.test.ts
View file

@ -1196,6 +1196,103 @@ describe('daemon UI normalizer and transcript reducer', () => {
);
});
it('preserves structured model stream interruption turn errors', () => {
expect(
normalizeDaemonEvent({
id: 44,
v: 1,
type: 'turn_error',
data: {
sessionId: 'session-1',
message: 'terminated',
code: '-32603',
errorKind: 'model_stream_interrupted',
promptId: 'prompt-1',
},
}),
).toMatchObject([
{
type: 'error',
source: 'turn_error',
recoverable: true,
code: '-32603',
errorKind: 'model_stream_interrupted',
promptId: 'prompt-1',
text: 'terminated',
},
]);
});
it('keeps structured model stream interruption on transcript blocks', () => {
const state = reduceDaemonTranscriptEvents(
createDaemonTranscriptState({ now: 100 }),
normalizeDaemonEvent({
id: 46,
v: 1,
type: 'turn_error',
data: {
sessionId: 'session-1',
message: 'terminated',
code: '-32603',
errorKind: 'model_stream_interrupted',
promptId: 'prompt-1',
},
}),
{ now: 101 },
);
expect(state.blocks).toMatchObject([
{
kind: 'error',
source: 'turn_error',
text: 'terminated',
code: '-32603',
errorKind: 'model_stream_interrupted',
promptId: 'prompt-1',
},
]);
});
it('keeps older daemon terminated turn error text unchanged', () => {
expect(
normalizeDaemonEvent({
id: 45,
v: 1,
type: 'turn_error',
data: {
sessionId: 'session-1',
message: 'terminated',
},
}),
).toMatchObject([
{
type: 'error',
source: 'turn_error',
text: 'terminated',
},
]);
});
it('drops unrecognized errorKind from turn error events', () => {
const [event] = normalizeDaemonEvent({
id: 47,
v: 1,
type: 'turn_error',
data: {
sessionId: 'session-1',
message: 'some error',
errorKind: 'some_future_kind',
},
});
expect(event).toMatchObject({
type: 'error',
source: 'turn_error',
text: 'some error',
});
expect(event).not.toHaveProperty('errorKind');
});
it('normalizes daemon lifecycle and control events', () => {
expect(
normalizeDaemonEvent({

43
packages/web-shell/client/App.test.tsx
View file

@ -808,6 +808,49 @@ describe('App session callbacks', () => {
);
});
it('allows manual retry after a model stream interrupted turn error', async () => {
const { container, rerender } = renderApp();
await flush();
testState.prompt = 'recover this stream';
await clickSubmit(container);
expect(mockSessionActions.sendPrompt).toHaveBeenCalledWith(
'recover this stream',
expect.objectContaining({ retry: undefined }),
);
mockSessionActions.sendPrompt.mockClear();
act(() => {
testState.blocks = [
{
kind: 'error',
source: 'turn_error',
id: 'turn-error-stream-interrupted',
errorKind: 'model_stream_interrupted',
text: 'terminated',
},
];
rerender();
});
expect(container.querySelector('[data-testid="retry"]')).not.toBeNull();
await act(async () => {
container
.querySelector<HTMLButtonElement>('[data-testid="retry"]')
?.click();
await Promise.resolve();
});
expect(mockSessionActions.sendPrompt).toHaveBeenCalledWith(
'recover this stream',
expect.objectContaining({
optimisticUserMessage: false,
retry: true,
}),
);
});
it('gates queued submissions and only enqueues after approval', async () => {
let approve: (() => void) | undefined;
const onSubmitBefore = vi.fn(

125
packages/web-shell/client/adapters/transcriptToMessages.test.ts
View file

@ -2179,6 +2179,131 @@ describe('transcriptBlocksToDaemonMessages', () => {
]);
});
it('renders model stream interruption errors from structured errorKind labels', () => {
const messages = transcriptBlocksToDaemonMessages(
[
{
id: 'err-1',
kind: 'error' as const,
source: 'turn_error' as const,
errorKind: 'model_stream_interrupted' as const,
text: 'terminated',
data: { diagnosticId: 'abc' },
clientReceivedAt: 1,
createdAt: 1,
updatedAt: 1,
},
],
{
labels: {
modelStreamInterrupted: 'Localized stream interruption.',
},
},
);
expect(messages).toEqual([
{
id: 'err-1',
role: 'system',
content: 'Localized stream interruption.',
variant: 'error',
retryable: true,
source: 'turn_error',
data: {
diagnosticId: 'abc',
errorKind: 'model_stream_interrupted',
},
timestamp: 1,
},
]);
});
it('upgrades older daemon terminated turn errors to localized text', () => {
const messages = transcriptBlocksToDaemonMessages(
[
{
id: 'err-1',
kind: 'error' as const,
source: 'turn_error' as const,
text: 'terminated',
clientReceivedAt: 1,
createdAt: 1,
updatedAt: 1,
},
],
{
labels: {
modelStreamInterrupted: 'Localized stream interruption.',
},
},
);
expect(messages[0]).toMatchObject({
content: 'Localized stream interruption.',
retryable: true,
source: 'turn_error',
});
});
it('does not add data solely for structured errorKind labels', () => {
const messages = transcriptBlocksToDaemonMessages(
[
{
id: 'err-1',
kind: 'error' as const,
source: 'turn_error' as const,
errorKind: 'model_stream_interrupted' as const,
text: 'terminated',
clientReceivedAt: 1,
createdAt: 1,
updatedAt: 1,
},
],
{
labels: {
modelStreamInterrupted: 'Localized stream interruption.',
},
},
);
expect(messages[0]).toMatchObject({
content: 'Localized stream interruption.',
retryable: true,
source: 'turn_error',
});
expect(messages[0]).not.toHaveProperty('data');
});
it('preserves non-object error data when adding structured errorKind', () => {
const messages = transcriptBlocksToDaemonMessages(
[
{
id: 'err-1',
kind: 'error' as const,
source: 'turn_error' as const,
errorKind: 'model_stream_interrupted' as const,
text: 'terminated',
data: 'diagnostic text',
clientReceivedAt: 1,
createdAt: 1,
updatedAt: 1,
},
],
{
labels: {
modelStreamInterrupted: 'Localized stream interruption.',
},
},
);
expect(messages[0]).toMatchObject({
data: {
value: 'diagnostic text',
errorKind: 'model_stream_interrupted',
},
});
});
it('converts debug blocks to system messages with info variant', () => {
const messages = transcriptBlocksToDaemonMessages([
{

44
packages/web-shell/client/adapters/transcriptToMessages.ts
View file

@ -33,11 +33,6 @@ type DaemonPermissionTranscriptBlock = Extract<
{ kind: 'permission' }
>;
type ExtendedDaemonStatusTranscriptBlock = DaemonStatusTranscriptBlock & {
source?: string;
data?: unknown;
};
type ExtendedDaemonTextTranscriptBlock = DaemonTextTranscriptBlock & {
meta?: {
source?: unknown;
@ -50,6 +45,7 @@ interface TranscriptMessageLabels {
promptCancelled?: string;
branchSuccess?: (name: string) => string;
midTurnInserted?: (message: string) => string;
modelStreamInterrupted?: string;
}
interface TranscriptMessageOptions {
@ -63,6 +59,35 @@ function isIgnoredWebShellStatus(text: string): boolean {
);
}
function getErrorDisplayText(
block: DaemonStatusTranscriptBlock,
labels?: TranscriptMessageLabels,
): string {
if (
block.errorKind === 'model_stream_interrupted' ||
// Older daemons emit this turn_error before they know about errorKind.
(block.source === 'turn_error' &&
block.text.trim().toLowerCase() === 'terminated')
) {
return labels?.modelStreamInterrupted ?? block.text;
}
return block.text;
}
function getErrorMessageData(
data: unknown,
errorKind: DaemonStatusTranscriptBlock['errorKind'],
): { data?: unknown } {
if (data === undefined) return {};
if (!errorKind) return { data };
return {
data: {
...(getRecord(data) ?? { value: data }),
errorKind,
},
};
}
function getSessionBranchDisplayName(data: unknown): string | null {
if (!data || typeof data !== 'object') return null;
const branchData = data as {
@ -530,7 +555,7 @@ export function transcriptBlocksToDaemonMessages(
case 'status':
case 'debug': {
const statusBlock = block as ExtendedDaemonStatusTranscriptBlock;
const statusBlock = block;
const branchDisplayName =
statusBlock.source === 'session_branched'
? getSessionBranchDisplayName(statusBlock.data)
@ -575,16 +600,17 @@ export function transcriptBlocksToDaemonMessages(
}
case 'error': {
const errorBlock = block as ExtendedDaemonStatusTranscriptBlock;
const errorBlock = block;
const errorKind = errorBlock.errorKind;
messages.push({
id: block.id,
role: 'system',
content: errorBlock.text,
content: getErrorDisplayText(errorBlock, options.labels),
variant: 'error',
retryable: errorBlock.source === 'turn_error',
timestamp: blockTime,
...(errorBlock.source ? { source: errorBlock.source } : {}),
...(errorBlock.data !== undefined ? { data: errorBlock.data } : {}),
...getErrorMessageData(errorBlock.data, errorKind),
});
needsNewContentMessage = true;
break;

1
packages/web-shell/client/hooks/useMessages.ts
View file

@ -14,6 +14,7 @@ export function useMessages(
promptCancelled: t('request.cancelled'),
branchSuccess: (name) => t('branch.success', { name }),
midTurnInserted: (message) => t('midTurn.inserted', { message }),
modelStreamInterrupted: t('error.modelStreamInterrupted'),
},
}),
[blocks, t],

3
packages/web-shell/client/i18n.tsx
View file

@ -827,6 +827,8 @@ const EN: Messages = {
'bug.submitted': 'Bug report opened in a new tab.',
'clear.blocked': 'Cannot clear while streaming — cancel first (Esc).',
'error.unknown': 'Unknown error',
'error.modelStreamInterrupted':
'Model response stream was interrupted. Please retry.',
'shell.command': 'Shell Command',
'compact.enabled': 'Compact mode enabled',
'compact.disabled': 'Compact mode disabled',
@ -2439,6 +2441,7 @@ const ZH: Messages = {
'bug.submitted': 'Bug 报告已在新标签页中打开。',
'clear.blocked': '流式输出中无法清屏 — 先按 Esc 取消。',
'error.unknown': '未知错误',
'error.modelStreamInterrupted': '模型响应流已中断，请重试。',
'shell.command': 'Shell 命令',
'compact.enabled': '紧凑模式已开启',
'compact.disabled': '紧凑模式已关闭',

83
packages/webui/src/daemon/transcriptAdapter.test.ts
View file

@ -31,6 +31,89 @@ describe('daemonTranscriptToUnifiedMessages', () => {
});
});
it('renders model stream interruption turn errors with user-facing text', () => {
const [message] = daemonTranscriptToUnifiedMessages([
{
id: 'error-1',
kind: 'error',
source: 'turn_error',
errorKind: 'model_stream_interrupted',
text: 'terminated',
clientReceivedAt: 1,
createdAt: 1,
updatedAt: 1,
},
]);
expect(message).toMatchObject({
type: 'tool_call',
toolCall: {
kind: 'system_error',
rawOutput: 'Model response stream was interrupted. Please retry.',
content: [
{
content: {
text: 'Model response stream was interrupted. Please retry.',
error: 'Model response stream was interrupted. Please retry.',
},
},
],
},
});
});
it('renders older daemon terminated turn errors with user-facing text', () => {
const [message] = daemonTranscriptToUnifiedMessages([
{
id: 'error-legacy',
kind: 'error',
source: 'turn_error',
text: 'terminated',
clientReceivedAt: 1,
createdAt: 1,
updatedAt: 1,
},
]);
expect(message).toMatchObject({
type: 'tool_call',
toolCall: {
kind: 'system_error',
rawOutput: 'Model response stream was interrupted. Please retry.',
content: [
{
content: {
text: 'Model response stream was interrupted. Please retry.',
error: 'Model response stream was interrupted. Please retry.',
},
},
],
},
});
});
it('passes non-terminated turn error text through unchanged', () => {
const [message] = daemonTranscriptToUnifiedMessages([
{
id: 'error-other',
kind: 'error',
source: 'turn_error',
text: 'context deadline exceeded',
clientReceivedAt: 1,
createdAt: 1,
updatedAt: 1,
},
]);
expect(message).toMatchObject({
type: 'tool_call',
toolCall: {
kind: 'system_error',
rawOutput: 'context deadline exceeded',
},
});
});
it('maps daemon tool statuses without leaving terminal states spinning', () => {
const messages = daemonTranscriptToUnifiedMessages([
createToolBlock('cancelled-tool', 'cancelled'),

25
packages/webui/src/daemon/transcriptAdapter.ts
View file

@ -10,6 +10,7 @@ import {
isDaemonUiSensitiveKey,
sanitizeDaemonTerminalText,
type DaemonTranscriptBlock,
type DaemonStatusTranscriptBlock,
type DaemonToolTranscriptBlock,
} from '@qwen-code/sdk/daemon';
import type { UnifiedMessage } from '../adapters/types.js';
@ -43,6 +44,9 @@ export interface DaemonTranscriptAdapterOptions {
enrichToolDetailsWithPreview?: boolean;
}
const MODEL_STREAM_INTERRUPTED_MESSAGE =
'Model response stream was interrupted. Please retry.';
export function daemonTranscriptToUnifiedMessages(
blocks: readonly DaemonTranscriptBlock[],
options: DaemonTranscriptAdapterOptions = {},
@ -145,7 +149,8 @@ export function daemonTranscriptToUnifiedMessages(
isLast,
},
];
case 'error':
case 'error': {
const text = getErrorDisplayText(block);
return [
{
id: block.id,
@ -156,14 +161,14 @@ export function daemonTranscriptToUnifiedMessages(
kind: 'system_error',
title: 'System error',
status: 'failed',
rawOutput: sanitizeDisplayText(block.text),
rawOutput: sanitizeDisplayText(text),
content: [
{
type: 'content',
content: {
type: 'error',
text: sanitizeDisplayText(block.text),
error: sanitizeDisplayText(block.text),
text: sanitizeDisplayText(text),
error: sanitizeDisplayText(text),
},
},
],
@ -172,6 +177,7 @@ export function daemonTranscriptToUnifiedMessages(
isLast,
},
];
}
case 'status':
return [
{
@ -196,6 +202,17 @@ export function daemonTranscriptToUnifiedMessages(
});
}
function getErrorDisplayText(block: DaemonStatusTranscriptBlock): string {
if (
block.errorKind === 'model_stream_interrupted' ||
(block.source === 'turn_error' &&
block.text.trim().toLowerCase() === 'terminated')
) {
return MODEL_STREAM_INTERRUPTED_MESSAGE;
}
return block.text;
}
function daemonToolBlockToToolCallData(
block: DaemonToolTranscriptBlock,
enrichDetails: boolean = false,