diff --git a/docs/design/web-shell-loop-detection-turn-error.md b/docs/design/web-shell-loop-detection-turn-error.md new file mode 100644 index 00000000000..a7dc61d018f --- /dev/null +++ b/docs/design/web-shell-loop-detection-turn-error.md @@ -0,0 +1,21 @@ +# Web Shell loop-detection turn errors + +## Problem + +ACP loop protection currently records unstarted tool calls as failures and then completes the prompt with `stopReason: end_turn`. Web Shell therefore presents the internal tool skip text as the only explanation and treats the turn as successful. + +## Design + +When a foreground ACP prompt is stopped by loop protection, preserve completed and skipped tool results as today, then reject that prompt with a structured ACP request error. The bridge publishes the existing `turn_error` terminal with `errorKind: loop_detected` and the detector's `loopType`. Cancellation continues to take precedence when it races the loop stop. + +Web Shell renders `loop_detected` from the structured kind, using localized plain language: the model repeated tool use or reached a safety limit, only the current turn stopped, and the user can continue with a more specific instruction. No client matches the internal English tool error. + +Skipped tools keep their existing failed terminal update and error details so they cannot remain pending and their display behavior does not change. The additional `turn_error` provides the user-facing explanation for the stopped turn. + +The session remains alive and the per-turn loop state is recreated for the next prompt. Cron, background-notification, channel-classified, and goal turns keep their existing non-interactive handling: only interactive foreground prompts reject. Channel classification comes from the authenticated channel-prompt marker alone; the caller-requested delivery meta still schedules the delivery but keeps the foreground rejection, so it cannot opt a turn out of loop protection. Goal turns bypass the bridge entirely, so rejecting one would settle it as failed and pause the goal without publishing any `turn_error`; they resolve `end_turn` like the other automatic turn types. A loop-detected rejection still drains the cron/notification queues, preserving the invariant that a loop-stopped turn never strands queued automatic work. + +When Web Shell reloads a live session from paginated persisted history, the bridge appends the current in-memory `turn_error` to that replay. This keeps the terminal error visible across a page refresh while the session remains idle; newer turn content — including automatic turns the rejection itself drains — supersedes it by design. + +## Compatibility + +`turn_error` already terminates prompts and returns the UI to idle. Adding a known error kind and optional metadata is backward-compatible: older clients show the daemon message, while updated clients show localized guidance. diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index ca513fca62c..db18640da71 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -4518,6 +4518,1473 @@ describe('createAcpSessionBridge', () => { await bridge.shutdown(); }); + it('keeps the current turn error when refreshing from persisted history', async () => { + const handle = makeChannel({ + promptImpl: () => { + throw new RequestError(-32603, 'Loop protection stopped this turn', { + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }); + }, + extMethodImpl: (method, params) => { + if (method !== SERVE_STATUS_EXT_METHODS.sessionTranscript) { + throw new Error(`unexpected extMethod ${method}`); + } + return { + v: 1, + sessionId: params['sessionId'], + events: [ + { + v: 1, + type: 'session_update', + data: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text: 'persisted turn content' }, + _meta: { 'qwen.session.recordId': 'record-loop-page' }, + }, + }, + ], + hasMore: false, + }; + }, + }); + const bridge = makeBridge({ channelFactory: async () => handle.channel }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + await expect( + bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'loop' }], + }, + undefined, + { promptId: 'prompt-loop' }, + ), + ).rejects.toThrow('Loop protection stopped this turn'); + + const refreshed = await bridge.loadSession({ + sessionId: session.sessionId, + workspaceCwd: WS_A, + clientId: session.clientId, + historyReplay: 'response', + historyPageSize: 100, + }); + + const compactedReplay = refreshed.compactedReplay ?? []; + expect(compactedReplay).toHaveLength(2); + // Anchor the replay on the persisted page so the degenerate in-memory + // fallback (identical shape for a single-turn session) cannot satisfy + // the assertion. + expect(compactedReplay[0]).toMatchObject({ + type: 'session_update', + data: expect.objectContaining({ + content: { type: 'text', text: 'persisted turn content' }, + }), + }); + expect(compactedReplay[compactedReplay.length - 1]).toMatchObject({ + type: 'turn_error', + promptId: 'prompt-loop', + data: expect.objectContaining({ + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }), + }); + + await bridge.shutdown(); + }); + + it('keeps the turn error on refresh when queue bookkeeping lands after it', async () => { + // A queued-then-promoted prompt that trips loop protection publishes + // `pending_prompt_completed` AFTER its `turn_error` terminal (the + // queue-view bookkeeping from `result.finally`). That bookkeeping must + // not defeat the refresh-append of the terminal error. + let releaseFirst!: () => void; + const firstPrompt = new Promise<{ stopReason: 'end_turn' }>((resolve) => { + releaseFirst = () => resolve({ stopReason: 'end_turn' }); + }); + let promptCalls = 0; + const handle = makeChannel({ + promptImpl: () => { + promptCalls += 1; + if (promptCalls === 1) return firstPrompt; + throw new RequestError(-32603, 'Loop protection stopped this turn', { + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }); + }, + extMethodImpl: (method, params) => { + if (method !== SERVE_STATUS_EXT_METHODS.sessionTranscript) { + throw new Error(`unexpected extMethod ${method}`); + } + return { + v: 1, + sessionId: params['sessionId'], + events: [ + { + v: 1, + type: 'session_update', + data: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text: 'persisted turn content' }, + _meta: { 'qwen.session.recordId': 'record-loop-queued-page' }, + }, + }, + ], + hasMore: false, + }; + }, + }); + 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 seenTypes: string[] = []; + const collectUntilQueueBookkeeping = (async () => { + for await (const event of iter) { + seenTypes.push(event.type); + if ( + event.type === 'pending_prompt_completed' && + event.promptId === 'prompt-loop' + ) { + return; + } + } + throw new Error('pending_prompt_completed was not published'); + })(); + + const first = bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'first' }], + }, + undefined, + { promptId: 'prompt-first' }, + ); + await vi.waitFor(() => { + expect(handle.agent.promptCalls).toHaveLength(1); + }); + + const second = bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'loop' }], + }, + undefined, + { promptId: 'prompt-loop' }, + ); + await vi.waitFor(() => { + expect(bridge.getPendingPrompts(session.sessionId)).toHaveLength(2); + }); + + releaseFirst(); + await expect(first).resolves.toEqual({ stopReason: 'end_turn' }); + await expect(second).rejects.toThrow('Loop protection stopped this turn'); + await collectUntilQueueBookkeeping; + // The bookkeeping event lands AFTER the terminal — exactly the ordering + // that must not hide the error on refresh. + expect(seenTypes.indexOf('turn_error')).toBeGreaterThanOrEqual(0); + expect(seenTypes.indexOf('turn_error')).toBeLessThan( + seenTypes.lastIndexOf('pending_prompt_completed'), + ); + + const refreshed = await bridge.loadSession({ + sessionId: session.sessionId, + workspaceCwd: WS_A, + clientId: session.clientId, + historyReplay: 'response', + historyPageSize: 100, + }); + + const compactedReplay = refreshed.compactedReplay ?? []; + expect(compactedReplay).toHaveLength(2); + expect(compactedReplay[0]).toMatchObject({ + type: 'session_update', + data: expect.objectContaining({ + content: { type: 'text', text: 'persisted turn content' }, + }), + }); + expect(compactedReplay[compactedReplay.length - 1]).toMatchObject({ + type: 'turn_error', + promptId: 'prompt-loop', + data: expect.objectContaining({ + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }), + }); + + abort.abort(); + await bridge.shutdown(); + }); + + it.each([ + 'model_switch_failed', + 'language_changed', + 'session_metadata_updated', + 'session_cwd_changed', + 'artifact_changed', + 'settings_changed', + 'extensions_changed', + 'mcp_server_changed', + 'mcp_server_added', + 'mcp_server_removed', + 'approval_mode_changed', + 'model_switched', + 'prompt_cancelled', + 'tool_toggled', + 'workspace_initialized', + 'mcp_server_restarted', + 'mcp_server_restart_refused', + 'settings_reloaded', + 'trust_change_requested', + 'memory_changed', + 'agent_changed', + 'git_status_changed', + 'git_branch_changed', + 'github_setup_completed', + 'auth_device_flow_started', + 'auth_device_flow_throttled', + 'auth_device_flow_authorized', + 'auth_device_flow_failed', + 'auth_device_flow_cancelled', + ] as const)( + 'keeps the turn error on refresh when %s bookkeeping lands after it', + async (bookkeepingType) => { + // Idle-reachable bookkeeping (a rejected model switch, language + // change, rename, cwd change, client artifact) carries no turn + // content and must not defeat the refresh-append of the terminal. + const promptImpl = () => { + throw new RequestError(-32603, 'Loop protection stopped this turn', { + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }); + }; + const extMethodImpl = ( + method: string, + params: Record, + ): Record => { + if (method === SERVE_STATUS_EXT_METHODS.sessionTranscript) { + return { + v: 1, + sessionId: params['sessionId'], + events: [ + { + v: 1, + type: 'session_update', + data: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text: 'persisted turn content' }, + _meta: { + 'qwen.session.recordId': + 'record-loop-idle-bookkeeping-page', + }, + }, + }, + ], + hasMore: false, + }; + } + if (method === SERVE_CONTROL_EXT_METHODS.sessionLanguage) { + return { language: 'zh-CN', outputLanguage: null, refreshed: false }; + } + if (method === SERVE_CONTROL_EXT_METHODS.sessionCd) { + return { previousCwd: WS_A, newCwd: WS_B, warnings: [] }; + } + if (method === SERVE_CONTROL_EXT_METHODS.sessionApprovalMode) { + return { previous: 'default', current: ApprovalMode.PLAN }; + } + if (method === SERVE_CONTROL_EXT_METHODS.workspaceMcpRuntimeAdd) { + return { name: params['name'], toolCount: 1 }; + } + if (method === SERVE_CONTROL_EXT_METHODS.workspaceMcpRuntimeRemove) { + return { + name: params['name'], + removed: true, + wasShadowingSettings: false, + originatorClientId: '', + }; + } + if (method === SERVE_CONTROL_EXT_METHODS.workspaceMcpManage) { + return { + serverName: params['serverName'], + action: params['action'], + ok: true, + }; + } + return {}; + }; + let bridge: ReturnType; + const isModelSwitchCase = + bookkeepingType === 'model_switch_failed' || + bookkeepingType === 'model_switched' || + bookkeepingType === 'settings_changed'; + if (isModelSwitchCase) { + const factory: ChannelFactory = async () => { + const { clientStream, agentStream } = createInMemoryChannel(); + const fakeAgent = new FakeAgent({ promptImpl, extMethodImpl }); + const augmented = new Proxy(fakeAgent, { + get(target, prop) { + if (prop === 'unstable_setSessionModel') { + return async () => { + if (bookkeepingType === 'model_switch_failed') { + throw new Error('agent denied'); + } + return {}; + }; + } + // eslint-disable-next-line @typescript-eslint/no-explicit-any + return (target as any)[prop]; + }, + }); + new AgentSideConnection(() => augmented as Agent, agentStream); + return { + stream: clientStream, + exited: new Promise< + | { exitCode: number | null; signalCode: NodeJS.Signals | null } + | undefined + >(() => {}), + kill: async () => {}, + killSync: () => {}, + }; + }; + bridge = makeBridge({ channelFactory: factory }); + } else { + const handle = makeChannel({ promptImpl, extMethodImpl }); + bridge = makeBridge({ channelFactory: async () => handle.channel }); + } + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + await expect( + bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'loop' }], + }, + undefined, + { promptId: 'prompt-loop' }, + ), + ).rejects.toThrow('Loop protection stopped this turn'); + + const abort = new AbortController(); + const iter = bridge.subscribeEvents(session.sessionId, { + signal: abort.signal, + }); + const collectUntilBookkeeping = (async () => { + for await (const event of iter) { + if (event.type === bookkeepingType) return; + } + throw new Error(`${bookkeepingType} was not published`); + })(); + + if (bookkeepingType === 'model_switch_failed') { + await expect( + bridge.setSessionModel(session.sessionId, { + sessionId: session.sessionId, + modelId: 'rejected-model', + }), + ).rejects.toThrow(); + } else if ( + bookkeepingType === 'model_switched' || + bookkeepingType === 'settings_changed' + ) { + // The successful model-switch path publishes `model_switched` + // followed by a workspace-broadcast `settings_changed` with no + // `skipSessionId` — both land in every session's journal. + await bridge.setSessionModel(session.sessionId, { + sessionId: session.sessionId, + modelId: 'new-model', + }); + } else if (bookkeepingType === 'approval_mode_changed') { + await bridge.setSessionApprovalMode( + session.sessionId, + ApprovalMode.PLAN, + { persist: false }, + ); + } else if (bookkeepingType === 'extensions_changed') { + bridge.broadcastExtensionsChanged({ refreshed: 1, failed: 0 }); + } else if (bookkeepingType === 'mcp_server_added') { + await bridge.addRuntimeMcpServer('loop-mcp-server', { + command: 'loop-mcp', + }); + } else if (bookkeepingType === 'mcp_server_removed') { + await bridge.removeRuntimeMcpServer('loop-mcp-server'); + } else if (bookkeepingType === 'mcp_server_changed') { + await bridge.manageMcpServer('loop-mcp-server', 'enable', undefined); + } else if (bookkeepingType === 'prompt_cancelled') { + await bridge.cancelSession(session.sessionId); + } else if (bookkeepingType === 'language_changed') { + await bridge.setSessionLanguage(session.sessionId, { + language: 'zh-CN', + syncOutputLanguage: false, + }); + } else if (bookkeepingType === 'session_metadata_updated') { + await bridge.updateSessionMetadata(session.sessionId, { + displayName: 'Renamed after loop stop', + }); + } else if (bookkeepingType === 'session_cwd_changed') { + await bridge.changeSessionCwd(session.sessionId, { path: WS_B }); + } else if ( + bookkeepingType.startsWith('auth_device_flow_') || + bookkeepingType === 'tool_toggled' || + bookkeepingType === 'workspace_initialized' || + bookkeepingType === 'mcp_server_restarted' || + bookkeepingType === 'mcp_server_restart_refused' || + bookkeepingType === 'settings_reloaded' || + bookkeepingType === 'trust_change_requested' || + bookkeepingType === 'memory_changed' || + bookkeepingType === 'agent_changed' || + bookkeepingType === 'git_status_changed' || + bookkeepingType === 'git_branch_changed' || + bookkeepingType === 'github_setup_completed' + ) { + // The workspace service, git watcher, memory / agent CRUD, and the + // device-flow registry publish these through the workspace fan-out, + // not a session-scoped bridge method. + bridge.publishWorkspaceEvent({ + type: bookkeepingType, + data: { workspaceId: 'loop-bookkeeping-workspace' }, + }); + } else { + await bridge.addSessionArtifact( + session.sessionId, + { title: 'Client link', url: 'https://example.com/client' }, + { clientId: session.clientId }, + ); + } + await collectUntilBookkeeping; + + const refreshed = await bridge.loadSession({ + sessionId: session.sessionId, + workspaceCwd: WS_A, + clientId: session.clientId, + historyReplay: 'response', + historyPageSize: 100, + }); + + const compactedReplay = refreshed.compactedReplay ?? []; + expect(compactedReplay).toHaveLength(2); + // Anchor the replay on the persisted page: for a single-turn session + // the degenerate in-memory fallback has an identical shape, so only + // the persisted fixture's content proves the append path ran. + expect(compactedReplay[0]).toMatchObject({ + type: 'session_update', + data: expect.objectContaining({ + content: { type: 'text', text: 'persisted turn content' }, + }), + }); + expect(compactedReplay[compactedReplay.length - 1]).toMatchObject({ + type: 'turn_error', + promptId: 'prompt-loop', + data: expect.objectContaining({ + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }), + }); + + abort.abort(); + await bridge.shutdown(); + }, + ); + + it('keeps the turn error on refresh when an idle user-shell command streams output after it', async () => { + // User-shell activity publishes `user_shell_command`, `session_update` + // (shell output, `_meta.source: 'user-shell'`), and `user_shell_result` + // on the session bus while idle. The output is injected into the model + // conversation history, not the persisted transcript the refresh pages, + // so none of the three may defeat the append. + const shellSpy = vi + .spyOn(ShellExecutionService, 'execute') + .mockImplementation(async (_command, _cwd, onEvent) => { + onEvent({ type: 'data', chunk: 'hello\n' }); + return { + pid: 123, + result: Promise.resolve({ + rawOutput: Buffer.from('hello\n'), + output: 'hello\n', + exitCode: 0, + signal: null, + error: null, + aborted: false, + pid: 123, + executionMethod: 'none', + }), + }; + }); + try { + const handle = makeChannel({ + promptImpl: () => { + throw new RequestError(-32603, 'Loop protection stopped this turn', { + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }); + }, + extMethodImpl: (method, params) => { + if (method !== SERVE_STATUS_EXT_METHODS.sessionTranscript) { + return {}; + } + return { + v: 1, + sessionId: params['sessionId'], + events: [ + { + v: 1, + type: 'session_update', + data: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text: 'persisted turn content' }, + _meta: { 'qwen.session.recordId': 'record-loop-shell-page' }, + }, + }, + ], + hasMore: false, + }; + }, + }); + const bridge = makeBridge({ + sessionShellCommandEnabled: true, + channelFactory: async () => handle.channel, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + await expect( + bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'loop' }], + }, + undefined, + { promptId: 'prompt-loop' }, + ), + ).rejects.toThrow('Loop protection stopped this turn'); + + await bridge.executeShellCommand( + session.sessionId, + 'echo hello', + undefined, + { clientId: session.clientId }, + ); + + const refreshed = await bridge.loadSession({ + sessionId: session.sessionId, + workspaceCwd: WS_A, + clientId: session.clientId, + historyReplay: 'response', + historyPageSize: 100, + }); + + const compactedReplay = refreshed.compactedReplay ?? []; + expect(compactedReplay).toHaveLength(2); + // Anchor the replay on the persisted page so a regression of the + // bounded-append branch into the in-memory fallback cannot satisfy + // the assertions with an identically shaped replay. + expect(compactedReplay[0]).toMatchObject({ + type: 'session_update', + data: expect.objectContaining({ + content: { type: 'text', text: 'persisted turn content' }, + }), + }); + expect(compactedReplay[compactedReplay.length - 1]).toMatchObject({ + type: 'turn_error', + promptId: 'prompt-loop', + data: expect.objectContaining({ + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }), + }); + + await bridge.shutdown(); + } finally { + shellSpy.mockRestore(); + } + }); + + it('keeps the turn error on refresh when journal truncation marks the idle tail', async () => { + // With a pinned small journal cap, idle user-shell output evicts its + // own older entries after the loop terminal; `liveJournalSnapshot` + // then unshifts the synthetic `history_truncated` marker. The marker + // is size accounting, never ingested turn content — it must not + // defeat the append the way real newer content does. + const shellSpy = vi + .spyOn(ShellExecutionService, 'execute') + .mockImplementation(async (_command, _cwd, onEvent) => { + onEvent({ type: 'data', chunk: 'first chunk\n' }); + onEvent({ type: 'data', chunk: 'second chunk\n' }); + return { + pid: 123, + result: Promise.resolve({ + rawOutput: Buffer.from('first chunk\nsecond chunk\n'), + output: 'first chunk\nsecond chunk\n', + exitCode: 0, + signal: null, + error: null, + aborted: false, + pid: 123, + executionMethod: 'none', + }), + }; + }); + try { + const handle = makeChannel({ + promptImpl: () => { + throw new RequestError(-32603, 'Loop protection stopped this turn', { + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }); + }, + extMethodImpl: (method, params) => { + if (method !== SERVE_STATUS_EXT_METHODS.sessionTranscript) { + return {}; + } + return { + v: 1, + sessionId: params['sessionId'], + events: [ + { + v: 1, + type: 'session_update', + data: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text: 'persisted turn content' }, + _meta: { + 'qwen.session.recordId': + 'record-loop-truncated-journal-page', + }, + }, + }, + ], + hasMore: false, + }; + }, + }); + const bridge = makeBridge({ + sessionShellCommandEnabled: true, + maxJournalEvents: 1, + channelFactory: async () => handle.channel, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + await expect( + bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'loop' }], + }, + undefined, + { promptId: 'prompt-loop' }, + ), + ).rejects.toThrow('Loop protection stopped this turn'); + + await bridge.executeShellCommand( + session.sessionId, + 'echo truncated', + undefined, + { clientId: session.clientId }, + ); + + const refreshed = await bridge.loadSession({ + sessionId: session.sessionId, + workspaceCwd: WS_A, + clientId: session.clientId, + historyReplay: 'response', + historyPageSize: 100, + }); + + const compactedReplay = refreshed.compactedReplay ?? []; + expect(compactedReplay).toHaveLength(2); + expect(compactedReplay[0]).toMatchObject({ + type: 'session_update', + data: expect.objectContaining({ + content: { type: 'text', text: 'persisted turn content' }, + }), + }); + expect(compactedReplay[compactedReplay.length - 1]).toMatchObject({ + type: 'turn_error', + promptId: 'prompt-loop', + data: expect.objectContaining({ + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }), + }); + + await bridge.shutdown(); + } finally { + shellSpy.mockRestore(); + } + }); + + it.each(['available_commands_update', 'current_mode_update'] as const)( + 'keeps the turn error on refresh when an idle %s session_update lands after it', + async (subtype) => { + // Latest-wins state snapshots fan out to idle sessions (a workspace + // skills/settings refresh, an approval-mode change). They carry no + // turn content for the persisted transcript and must not defeat the + // refresh-append of the terminal. + const handle = makeChannel({ + promptImpl: () => { + throw new RequestError(-32603, 'Loop protection stopped this turn', { + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }); + }, + extMethodImpl: (method, params) => { + if (method !== SERVE_STATUS_EXT_METHODS.sessionTranscript) { + return {}; + } + return { + v: 1, + sessionId: params['sessionId'], + events: [ + { + v: 1, + type: 'session_update', + data: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text: 'persisted turn content' }, + _meta: { + 'qwen.session.recordId': `record-loop-${subtype}-page`, + }, + }, + }, + ], + hasMore: false, + }; + }, + }); + const bridge = makeBridge({ channelFactory: async () => handle.channel }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + await expect( + bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'loop' }], + }, + undefined, + { promptId: 'prompt-loop' }, + ), + ).rejects.toThrow('Loop protection stopped this turn'); + + const abort = new AbortController(); + const iter = bridge.subscribeEvents(session.sessionId, { + signal: abort.signal, + }); + const sawIdleUpdate = (async () => { + for await (const event of iter) { + if ( + event.type === 'session_update' && + (event.data as { update?: { sessionUpdate?: string } })?.update + ?.sessionUpdate === subtype + ) { + return; + } + } + throw new Error(`${subtype} was not published`); + })(); + await handle.agentConnection.sessionUpdate({ + sessionId: session.sessionId, + update: + subtype === 'available_commands_update' + ? { + sessionUpdate: subtype, + availableCommands: [], + _meta: { availableSkills: [] }, + } + : { sessionUpdate: subtype, currentModeId: 'plan' }, + }); + await sawIdleUpdate; + + const refreshed = await bridge.loadSession({ + sessionId: session.sessionId, + workspaceCwd: WS_A, + clientId: session.clientId, + historyReplay: 'response', + historyPageSize: 100, + }); + + const compactedReplay = refreshed.compactedReplay ?? []; + expect(compactedReplay).toHaveLength(2); + expect(compactedReplay[0]).toMatchObject({ + type: 'session_update', + data: expect.objectContaining({ + content: { type: 'text', text: 'persisted turn content' }, + }), + }); + expect(compactedReplay[compactedReplay.length - 1]).toMatchObject({ + type: 'turn_error', + promptId: 'prompt-loop', + data: expect.objectContaining({ + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }), + }); + + abort.abort(); + await bridge.shutdown(); + }, + ); + + it('keeps the turn error on refresh when a queued deadline terminal lands after it', async () => { + // A queued prompt's terminal publishes the event alone without mutating + // turn state; it must not erase the refresh-replay record of the active + // turn's failure either. The held cancel ack keeps the queued prompt's + // FIFO promotion blocked on the cancel-forward drain, so its deadline + // expires while it is still queued AFTER the loop terminal has landed. + let promptCalls = 0; + const heldTurn = deferred(); + let releaseCancel!: () => void; + const heldCancel = new Promise((resolve) => { + releaseCancel = resolve; + }); + const handle = makeChannel({ + promptImpl: () => { + promptCalls += 1; + if (promptCalls === 1) return heldTurn.promise; + return { stopReason: 'end_turn' }; + }, + cancelImpl: () => heldCancel, + extMethodImpl: (method, params) => { + if (method !== SERVE_STATUS_EXT_METHODS.sessionTranscript) { + return {}; + } + return { + v: 1, + sessionId: params['sessionId'], + events: [ + { + v: 1, + type: 'session_update', + data: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text: 'persisted turn content' }, + _meta: { + 'qwen.session.recordId': 'record-loop-queued-deadline-page', + }, + }, + }, + ], + hasMore: false, + }; + }, + }); + 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 terminalOrder: string[] = []; + const sawQueuedDeadlineTerminal = (async () => { + for await (const event of iter) { + if (event.type !== 'turn_error') continue; + if (event.promptId === 'prompt-first') terminalOrder.push('loop'); + if ( + event.promptId === 'prompt-queued' && + (event.data as { code?: string }).code === 'prompt_deadline_exceeded' + ) { + terminalOrder.push('deadline'); + return; + } + } + throw new Error('queued deadline terminal was not published'); + })(); + + const first = bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'first' }], + }, + undefined, + { promptId: 'prompt-first' }, + ); + first.catch(() => {}); + await vi.waitFor(() => { + expect(handle.agent.promptCalls).toHaveLength(1); + }); + + const second = bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'queued' }], + }, + undefined, + { promptId: 'prompt-queued', deadlineMs: 120 }, + ); + second.catch(() => {}); + await vi.waitFor(() => { + expect(bridge.getPendingPrompts(session.sessionId)).toHaveLength(2); + }); + + // Cancel the running turn, then reject it with the loop error. The held + // cancel ack blocks the queued prompt's promotion on the cancel-forward + // drain, so it is still queued when the deadline expires. + void bridge.cancelSession(session.sessionId).catch(() => {}); + await vi.waitFor(() => { + expect(handle.agent.cancelCalls).toHaveLength(1); + }); + heldTurn.reject( + new RequestError(-32603, 'Loop protection stopped this turn', { + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }), + ); + await expect(first).rejects.toThrow('Loop protection stopped this turn'); + + await sawQueuedDeadlineTerminal; + // Pin the scenario's premise — the deadline expires while the prompt is + // still queued, AFTER the loop terminal has landed. The 120 ms budget + // can invert under load; without this the assertions below would stay + // green through the inverted ordering. + expect(terminalOrder.indexOf('loop')).toBeGreaterThanOrEqual(0); + expect(terminalOrder.indexOf('loop')).toBeLessThan( + terminalOrder.lastIndexOf('deadline'), + ); + + const refreshed = await bridge.loadSession({ + sessionId: session.sessionId, + workspaceCwd: WS_A, + clientId: session.clientId, + historyReplay: 'response', + historyPageSize: 100, + }); + + const compactedReplay = refreshed.compactedReplay ?? []; + // Anchor the replay on the persisted page so a regression of the + // bounded-append branch into the in-memory fallback cannot satisfy + // the assertions with an identically shaped replay. + expect(compactedReplay[0]).toMatchObject({ + type: 'session_update', + data: expect.objectContaining({ + content: { type: 'text', text: 'persisted turn content' }, + }), + }); + expect( + compactedReplay.some( + (event) => + event.type === 'turn_error' && + (event.data as { errorKind?: string }).errorKind === 'loop_detected', + ), + ).toBe(true); + // The queued terminal must not overwrite the session-scoped summary + // either: it belongs to a prompt that never ran. + expect(bridge.getSessionSummary(session.sessionId).turnError).toMatchObject( + { code: 'LOOP_DETECTED' }, + ); + + abort.abort(); + releaseCancel(); + await bridge.shutdown(); + }); + + it('keeps the turn error on refresh when a queued prompt is removed after it', async () => { + // DAEMON-004 variant: removing a still-queued prompt publishes its + // `cancelled` terminal alone. Like the deadline variant it must not + // clear the active turn's refresh-replay record. + let promptCalls = 0; + const heldTurn = deferred(); + let releaseCancel!: () => void; + const heldCancel = new Promise((resolve) => { + releaseCancel = resolve; + }); + const handle = makeChannel({ + promptImpl: () => { + promptCalls += 1; + if (promptCalls === 1) return heldTurn.promise; + return { stopReason: 'end_turn' }; + }, + cancelImpl: () => heldCancel, + extMethodImpl: (method, params) => { + if (method !== SERVE_STATUS_EXT_METHODS.sessionTranscript) { + return {}; + } + return { + v: 1, + sessionId: params['sessionId'], + events: [ + { + v: 1, + type: 'session_update', + data: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text: 'persisted turn content' }, + _meta: { + 'qwen.session.recordId': 'record-loop-queued-removed-page', + }, + }, + }, + ], + hasMore: false, + }; + }, + }); + 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 sawQueuedRemovedTerminal = (async () => { + for await (const event of iter) { + if ( + event.type === 'turn_complete' && + event.promptId === 'prompt-queued' && + (event.data as { stopReason?: string }).stopReason === 'cancelled' + ) { + return; + } + } + throw new Error('queued removed terminal was not published'); + })(); + + const first = bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'first' }], + }, + undefined, + { promptId: 'prompt-first' }, + ); + first.catch(() => {}); + await vi.waitFor(() => { + expect(handle.agent.promptCalls).toHaveLength(1); + }); + + const second = bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'queued' }], + }, + undefined, + { promptId: 'prompt-queued' }, + ); + second.catch(() => {}); + await vi.waitFor(() => { + expect(bridge.getPendingPrompts(session.sessionId)).toHaveLength(2); + }); + + void bridge.cancelSession(session.sessionId).catch(() => {}); + await vi.waitFor(() => { + expect(handle.agent.cancelCalls).toHaveLength(1); + }); + heldTurn.reject( + new RequestError(-32603, 'Loop protection stopped this turn', { + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }), + ); + await expect(first).rejects.toThrow('Loop protection stopped this turn'); + + // The queued prompt is removed while its promotion is still blocked on + // the cancel-forward drain. + expect( + bridge.removePendingPrompt(session.sessionId, 'prompt-queued'), + ).toEqual({ removed: true }); + await sawQueuedRemovedTerminal; + + const refreshed = await bridge.loadSession({ + sessionId: session.sessionId, + workspaceCwd: WS_A, + clientId: session.clientId, + historyReplay: 'response', + historyPageSize: 100, + }); + + const compactedReplay = refreshed.compactedReplay ?? []; + // Anchor the replay on the persisted page so a regression of the + // bounded-append branch into the in-memory fallback cannot satisfy + // the assertions with an identically shaped replay. + expect(compactedReplay[0]).toMatchObject({ + type: 'session_update', + data: expect.objectContaining({ + content: { type: 'text', text: 'persisted turn content' }, + }), + }); + expect( + compactedReplay.some( + (event) => + event.type === 'turn_error' && + (event.data as { errorKind?: string }).errorKind === 'loop_detected', + ), + ).toBe(true); + expect(bridge.getSessionSummary(session.sessionId).turnError).toMatchObject( + { code: 'LOOP_DETECTED' }, + ); + + abort.abort(); + releaseCancel(); + await bridge.shutdown(); + }); + + it('drops the stale turn error when a queued terminal folds newer automatic-turn content', async () => { + // A queued terminal is a turn boundary on the bus: ingesting it folds + // and resets the live journal. When newer turn content was journaled + // after the pending loop terminal, that content supersedes the stale + // error before the fold erases it — otherwise the refresh-append would + // re-place the stale loop error AFTER the newer automatic content, the + // exact misplacement the guard exists to prevent. + let promptCalls = 0; + const heldTurn = deferred(); + let releaseCancel!: () => void; + const heldCancel = new Promise((resolve) => { + releaseCancel = resolve; + }); + const handle = makeChannel({ + promptImpl: () => { + promptCalls += 1; + if (promptCalls === 1) return heldTurn.promise; + return { stopReason: 'end_turn' }; + }, + cancelImpl: () => heldCancel, + extMethodImpl: (method, params) => { + if (method !== SERVE_STATUS_EXT_METHODS.sessionTranscript) { + return {}; + } + return { + v: 1, + sessionId: params['sessionId'], + events: [ + { + v: 1, + type: 'session_update', + data: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text: 'persisted turn content' }, + _meta: { + 'qwen.session.recordId': 'record-loop-queued-supersede-page', + }, + }, + }, + ], + hasMore: false, + }; + }, + }); + 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, + }); + // One consumer for both waypoints: two loops over the same iterator + // would race for each event and starve one of the waits. + const sawContentThenQueuedTerminal = (async () => { + let sawContent = false; + for await (const event of iter) { + if ( + !sawContent && + event.type === 'session_update' && + JSON.stringify(event.data).includes('automatic turn content') + ) { + sawContent = true; + continue; + } + if ( + sawContent && + event.type === 'turn_complete' && + event.promptId === 'prompt-queued' && + (event.data as { stopReason?: string }).stopReason === 'cancelled' + ) { + return; + } + } + throw new Error('automatic content or queued terminal not published'); + })(); + + const first = bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'first' }], + }, + undefined, + { promptId: 'prompt-first' }, + ); + first.catch(() => {}); + await vi.waitFor(() => { + expect(handle.agent.promptCalls).toHaveLength(1); + }); + + const second = bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'queued' }], + }, + undefined, + { promptId: 'prompt-queued' }, + ); + second.catch(() => {}); + await vi.waitFor(() => { + expect(bridge.getPendingPrompts(session.sessionId)).toHaveLength(2); + }); + + void bridge.cancelSession(session.sessionId).catch(() => {}); + await vi.waitFor(() => { + expect(handle.agent.cancelCalls).toHaveLength(1); + }); + heldTurn.reject( + new RequestError(-32603, 'Loop protection stopped this turn', { + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }), + ); + await expect(first).rejects.toThrow('Loop protection stopped this turn'); + + // An automatic turn journals content after the loop terminal; the + // queued terminal that follows folds it. + await handle.agentConnection.sessionUpdate({ + sessionId: session.sessionId, + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: 'automatic turn content' }, + }, + }); + + expect( + bridge.removePendingPrompt(session.sessionId, 'prompt-queued'), + ).toEqual({ removed: true }); + await sawContentThenQueuedTerminal; + + const refreshed = await bridge.loadSession({ + sessionId: session.sessionId, + workspaceCwd: WS_A, + clientId: session.clientId, + historyReplay: 'response', + historyPageSize: 100, + }); + + const compactedReplay = refreshed.compactedReplay ?? []; + expect(compactedReplay[0]).toMatchObject({ + type: 'session_update', + data: expect.objectContaining({ + content: { type: 'text', text: 'persisted turn content' }, + }), + }); + expect( + compactedReplay.some( + (event) => + event.type === 'turn_error' && + (event.data as { errorKind?: string }).errorKind === 'loop_detected', + ), + ).toBe(false); + // The supersede only drops the refresh-replay record; the session + // summary still carries the active turn's failure. + expect(bridge.getSessionSummary(session.sessionId).turnError).toMatchObject( + { code: 'LOOP_DETECTED' }, + ); + + abort.abort(); + releaseCancel(); + await bridge.shutdown(); + }); + + it('drops the turn error on refresh after a subsequent successful interactive turn', async () => { + // Loop reject, then a successful prompt, then refresh: the newer turn's + // terminal supersedes the pending append, so the stale loop error must + // not reappear after the newer turn's content. + let promptCalls = 0; + const handle = makeChannel({ + promptImpl: () => { + promptCalls += 1; + if (promptCalls === 1) { + throw new RequestError(-32603, 'Loop protection stopped this turn', { + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }); + } + return { stopReason: 'end_turn' }; + }, + extMethodImpl: (method, params) => { + if (method !== SERVE_STATUS_EXT_METHODS.sessionTranscript) { + return {}; + } + return { + v: 1, + sessionId: params['sessionId'], + events: [ + { + v: 1, + type: 'session_update', + data: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text: 'persisted turn content' }, + _meta: { 'qwen.session.recordId': 'record-loop-recovery-page' }, + }, + }, + ], + hasMore: false, + }; + }, + }); + const bridge = makeBridge({ channelFactory: async () => handle.channel }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + await expect( + bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'loop' }], + }, + undefined, + { promptId: 'prompt-loop' }, + ), + ).rejects.toThrow('Loop protection stopped this turn'); + + await expect( + bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'recovery' }], + }, + undefined, + { promptId: 'prompt-recovery' }, + ), + ).resolves.toEqual({ stopReason: 'end_turn' }); + + const refreshed = await bridge.loadSession({ + sessionId: session.sessionId, + workspaceCwd: WS_A, + clientId: session.clientId, + historyReplay: 'response', + historyPageSize: 100, + }); + + const compactedReplay = refreshed.compactedReplay ?? []; + expect(compactedReplay.some((event) => event.type === 'turn_error')).toBe( + false, + ); + + await bridge.shutdown(); + }); + + it('drops the stale turn error on refresh after newer automatic-turn content', async () => { + const handle = makeChannel({ + promptImpl: () => { + throw new RequestError(-32603, 'Loop protection stopped this turn', { + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }); + }, + extMethodImpl: (method, params) => { + if (method !== SERVE_STATUS_EXT_METHODS.sessionTranscript) { + throw new Error(`unexpected extMethod ${method}`); + } + return { + v: 1, + sessionId: params['sessionId'], + events: [ + { + v: 1, + type: 'session_update', + data: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text: 'persisted turn content' }, + _meta: { 'qwen.session.recordId': 'record-loop-stale-page' }, + }, + }, + ], + hasMore: false, + }; + }, + }); + const bridge = makeBridge({ channelFactory: async () => handle.channel }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + await expect( + bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'loop' }], + }, + undefined, + { promptId: 'prompt-loop' }, + ), + ).rejects.toThrow('Loop protection stopped this turn'); + + // An automatic turn (cron/background notification) runs after the loop + // error without an interactive dispatch: its content is journaled via + // the ordinary session/update fan-in. + const abort = new AbortController(); + const iter = bridge.subscribeEvents(session.sessionId, { + signal: abort.signal, + }); + const sawAutomaticContent = (async () => { + for await (const event of iter) { + if ( + event.type === 'session_update' && + JSON.stringify(event.data).includes('automatic turn content') + ) { + return; + } + } + throw new Error('automatic turn content was not published'); + })(); + await handle.agentConnection.sessionUpdate({ + sessionId: session.sessionId, + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: 'automatic turn content' }, + }, + }); + await sawAutomaticContent; + + const refreshed = await bridge.loadSession({ + sessionId: session.sessionId, + workspaceCwd: WS_A, + clientId: session.clientId, + historyReplay: 'response', + historyPageSize: 100, + }); + + const compactedReplay = refreshed.compactedReplay ?? []; + // The retained event must BE the persisted-page event — a stale branch + // that rebuilds the replay from anything else ships green without this. + expect(compactedReplay[0]).toMatchObject({ + type: 'session_update', + data: expect.objectContaining({ + content: { type: 'text', text: 'persisted turn content' }, + }), + }); + expect(compactedReplay).toHaveLength(1); + expect(compactedReplay.some((event) => event.type === 'turn_error')).toBe( + false, + ); + + abort.abort(); + await bridge.shutdown(); + }); + it('skips empty persisted pages when refreshing an attached load', async () => { const handle = makeChannel({ loadSessionImpl: () => ({}), @@ -10640,6 +12107,45 @@ describe('createAcpSessionBridge', () => { await bridge.shutdown(); }); + it('strips spoofed channel-prompt classification and injects only trusted context', async () => { + // `qwen.channel.prompt` opts a turn out of loop-detected rejection, + // so a forged key must not reach the child; only the authenticated + // channel-worker flag on the trusted context re-arms it. + const handle = makeChannel(); + const bridge = makeBridge({ channelFactory: async () => handle.channel }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + await bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'forged channel turn' }], + _meta: { 'qwen.channel.prompt': true }, + } as PromptRequest, + undefined, + { promptId: 'prompt-forged' }, + ); + expect( + handle.agent.promptCalls[0]?._meta?.['qwen.channel.prompt'], + ).toBeUndefined(); + + await bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'trusted channel turn' }], + _meta: { 'qwen.channel.prompt': true }, + } as PromptRequest, + undefined, + { promptId: 'prompt-trusted', channelPrompt: true }, + ); + expect(handle.agent.promptCalls[1]?._meta?.['qwen.channel.prompt']).toBe( + true, + ); + + await bridge.shutdown(); + }); + it('strips both spoofed retry and continue meta keys from one prompt', async () => { const handle = makeChannel(); const bridge = makeBridge({ channelFactory: async () => handle.channel }); @@ -10961,6 +12467,66 @@ describe('createAcpSessionBridge', () => { await bridge.shutdown(); }); + it('preserves structured loop detection details on turn_error', async () => { + const handle = makeChannel({ + promptImpl: () => { + throw new RequestError(-32603, 'Loop protection stopped this turn', { + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + }); + }, + }); + 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 emittedTypes: string[] = []; + const turnError = (async () => { + for await (const event of iter) { + emittedTypes.push(event.type); + 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: 'loop' }], + }, + undefined, + { promptId: 'prompt-loop' }, + ), + ).rejects.toThrow('Loop protection stopped this turn'); + + await expect(turnError).resolves.toMatchObject({ + type: 'turn_error', + promptId: 'prompt-loop', + data: { + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: 'turn_tool_call_cap', + promptId: 'prompt-loop', + }, + }); + // Structured rejections already ran on the daemon: the forward-failure + // phantom (`prompt_cancelled{forward_failed}`) must be suppressed. + expect(emittedTypes).not.toContain('prompt_cancelled'); + expect(bridge.getSessionSummary(session.sessionId).turnError).toEqual({ + message: 'Loop protection stopped this turn', + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + }); + + 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 diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index abbd6b6b181..81319c11a3c 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -130,6 +130,7 @@ import { type ActiveWorkHeartbeatCapabilityV1, type ActiveWorkHoldCategory, type ActiveWorkSnapshotV1, + CHANNEL_PROMPT_META_KEY, CHANNEL_STARTUP_PROFILE_META_KEY, CHANNEL_STARTUP_PROFILE_VERSION, DAEMON_CHANNEL_DELIVERY_META_KEY, @@ -1086,6 +1087,15 @@ interface SessionEntry { code?: string; errorKind?: string; }; + /** + * The journaled `turn_error` event behind `turnError`, when the failed + * turn published one. A bounded refresh replays it onto persisted + * history so the terminal survives a page refresh; any newer turn + * terminal — or newer turn content about to be folded by a queued + * terminal boundary — clears it so a stale error is never re-appended + * after newer content. Not part of the session summary. + */ + turnErrorEvent?: BridgeEvent; retryAllowed: boolean; /** Prompt id whose `prompt_cancelled` event has already been broadcast. */ cancelBroadcastPromptId?: string; @@ -1531,9 +1541,10 @@ function broadcastTurnComplete( promptResult: { stopReason?: string; [k: string]: unknown }, promptId: string | undefined, originatorClientId: string | undefined, + mutateTurnState: boolean, ): void { try { - entry.events.publish({ + const published = entry.events.publish({ type: 'turn_complete', ...(promptId ? { promptId } : {}), data: { @@ -1543,6 +1554,13 @@ function broadcastTurnComplete( }, ...(originatorClientId ? { originatorClientId } : {}), }); + // A newer turn terminal supersedes any pending refresh-append error — + // but only for a prompt that actually ran. A queued prompt's terminal + // (deadline expiry, queued removal) publishes the event alone without + // mutating turn state, so it must not erase the refresh-replay record + // of the active turn's failure either. + if (mutateTurnState && published !== undefined) + entry.turnErrorEvent = undefined; } catch { /* bus may be closed during session teardown */ } @@ -1584,6 +1602,17 @@ function extractJsonRpcErrorDetail(data: unknown): string | undefined { return undefined; } +function extractJsonRpcErrorField( + err: unknown, + field: string, +): string | undefined { + if (typeof err !== 'object' || err === null) return undefined; + const data = (err as { data?: unknown }).data; + if (typeof data !== 'object' || data === null) return undefined; + const value = (data as Record)[field]; + return typeof value === 'string' && value.length > 0 ? value : undefined; +} + export function extractErrorCode(err: unknown): string | undefined { if (typeof err !== 'object' || err === null || !('code' in err)) return undefined; @@ -1593,6 +1622,111 @@ export function extractErrorCode(err: unknown): string | undefined { return undefined; } +/** + * Event types that may be published after a turn terminal without adding + * turn content (prompt-queue bookkeeping, config changes, and other + * idle-reachable session bookkeeping). The bounded refresh-append guard + * skips these when deciding whether the in-memory `turn_error` is still + * the newest meaningful terminal; any other event type blocks the append. + * `pending_prompt_started` is deliberately absent: it is published before + * admission clears `turnErrorEvent`, so blocking the append in that window + * keeps a stale error from trailing a turn that is already starting. + * + * Audit the full `broadcastWorkspaceEvent` vocabulary (and any other + * idle-reachable session-bus publish) before adding a new event type to + * the bus: every idle non-turn event belongs here, or an otherwise-idle + * activity — a model switch, an extension refresh, an MCP server change, + * a user-shell command — defeats the append and the loop terminal + * disappears from the refreshed transcript. `session_update` events are + * turn content except the idle bookkeeping subtypes skipped via + * `isIdleBookkeepingSessionUpdate`: the user-shell output stream (its + * history goes to the model conversation, not the persisted transcript + * the refresh pages) and the latest-wins state snapshots + * (`available_commands_update`, `current_mode_update`) that settings and + * approval-mode refreshes fan out to idle sessions. + */ +const REFRESH_APPEND_BOOKKEEPING_EVENT_TYPES = new Set([ + 'pending_prompt_added', + 'pending_prompt_completed', + 'prompt_cancelled', + 'model_switched', + 'model_switch_failed', + 'approval_mode_changed', + 'language_changed', + 'session_metadata_updated', + 'session_cwd_changed', + 'artifact_changed', + 'settings_changed', + 'extensions_changed', + 'mcp_server_changed', + 'mcp_server_added', + 'mcp_server_removed', + 'user_shell_command', + 'user_shell_result', + // Workspace-level fan-out (workspace service, git watcher, memory / + // agent CRUD, device-flow registry) reaches every session bus via + // `publishWorkspaceEvent` while idle. The `auth_device_flow_*` members + // mirror the closed `DeviceFlowEventEmission` union — audit that union + // when it grows. + 'tool_toggled', + 'workspace_initialized', + 'mcp_server_restarted', + 'mcp_server_restart_refused', + 'settings_reloaded', + 'trust_change_requested', + 'memory_changed', + 'agent_changed', + 'git_status_changed', + 'git_branch_changed', + 'github_setup_completed', + 'auth_device_flow_started', + 'auth_device_flow_throttled', + 'auth_device_flow_authorized', + 'auth_device_flow_failed', + 'auth_device_flow_cancelled', +]); + +/** + * `session_update` frames published while idle that carry no turn content + * for the persisted transcript, so they must not defeat the refresh-append + * of a pending terminal error the way a real turn's `session_update` does: + * the user-shell output stream (injected into the model conversation + * history instead of the transcript the refresh pages) and the + * latest-wins state snapshots (`available_commands_update` from a + * skills/settings refresh, the legacy dual-emit `current_mode_update`). + */ +function isIdleBookkeepingSessionUpdate(event: BridgeEvent): boolean { + if (event.type !== 'session_update') return false; + const data = event.data; + if (!data || typeof data !== 'object' || Array.isArray(data)) return false; + const update = (data as Record)['update']; + if (!update || typeof update !== 'object' || Array.isArray(update)) + return false; + const updateRecord = update as Record; + const subtype = updateRecord['sessionUpdate']; + if ( + subtype === 'available_commands_update' || + subtype === 'current_mode_update' + ) { + return true; + } + const meta = updateRecord['_meta']; + if (!meta || typeof meta !== 'object' || Array.isArray(meta)) return false; + return (meta as Record)['source'] === 'user-shell'; +} + +/** + * Turn-content test behind the refresh-append guard: everything that is + * neither idle bookkeeping nor the synthetic journal-truncation marker + * (which `liveJournalSnapshot` unshifts without ever ingesting it as + * content) counts as newer turn content. + */ +function isRefreshAppendTurnContent(event: BridgeEvent): boolean { + if (event.type === 'history_truncated') return false; + if (REFRESH_APPEND_BOOKKEEPING_EVENT_TYPES.has(event.type)) return false; + return !isIdleBookkeepingSessionUpdate(event); +} + export function classifyTurnErrorKind( message: string, ): 'model_stream_interrupted' | undefined { @@ -1610,8 +1744,13 @@ function broadcastTurnError( mutateTurnState: boolean, ): void { const message = extractErrorMessage(err); - const code = extractErrorCode(err); - const errorKind = classifyTurnErrorKind(message); + const structuredErrorKind = extractJsonRpcErrorField(err, 'errorKind'); + const errorKind = structuredErrorKind ?? classifyTurnErrorKind(message); + const code = + structuredErrorKind !== undefined + ? (extractJsonRpcErrorField(err, 'code') ?? extractErrorCode(err)) + : extractErrorCode(err); + const loopType = extractJsonRpcErrorField(err, 'loopType'); if (errorKind) { writeServeDebugLine( `turn_error classified session=${JSON.stringify(sessionId)} ` + @@ -1636,7 +1775,7 @@ function broadcastTurnError( }; } try { - entry.events.publish({ + const published = entry.events.publish({ type: 'turn_error', ...(promptId ? { promptId } : {}), data: { @@ -1644,10 +1783,21 @@ function broadcastTurnError( message, ...(code ? { code } : {}), ...(errorKind ? { errorKind } : {}), + ...(loopType ? { loopType } : {}), ...(promptId ? { promptId } : {}), }, ...(originatorClientId ? { originatorClientId } : {}), }); + if (mutateTurnState) { + // Undefined when the bus dropped the publish (closed mid-teardown); + // the refresh-append guard then simply has nothing to replay. A + // queued prompt's terminal (mutateTurnState=false) publishes the + // event alone and leaves the active turn's refresh-replay record + // untouched — the prompt never ran, so its terminal is not a newer + // turn boundary for the replay (but see `publishPromptTerminal`: a + // queued boundary that folds newer turn content supersedes it). + entry.turnErrorEvent = published; + } } catch { /* bus may be closed during session teardown */ } @@ -1690,6 +1840,26 @@ function publishPromptTerminal( } pendingEntry.terminalPublished = true; const originatorClientId = pendingEntry.originatorClientId; + // Only a running prompt's terminal belongs to the active turn. The + // `state === 'running'` gate (not `activePromptId`) is deliberate: on + // the normal settle path `settleActivePromptState` runs in + // `promptPromise.finally` BEFORE the terminal is published, so + // `activePromptId` is already cleared when a genuine active terminal + // lands here. Queued terminals publish their event alone and must + // neither set nor clear session-scoped turn state. + const mutateTurnState = pendingEntry.state === 'running'; + if (!mutateTurnState && entry.turnErrorEvent) { + // A queued terminal is still a turn boundary on the bus: ingesting it + // folds and resets the live journal, erasing the guard's only evidence + // of newer turn content journaled since the pending error terminal. + // That content supersedes the stale error, so drop the refresh-replay + // record before the fold — otherwise the append would re-place the + // stale error AFTER the newer content. + const journal = entry.events.liveJournalSnapshot() ?? []; + if (journal.some(isRefreshAppendTurnContent)) { + entry.turnErrorEvent = undefined; + } + } if (terminal.kind === 'complete') { broadcastTurnComplete( entry, @@ -1697,6 +1867,7 @@ function publishPromptTerminal( terminal.result, pendingEntry.promptId, originatorClientId, + mutateTurnState, ); } else if (terminal.kind === 'cancelled') { broadcastTurnComplete( @@ -1705,6 +1876,7 @@ function publishPromptTerminal( { stopReason: 'cancelled' }, pendingEntry.promptId, originatorClientId, + mutateTurnState, ); } else { broadcastTurnError( @@ -1713,13 +1885,7 @@ function publishPromptTerminal( terminal.err, pendingEntry.promptId, originatorClientId, - // Only a running prompt's failure is the active turn's failure. The - // `state === 'running'` gate (not `activePromptId`) is deliberate: - // on the normal settle path `settleActivePromptState` runs in - // `promptPromise.finally` BEFORE the terminal is published, so - // `activePromptId` is already cleared when a genuine active failure - // lands here. - pendingEntry.state === 'running', + mutateTurnState, ); } } @@ -5608,8 +5774,29 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { entry.events.epoch === eventEpoch && entry.events.lastEventId === lastEventId ) { + let compactedReplay = page.events; + const turnErrorEvent = entry.turnErrorEvent; + if (turnErrorEvent) { + // Append only while no newer turn content was journaled after + // the in-memory terminal: automatic turns (cron/background + // notification) run without clearing entry.turnError, and + // re-appending the stale error after their newer content would + // misplace it in the refreshed transcript. Bookkeeping events + // carry no turn content and must not defeat the append; a + // newer turn terminal clears turnErrorEvent at broadcast. The + // journal holds exactly the events published since the last + // turn boundary (the terminal itself folds into the replay + // window), so no history scan is needed. + const journal = entry.events.liveJournalSnapshot() ?? []; + const hasNewerTurnContent = journal.some( + isRefreshAppendTurnContent, + ); + if (!hasNewerTurnContent) { + compactedReplay = [...page.events, turnErrorEvent]; + } + } return { - compactedReplay: page.events, + compactedReplay, liveJournal: [], lastEventId, eventEpoch, @@ -7636,6 +7823,11 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { delete meta[DAEMON_CHANNEL_DELIVERY_META_KEY]; delete meta[DAEMON_PROMPT_DISPLAY_TEXT_META_KEY]; delete meta[DAEMON_MODEL_PROMPT_META_KEY]; + // Channel classification is authenticated channel-worker + // metadata; the daemon prompt route validates the worker + // authorization and re-arms it through the trusted + // `channelPrompt` context flag below. + delete meta[CHANNEL_PROMPT_META_KEY]; if (isRetry) { meta[DAEMON_RETRY_META_KEY] = true; } @@ -7653,6 +7845,9 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { if (modelPrompt !== undefined) { meta[DAEMON_MODEL_PROMPT_META_KEY] = modelPrompt; } + if (context?.channelPrompt === true) { + meta[CHANNEL_PROMPT_META_KEY] = true; + } meta[INVOCATION_CONTEXT_META_KEY] = invocationContext; if (Object.keys(meta).length > 0) { copy._meta = meta; @@ -7665,6 +7860,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { entry.activePromptId = pendingEntry.promptId; delete entry.cancelBroadcastWithoutPrompt; delete entry.turnError; + delete entry.turnErrorEvent; activePromptCounter++; entry.sessionLastSeenAt = Date.now(); touchActivity(); @@ -7776,6 +7972,16 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { ); return; } + if (extractJsonRpcErrorField(err, 'errorKind')) { + // Structured turn error (e.g. loop_detected): the + // forward succeeded and the daemon rejected the turn + // after running it. The formal turn_error terminal + // already ends the turn visibly; a phantom + // forward-failure line and prompt_cancelled broadcast + // would misreport it. + cancelPendingForSession(sessionId); + return; + } writeStderrLine( `sendPrompt: forward failed for session ${sessionId}: ${extractErrorMessage(err)}`, ); diff --git a/packages/acp-bridge/src/bridgeTypes.ts b/packages/acp-bridge/src/bridgeTypes.ts index 17bdfdfdc5a..39143e9d021 100644 --- a/packages/acp-bridge/src/bridgeTypes.ts +++ b/packages/acp-bridge/src/bridgeTypes.ts @@ -698,6 +698,13 @@ export interface BridgeClientRequestContext { modelPrompt?: string; /** User-facing projection supplied by an authenticated channel worker. */ promptDisplayText?: string; + /** + * Trusted channel-turn classification injected by the daemon prompt route + * after validating the channel-worker prompt authorization. Never + * populated from caller-controlled ACP metadata: `sendPrompt` strips the + * wire key from untrusted callers and re-injects it only from this flag. + */ + channelPrompt?: boolean; /** Trusted Channel delivery correlation injected by the daemon prompt * route. Never populated from caller-controlled ACP metadata. */ channelDelivery?: { @@ -739,6 +746,9 @@ export function isValidTrustedModelPrompt(value: unknown): value is string { export const DAEMON_CHANNEL_DELIVERY_META_KEY = 'qwen.daemon.channelDelivery'; export const DAEMON_PROMPT_DISPLAY_TEXT_META_KEY = 'qwen.daemon.promptDisplayText'; +// Wire twin of channel-base's CHANNEL_PROMPT_META_KEY; the packages have no +// dependency path between them, so a cross-package test pins the value. +export const CHANNEL_PROMPT_META_KEY = 'qwen.channel.prompt'; /** * Returned from `recordHeartbeat`. `lastSeenAt` is the server-side diff --git a/packages/acp-bridge/src/compactionEngine.ts b/packages/acp-bridge/src/compactionEngine.ts index 9355f6f50de..5c412ef87d6 100644 --- a/packages/acp-bridge/src/compactionEngine.ts +++ b/packages/acp-bridge/src/compactionEngine.ts @@ -351,6 +351,20 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { this.makeHistoryTruncatedEvent(compactedTurns.length), ); } + return { + compactedTurns, + liveJournal: this.liveJournalSnapshot(liveReplayMode), + lastEventId: this.lastEventId, + }; + } + + /** + * Snapshot of only the in-flight live journal — the events ingested + * since the last turn boundary (a boundary folds its turn into the + * replay window and resets the journal). Cheaper than `snapshot()`: + * no replay-window flatten. + */ + liveJournalSnapshot(liveReplayMode: LiveReplayMode = 'full'): BridgeEvent[] { const journal = liveReplayMode === 'summary' ? this.summaryJournal : this.fullJournal; const journalRecordId = @@ -389,11 +403,7 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { }, }); } - return { - compactedTurns, - liveJournal, - lastEventId: this.lastEventId, - }; + return liveJournal; } seed(snapshot: { compactedTurns: BridgeEvent[]; lastEventId: number }): void { diff --git a/packages/acp-bridge/src/eventBus.ts b/packages/acp-bridge/src/eventBus.ts index 9eee0011bd6..0d3aa08a244 100644 --- a/packages/acp-bridge/src/eventBus.ts +++ b/packages/acp-bridge/src/eventBus.ts @@ -46,6 +46,12 @@ export interface CompactionEngine { ingest(event: BridgeEvent, byteLength?: number): void; seedReplayEvents(events: BridgeEvent[]): void; snapshot(liveReplayMode?: LiveReplayMode): SessionReplaySnapshot; + /** + * In-flight journal only — events ingested since the last turn + * boundary — without flattening the compacted replay window. Optional: + * consumers fall back to `snapshot()` semantics when absent. + */ + liveJournalSnapshot?(liveReplayMode?: LiveReplayMode): BridgeEvent[]; close(): void; /** * Current live-journal caps — may exceed the configured baseline when @@ -406,6 +412,18 @@ export class EventBus { return snapshot; } + /** + * Events ingested since the last turn boundary (the boundary itself is + * folded into the replay window), without flattening that window. + * Undefined when no compaction engine is wired or it exposes no journal + * snapshot. + */ + liveJournalSnapshot( + liveReplayMode: LiveReplayMode = 'full', + ): BridgeEvent[] | undefined { + return this.compactionEngine?.liveJournalSnapshot?.(liveReplayMode); + } + /** * The engine's current live-journal caps — may have grown past the * configured baseline under adaptive growth. Read by the bridge's diff --git a/packages/channels/base/src/ChannelAgentBridge.ts b/packages/channels/base/src/ChannelAgentBridge.ts index dfa33635cd4..01fff0cb201 100644 --- a/packages/channels/base/src/ChannelAgentBridge.ts +++ b/packages/channels/base/src/ChannelAgentBridge.ts @@ -7,7 +7,9 @@ export const CHANNEL_PROMPT_DISPLAY_TEXT_META_KEY = 'qwen.daemon.promptDisplayText'; export const CHANNEL_PROMPT_AUTHORIZATION_META_KEY = 'qwen.daemon.channelPromptAuthorization'; -// Client-supplied routing hint only; never use it as an authorization boundary. +// Channel-turn classification marker. Trusted-parent metadata: the daemon +// strips it from untrusted callers and honors it only when an authenticated +// channel worker (or a private-parent channel bridge) set it. export const CHANNEL_PROMPT_META_KEY = 'qwen.channel.prompt'; // Private-parent capability handshake with the spawned `qwen --acp` child // (packages/core/src/utils/invocation-context.ts owns the same constants). diff --git a/packages/channels/base/src/DaemonChannelBridge.test.ts b/packages/channels/base/src/DaemonChannelBridge.test.ts index d32e4e89e26..b390585664d 100644 --- a/packages/channels/base/src/DaemonChannelBridge.test.ts +++ b/packages/channels/base/src/DaemonChannelBridge.test.ts @@ -1976,6 +1976,40 @@ describe('DaemonChannelBridge', () => { bridge.stop(); }); + it('presents the prompt authorization even without a display text', async () => { + // The daemon validates the token for the channel-turn classification + // too; a channel prompt without display text still needs it to keep + // its classification (and the loop-rejection opt-out that rides it). + const events = new EventQueue(); + const session = createFakeSession(events); + const bridge = new DaemonChannelBridge({ + cwd: '/repo', + sessionFactory: vi.fn().mockResolvedValue(session), + promptAuthorization: 'worker-token', + }); + + await bridge.start(); + await bridge.newSession('/repo'); + + const promptPromise = bridge.prompt('session-1', 'hello'); + await waitFor(() => expect(session.prompt).toHaveBeenCalledOnce()); + expect(session.prompt).toHaveBeenCalledWith( + { + prompt: [{ type: 'text', text: 'hello' }], + _meta: { + [CHANNEL_PROMPT_META_KEY]: true, + [CHANNEL_PROMPT_AUTHORIZATION_META_KEY]: 'worker-token', + }, + }, + expect.any(AbortSignal), + ); + + events.push(turnCompleteEvent()); + await promptPromise; + events.close(); + bridge.stop(); + }); + it('aborts in-flight prompts when the bridge stops', async () => { const events = new EventQueue(); const session = createFakeSession(events); diff --git a/packages/channels/base/src/DaemonChannelBridge.ts b/packages/channels/base/src/DaemonChannelBridge.ts index b036b533f10..2441b915fed 100644 --- a/packages/channels/base/src/DaemonChannelBridge.ts +++ b/packages/channels/base/src/DaemonChannelBridge.ts @@ -417,10 +417,10 @@ export class DaemonChannelBridge }); } prompt.push({ type: 'text', text }); - const promptAuthorization = - options?.displayText !== undefined - ? this.options.promptAuthorization - : undefined; + // Always presented: the daemon validates it for the channel-turn + // classification as well as the display projection, and channel + // prompts without display text still need the classification. + const promptAuthorization = this.options.promptAuthorization; try { const result = await session.prompt( diff --git a/packages/cli/src/acp-integration/acpAgent.test.ts b/packages/cli/src/acp-integration/acpAgent.test.ts index 1bf779c96d1..973fdb365ff 100644 --- a/packages/cli/src/acp-integration/acpAgent.test.ts +++ b/packages/cli/src/acp-integration/acpAgent.test.ts @@ -2190,6 +2190,11 @@ describe('QwenAgent MCP SSE/HTTP support', () => { 'qwen-code/private-parent-capability': 'must-not-propagate', 'qwen.daemon.modelPrompt': 'trusted model-only prompt', 'qwen.daemon.promptDisplayText': 'trusted display text', + 'qwen.channel.prompt': true, + 'qwen.daemon.channelDelivery': { + deliveryId: 'delivery-trusted', + target: { channelName: 'dingtalk', type: 'user', id: 'user-1' }, + }, }, }); @@ -2200,6 +2205,11 @@ describe('QwenAgent MCP SSE/HTTP support', () => { _meta: { keep: true, 'qwen.daemon.promptDisplayText': 'trusted display text', + 'qwen.channel.prompt': true, + 'qwen.daemon.channelDelivery': { + deliveryId: 'delivery-trusted', + target: { channelName: 'dingtalk', type: 'user', id: 'user-1' }, + }, }, }, invocation, @@ -2210,6 +2220,57 @@ describe('QwenAgent MCP SSE/HTTP support', () => { await agentPromise; }); + it('strips channel classification from untrusted callers', async () => { + // `qwen.channel.prompt` marks a turn as a channel turn, opting it out + // of loop-detected rejection and the repeated-failure guard, and + // `qwen.daemon.channelDelivery` schedules the response delivery. Only + // trusted parents (the channel bridges and the daemon bridge) may set + // them; an untrusted client marking its own prompt must not reach the + // session. A plain `qwen --acp` child has no expected capability and + // initializes untrusted. + await setupSessionMocks('untrusted-session'); + const agentPromise = runAcpAgent( + mockConfig, + makeSessionSettings(), + mockArgv, + ); + await vi.waitFor(() => expect(capturedAgentFactory).toBeDefined()); + const agent = capturedAgentFactory!({ + get closed() { + return mockConnectionState.promise; + }, + }) as AgentLike; + await agent.initialize({ clientCapabilities: {} }); + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + + await agent.prompt({ + sessionId: 'untrusted-session', + prompt: [{ type: 'text', text: 'hello' }], + _meta: { + keep: true, + 'qwen.channel.prompt': true, + 'qwen.daemon.channelDelivery': { + deliveryId: 'delivery-forged', + target: { channelName: 'dingtalk', type: 'user', id: 'user-1' }, + }, + }, + }); + + expect(lastSessionMock?.prompt).toHaveBeenCalledWith( + { + sessionId: 'untrusted-session', + prompt: [{ type: 'text', text: 'hello' }], + _meta: { keep: true }, + }, + undefined, + expect.any(AbortSignal), + undefined, + ); + + mockConnectionState.resolve(); + await agentPromise; + }); + it('closes managed writers before resource shutdown on connection EOF', async () => { const innerConfig = await setupSessionMocks('managed-session'); const order: string[] = []; diff --git a/packages/cli/src/acp-integration/acpAgent.ts b/packages/cli/src/acp-integration/acpAgent.ts index f1f8f640480..52ace9ca899 100644 --- a/packages/cli/src/acp-integration/acpAgent.ts +++ b/packages/cli/src/acp-integration/acpAgent.ts @@ -173,7 +173,10 @@ import { } from './authMethods.js'; import { AcpFileSystemService } from './service/filesystem.js'; import { ndJsonStream } from '@qwen-code/acp-bridge/ndJsonStream'; -import { ACP_EVENT_LOOP_STALL_RESTART_MS } from '@qwen-code/channel-base'; +import { + ACP_EVENT_LOOP_STALL_RESTART_MS, + CHANNEL_PROMPT_META_KEY, +} from '@qwen-code/channel-base'; import { Readable, Writable } from 'node:stream'; import { normalizeDisabledToolList } from '../config/normalizeDisabledTools.js'; import { pipeline } from 'node:stream/promises'; @@ -342,6 +345,7 @@ import { CHANNEL_STARTUP_PROFILE_META_KEY, CHANNEL_STARTUP_PROFILE_VERSION, CLIENT_MCP_OVER_WS_CONFIG_FLAG, + DAEMON_CHANNEL_DELIVERY_META_KEY, DAEMON_MODEL_PROMPT_META_KEY, DAEMON_PROMPT_DISPLAY_TEXT_META_KEY, LOAD_REPLAY_BULK_MODE, @@ -5742,10 +5746,14 @@ class QwenAgent implements Agent { const suppliedContext = meta[INVOCATION_CONTEXT_META_KEY]; const suppliedModelPrompt = meta[DAEMON_MODEL_PROMPT_META_KEY]; const suppliedPromptDisplayText = meta[DAEMON_PROMPT_DISPLAY_TEXT_META_KEY]; + const suppliedChannelPrompt = meta[CHANNEL_PROMPT_META_KEY]; + const suppliedChannelDelivery = meta[DAEMON_CHANNEL_DELIVERY_META_KEY]; delete meta[INVOCATION_CONTEXT_META_KEY]; delete meta[DAEMON_MODEL_PROMPT_META_KEY]; delete meta[PRIVATE_PARENT_CAPABILITY_META_KEY]; delete meta[DAEMON_PROMPT_DISPLAY_TEXT_META_KEY]; + delete meta[CHANNEL_PROMPT_META_KEY]; + delete meta[DAEMON_CHANNEL_DELIVERY_META_KEY]; // The user-facing display projection is caller-controlled metadata; honor // it only for trusted parents (the daemon bridge re-injects the trusted // channel-worker value here). A plain delete would drop that re-injection. @@ -5755,6 +5763,26 @@ class QwenAgent implements Agent { ) { meta[DAEMON_PROMPT_DISPLAY_TEXT_META_KEY] = suppliedPromptDisplayText; } + // Channel classification is trusted-parent metadata: only the channel + // bridges and the daemon bridge hold the private parent capability. An + // untrusted caller must not be able to mark its own prompt as a channel + // turn — that opts the turn out of loop-detected rejection and the + // repeated-failure guard. + if ( + this.privateParentState === 'trusted' && + suppliedChannelPrompt === true + ) { + meta[CHANNEL_PROMPT_META_KEY] = true; + } + // Channel delivery is a daemon-managed side effect (the prompt route + // injects it from the trusted context); an untrusted direct-ACP caller + // must not self-schedule its own response delivery through the key. + if ( + this.privateParentState === 'trusted' && + suppliedChannelDelivery !== undefined + ) { + meta[DAEMON_CHANNEL_DELIVERY_META_KEY] = suppliedChannelDelivery; + } if (Object.keys(meta).length > 0) { sanitizedParams._meta = meta; } else { diff --git a/packages/cli/src/acp-integration/session/Session.test.ts b/packages/cli/src/acp-integration/session/Session.test.ts index 632d4c4a461..0d423e91be2 100644 --- a/packages/cli/src/acp-integration/session/Session.test.ts +++ b/packages/cli/src/acp-integration/session/Session.test.ts @@ -15,6 +15,7 @@ import * as path from 'node:path'; import { computeInitialTurnFromHistory, fireSessionPermissionDeniedForAutoMode, + LOOP_DETECTED_TURN_ERROR_MESSAGE, resolveExistingFile, resolveHomeLoopResolverRoots, Session, @@ -7430,9 +7431,17 @@ describe('Session', () => { ) .mockResolvedValueOnce(createEmptyStream()); - await session.prompt({ - sessionId: 'test-session-id', - prompt: [{ type: 'text', text: 'ask me before continuing' }], + await expect( + session.prompt({ + sessionId: 'test-session-id', + prompt: [{ type: 'text', text: 'ask me before continuing' }], + }), + ).rejects.toMatchObject({ + data: { + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: core.LoopType.INVALID_TOOL_PARAMS_STAGNATION, + }, }); expect(build).toHaveBeenCalledTimes(3); @@ -7676,22 +7685,46 @@ describe('Session', () => { args: { file_path: `file_${index}.ts` }, })); functionCalls[101].id = 'read_0'; - mockChat.sendMessageStream = vi.fn().mockResolvedValueOnce( - createStreamWithChunks([ - { - type: core.StreamEventType.CHUNK, - value: { functionCalls }, - }, - ]), - ); + mockChat.sendMessageStream = vi + .fn() + .mockResolvedValueOnce( + createStreamWithChunks([ + { + type: core.StreamEventType.CHUNK, + value: { functionCalls }, + }, + ]), + ) + .mockResolvedValueOnce(createEmptyStream()); - await session.prompt({ - sessionId: 'test-session-id', - prompt: [{ type: 'text', text: 'read many files' }], + await expect( + session.prompt({ + sessionId: 'test-session-id', + prompt: [{ type: 'text', text: 'read many files' }], + }), + ).rejects.toMatchObject({ + message: LOOP_DETECTED_TURN_ERROR_MESSAGE, + data: { + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: core.LoopType.TURN_TOOL_CALL_CAP, + }, }); expect(mockChat.sendMessageStream).toHaveBeenCalledTimes(1); expect(mockToolRegistry.getTool).not.toHaveBeenCalled(); + const skippedUpdates = vi + .mocked(mockClient.sessionUpdate) + .mock.calls.map(([params]) => params.update) + .filter( + (update) => + update.sessionUpdate === 'tool_call_update' && + update.status === 'failed' && + JSON.stringify(update).includes( + 'Skipped because loop detection stopped the current turn', + ), + ); + expect(skippedUpdates).toHaveLength(101); expect(mockChat.addHistory).toHaveBeenCalledWith({ role: 'user', parts: expect.arrayContaining([ @@ -7728,6 +7761,86 @@ describe('Session', () => { 'Stopping ACP turn after 101 tool calls in one turn.', ), ); + + await expect( + session.prompt({ + sessionId: 'test-session-id', + prompt: [{ type: 'text', text: 'continue with a simpler step' }], + }), + ).resolves.toEqual({ stopReason: 'end_turn' }); + expect(mockChat.sendMessageStream).toHaveBeenCalledTimes(2); + }); + + it('drains cron work queued mid-turn when the turn rejects on loop protection', async () => { + mockConfig.getApprovalMode = vi.fn().mockReturnValue(ApprovalMode.YOLO); + mockConfig.getMaxToolCallsPerTurn = vi.fn().mockReturnValue(1); + mockConfig.isMaxToolCallsPerTurnExplicit = vi + .fn() + .mockReturnValue(true); + let fireCron!: (job: { prompt: string; cronExpr: string }) => void; + const scheduler = { + hasPendingWork: true, + enableDurable: vi.fn().mockResolvedValue(undefined), + start: vi.fn( + (callback: (job: { prompt: string; cronExpr: string }) => void) => { + fireCron = callback; + }, + ), + stop: vi.fn(), + list: vi.fn().mockReturnValue([]), + getExitSummary: vi.fn().mockReturnValue(undefined), + }; + mockConfig.isCronEnabled = vi.fn().mockReturnValue(true); + mockConfig.getCronScheduler = vi.fn().mockReturnValue(scheduler); + session.startCronScheduler(); + await vi.waitFor(() => expect(scheduler.start).toHaveBeenCalled()); + + // Gate the model stream so the cron fires while the foreground turn + // is still active; the turn then trips the explicit one-call cap and + // rejects. Loop-detected turns resolved end_turn before they became + // rejections (and drained), so the rejection path must drain too. + let releaseStream!: () => void; + const streamGate = new Promise((resolve) => { + releaseStream = resolve; + }); + async function* gatedCapTripStream() { + await streamGate; + yield { + type: core.StreamEventType.CHUNK, + value: { + functionCalls: [ + { id: 'cap-1', name: 'read_file', args: { path: 'a' } }, + { id: 'cap-2', name: 'read_file', args: { path: 'b' } }, + ], + }, + }; + } + mockChat.sendMessageStream = vi + .fn() + .mockResolvedValueOnce(gatedCapTripStream()) + .mockResolvedValue(createEmptyStream()); + + const prompt = session.prompt({ + sessionId: 'test-session-id', + prompt: [{ type: 'text', text: 'foreground work' }], + }); + await vi.waitFor(() => { + expect(mockChat.sendMessageStream).toHaveBeenCalledTimes(1); + }); + + fireCron({ prompt: 'scheduled work', cronExpr: '* * * * *' }); + const internals = session as unknown as { cronQueue: unknown[] }; + expect(internals.cronQueue).toHaveLength(1); + + releaseStream(); + await expect(prompt).rejects.toMatchObject({ + data: expect.objectContaining({ code: 'LOOP_DETECTED' }), + }); + + await vi.waitFor(() => { + expect(mockChat.sendMessageStream).toHaveBeenCalledTimes(2); + }); + expect(internals.cronQueue).toHaveLength(0); }); it('lets a productive turn continue past the default cap (adaptive)', async () => { @@ -8917,6 +9030,50 @@ describe('Session', () => { } }); + it('keeps the configured guard for delivery-marked prompts', async () => { + // The delivery meta is a caller-requested side effect, not a + // channel classification: it schedules the delivery on end_turn + // but must not opt the turn out of the repeated-failure guard or + // loop-detected rejection, or any caller could bypass loop + // protection by marking its own prompt for delivery. + recreateSessionWithGuardMode('enforce'); + try { + const execute = installFailingTool(); + queueMatchingFailureStreak(); + + await expect( + session.prompt({ + sessionId: 'test-session-id', + prompt: [{ type: 'text', text: 'channel delivery task' }], + _meta: { + 'qwen.daemon.channelDelivery': { + deliveryId: 'prompt-guard-off-delivery', + target: { + channelName: 'dingtalk', + type: 'user', + id: 'user-1', + }, + }, + }, + }), + ).rejects.toMatchObject({ + data: expect.objectContaining({ + code: 'LOOP_DETECTED', + loopType: core.LoopType.REPEATED_TOOL_EXECUTION_FAILURE, + }), + }); + + expect(execute).toHaveBeenCalledTimes(9); + expect(mockChat.sendMessageStream).toHaveBeenCalledTimes(3); + expect(mockClient.extMethod).not.toHaveBeenCalledWith( + 'qwen/control/channel-delivery', + expect.anything(), + ); + } finally { + restoreGuardMode(); + } + }); + it('forces channel-routed prompts off even when enforcement is configured', async () => { recreateSessionWithGuardMode('enforce'); try { @@ -9069,7 +9226,13 @@ describe('Session', () => { sessionId: 'test-session-id', prompt: [{ type: 'text', text: 'run the failing tool' }], }), - ).resolves.toEqual({ stopReason: 'end_turn' }); + ).rejects.toMatchObject({ + data: { + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + loopType: core.LoopType.REPEATED_TOOL_EXECUTION_FAILURE, + }, + }); expect(execute).toHaveBeenCalledTimes(9); expect(mockChat.sendMessageStream).toHaveBeenCalledTimes(3); @@ -9100,20 +9263,16 @@ describe('Session', () => { }), ]), }); - expect(mockClient.sessionUpdate).toHaveBeenCalledWith( - expect.objectContaining({ - sessionId: 'test-session-id', - update: expect.objectContaining({ - sessionUpdate: 'agent_message_chunk', - content: expect.objectContaining({ - type: 'text', - text: expect.stringContaining( - 'Automatic continuation stopped', - ), - }), - }), + expect( + vi.mocked(mockClient.sessionUpdate).mock.calls.some(([params]) => { + const update = params.update; + return ( + update.sessionUpdate === 'agent_message_chunk' && + update.content.type === 'text' && + update.content.text.includes('Automatic continuation stopped') + ); }), - ); + ).toBe(false); } finally { restoreGuardMode(); } @@ -9163,31 +9322,44 @@ describe('Session', () => { } }); - it('returns cancelled when cancellation arrives while the stop message is emitted', async () => { + it('returns cancelled when cancellation races the repeated-failure stop', async () => { recreateSessionWithGuardMode('enforce'); try { - installFailingTool(); - vi.mocked(mockClient.sessionUpdate).mockImplementation( - async ({ update }) => { - if ( - update.sessionUpdate === 'agent_message_chunk' && - update.content.type === 'text' && - update.content.text.includes('Automatic continuation stopped') - ) { - await session.cancelPendingPrompt(); - } - }, - ); + const execute = installFailingTool(); + let enterRewriterWait!: () => void; + const rewriterWaitStarted = new Promise((resolve) => { + enterRewriterWait = resolve; + }); + let releaseRewriterWait!: () => void; + const rewriterWaitGate = new Promise((resolve) => { + releaseRewriterWait = resolve; + }); + session.messageRewriter = { + interceptUpdate: vi.fn().mockResolvedValue(undefined), + waitForPendingRewrites: vi.fn(async () => { + enterRewriterWait(); + await rewriterWaitGate; + }), + } as unknown as NonNullable; queueMatchingFailureStreak(); - await expect( - session.prompt({ - sessionId: 'test-session-id', - prompt: [{ type: 'text', text: 'run the failing tool' }], - }), - ).resolves.toEqual({ stopReason: 'cancelled' }); + const prompt = session.prompt({ + sessionId: 'test-session-id', + prompt: [{ type: 'text', text: 'run the failing tool' }], + }); + await rewriterWaitStarted; + await session.cancelPendingPrompt(); + releaseRewriterWait(); - expect(mockChat.sendMessageStream).toHaveBeenCalledTimes(3); + await expect(prompt).resolves.toEqual({ stopReason: 'cancelled' }); + expect(execute).toHaveBeenCalledTimes(9); + expect(logLoopDetectedSpy).toHaveBeenCalledWith( + mockConfig, + expect.objectContaining({ + loop_type: core.LoopType.REPEATED_TOOL_EXECUTION_FAILURE, + }), + { recordToQwenLogger: false }, + ); } finally { restoreGuardMode(); } @@ -11974,6 +12146,127 @@ describe('Session', () => { expect(capture.writeToSpan).toHaveBeenCalledWith(agentTelemetry.span); }); + it('rejects a delivery-marked turn when loop protection stops it', async () => { + // The delivery meta alone does not classify a turn as a channel + // turn: the loop-detected stop rejects like any foreground prompt + // instead of resolving end_turn, and the failed turn schedules no + // delivery. + mockConfig.getApprovalMode = vi.fn().mockReturnValue(ApprovalMode.YOLO); + mockConfig.getMaxToolCallsPerTurn = vi.fn().mockReturnValue(1); + mockConfig.isMaxToolCallsPerTurnExplicit = vi + .fn() + .mockReturnValue(true); + mockChat.sendMessageStream = vi.fn().mockResolvedValueOnce( + createStreamWithChunks([ + { + type: core.StreamEventType.CHUNK, + value: { + functionCalls: [ + { + id: 'channel-loop-1', + name: 'read_file', + args: { file_path: 'a.ts' }, + }, + { + id: 'channel-loop-2', + name: 'read_file', + args: { file_path: 'b.ts' }, + }, + ], + }, + }, + ]), + ); + + await expect( + session.prompt({ + sessionId: 'test-session-id', + prompt: [{ type: 'text', text: 'channel work' }], + _meta: { + 'qwen.daemon.channelDelivery': { + deliveryId: 'prompt-loop-channel', + target: { + channelName: 'dingtalk', + type: 'user', + id: 'user-1', + }, + }, + }, + }), + ).rejects.toMatchObject({ + data: expect.objectContaining({ + code: 'LOOP_DETECTED', + loopType: core.LoopType.TURN_TOOL_CALL_CAP, + }), + }); + + expect(logLoopDetectedSpy).toHaveBeenCalledWith( + mockConfig, + expect.objectContaining({ + loop_type: core.LoopType.TURN_TOOL_CALL_CAP, + }), + {}, + ); + expect(mockClient.extMethod).not.toHaveBeenCalledWith( + 'qwen/control/channel-delivery', + expect.anything(), + ); + }); + + it('keeps a channel-prompt-meta turn graceful when loop protection stops it', async () => { + // DaemonChannelBridge/AcpBridge channel tasks prompt with + // CHANNEL_PROMPT_META_KEY; the authenticated classification must + // resolve end_turn so the bridge emits promptComplete with the + // collected response text instead of the rejection failing the + // non-interactive task. + mockConfig.getApprovalMode = vi.fn().mockReturnValue(ApprovalMode.YOLO); + mockConfig.getMaxToolCallsPerTurn = vi.fn().mockReturnValue(1); + mockConfig.isMaxToolCallsPerTurnExplicit = vi + .fn() + .mockReturnValue(true); + mockChat.sendMessageStream = vi.fn().mockResolvedValueOnce( + createStreamWithChunks([ + { + type: core.StreamEventType.CHUNK, + value: { + functionCalls: [ + { + id: 'channel-prompt-loop-1', + name: 'read_file', + args: { file_path: 'a.ts' }, + }, + { + id: 'channel-prompt-loop-2', + name: 'read_file', + args: { file_path: 'b.ts' }, + }, + ], + }, + }, + ]), + ); + + await expect( + session.prompt({ + sessionId: 'test-session-id', + prompt: [{ type: 'text', text: 'channel task' }], + _meta: { [CHANNEL_PROMPT_META_KEY]: true }, + }), + ).resolves.toEqual({ stopReason: 'end_turn' }); + + expect(logLoopDetectedSpy).toHaveBeenCalledWith( + mockConfig, + expect.objectContaining({ + loop_type: core.LoopType.TURN_TOOL_CALL_CAP, + }), + {}, + ); + expect(mockClient.extMethod).not.toHaveBeenCalledWith( + 'qwen/control/channel-delivery', + expect.anything(), + ); + }); + it('replaces the prompt candidate with a Stop-hook continuation final', async () => { agentTelemetry.getActiveInteractionSpan.mockReturnValue( agentTelemetry.span, @@ -15256,13 +15549,220 @@ describe('Session', () => { continuationContext: 'check weather', }); - // `releaseTurn`, not `finishTurn`: the turn never reached the model, - // so it is not an iteration the Goal made progress on. - await vi.waitFor(() => { - expect(mockGoalRuntime.releaseTurn).toHaveBeenCalledWith(turnKey); - }); - expect(mockChat.sendMessageStream).not.toHaveBeenCalled(); - expect(mockGoalRuntime.finishTurn).not.toHaveBeenCalled(); + // `releaseTurn`, not `finishTurn`: the turn never reached the model, + // so it is not an iteration the Goal made progress on. + await vi.waitFor(() => { + expect(mockGoalRuntime.releaseTurn).toHaveBeenCalledWith(turnKey); + }); + expect(mockChat.sendMessageStream).not.toHaveBeenCalled(); + expect(mockGoalRuntime.finishTurn).not.toHaveBeenCalled(); + }); + + it('keeps a Goal turn graceful when loop protection stops it', async () => { + // Goal continuations are non-interactive and bypass the bridge: a + // rejection would settle the turn as failed and pause the goal + // with no turn_error ever published. They resolve end_turn like + // cron and channel turns, settling the iteration normally. + mockConfig.getApprovalMode = vi.fn().mockReturnValue(ApprovalMode.YOLO); + mockConfig.getMaxToolCallsPerTurn = vi.fn().mockReturnValue(1); + mockConfig.isMaxToolCallsPerTurnExplicit = vi + .fn() + .mockReturnValue(true); + const permit: core.GoalTurnPermit = { + goalId: 'goal-1', + revision: 1, + turnId: 'turn-loop-cap', + }; + const turnKey = 'goal-runtime:turn-loop-cap'; + mockGoalRuntime.getSnapshot.mockReturnValue({ + v: 2, + activity: 'running', + goal: { + goalId: 'goal-1', + revision: 1, + objective: 'check weather', + status: 'active', + evidenceCursor: { recordId: 'cursor-1' }, + turnCount: 0, + activeTimeMs: 0, + createdAt: 1234, + updatedAt: 1234, + }, + }); + mockGoalRuntime.permitForTurn.mockImplementation((key: string) => + key === turnKey ? permit : undefined, + ); + mockChat.sendMessageStream = vi.fn().mockResolvedValueOnce( + createStreamWithChunks([ + { + type: core.StreamEventType.CHUNK, + value: { + functionCalls: [ + { + id: 'goal-loop-1', + name: 'read_file', + args: { file_path: 'a.ts' }, + }, + { + id: 'goal-loop-2', + name: 'read_file', + args: { file_path: 'b.ts' }, + }, + ], + }, + }, + ]), + ); + + expect(boundGoalHost).toBeDefined(); + await boundGoalHost!.startGoalTurn({ + permit, + continuationContext: 'check weather', + }); + + await vi.waitFor(() => { + expect(logLoopDetectedSpy).toHaveBeenCalledWith( + mockConfig, + expect.objectContaining({ + loop_type: core.LoopType.TURN_TOOL_CALL_CAP, + }), + {}, + ); + }); + // Graceful end_turn settles the iteration; the goal is not paused. + await vi.waitFor(() => { + expect(mockGoalRuntime.finishTurn).toHaveBeenCalledWith(permit); + }); + expect(mockGoalRuntime.dispatch).not.toHaveBeenCalled(); + }); + + it('keeps a Goal turn graceful when the repeated-failure guard stops it', async () => { + // Goal turns keep the configured guard mode (they are not channel + // turns) but get rejectOnLoopDetected=false, so an enforce-mode + // failure streak stops them through the graceful branch: end_turn + // settlement plus the transcript stop message, never a rejection + // that would pause the goal without a published turn_error. + const guardModeEnv = 'QWEN_CODE_ACP_REPEATED_TOOL_FAILURE_GUARD'; + const previousGuardMode = process.env[guardModeEnv]; + process.env[guardModeEnv] = 'enforce'; + try { + session = new Session( + 'test-session-id', + mockConfig, + mockClient, + mockSettings, + ); + mockConfig.getApprovalMode = vi + .fn() + .mockReturnValue(ApprovalMode.YOLO); + mockConfig.getDisableAllHooks = vi.fn().mockReturnValue(true); + const execute = vi.fn().mockResolvedValue({ + llmContent: 'failed', + returnDisplay: 'failed', + error: { + message: 'execution failed', + type: core.ToolErrorType.EXECUTION_FAILED, + }, + }); + mockToolRegistry.getTool.mockReturnValue({ + name: 'failing_tool', + kind: core.Kind.Execute, + displayName: 'Failing Tool', + description: 'Fails during execution', + build: vi.fn().mockReturnValue({ + params: {}, + execute, + getDefaultPermission: vi.fn().mockResolvedValue('allow'), + getDescription: vi.fn().mockReturnValue('Failing Tool'), + toolLocations: vi.fn().mockReturnValue([]), + }), + canUpdateOutput: false, + isOutputMarkdown: true, + }); + const streamForBatch = (batch: number, count: number) => + createStreamWithChunks([ + { + type: core.StreamEventType.CHUNK, + value: { + functionCalls: Array.from({ length: count }, (_, index) => ({ + id: `goal_failure_${batch}_${index}`, + name: 'failing_tool', + args: { attempt: `${batch}_${index}` }, + })), + }, + }, + ]); + mockChat.sendMessageStream = vi + .fn() + .mockResolvedValueOnce(streamForBatch(1, 4)) + .mockResolvedValueOnce(streamForBatch(2, 4)) + .mockResolvedValueOnce(streamForBatch(3, 1)) + .mockResolvedValueOnce(createEmptyStream()); + + const permit: core.GoalTurnPermit = { + goalId: 'goal-1', + revision: 1, + turnId: 'turn-guard-stop', + }; + const turnKey = 'goal-runtime:turn-guard-stop'; + mockGoalRuntime.getSnapshot.mockReturnValue({ + v: 2, + activity: 'running', + goal: { + goalId: 'goal-1', + revision: 1, + objective: 'check weather', + status: 'active', + evidenceCursor: { recordId: 'cursor-1' }, + turnCount: 0, + activeTimeMs: 0, + createdAt: 1234, + updatedAt: 1234, + }, + }); + mockGoalRuntime.permitForTurn.mockImplementation((key: string) => + key === turnKey ? permit : undefined, + ); + + expect(boundGoalHost).toBeDefined(); + await boundGoalHost!.startGoalTurn({ + permit, + continuationContext: 'check weather', + }); + + await vi.waitFor(() => { + expect(logLoopDetectedSpy).toHaveBeenCalledWith( + mockConfig, + expect.objectContaining({ + loop_type: core.LoopType.REPEATED_TOOL_EXECUTION_FAILURE, + }), + { recordToQwenLogger: false }, + ); + }); + // Graceful end_turn settles the iteration; the goal is not paused. + await vi.waitFor(() => { + expect(mockGoalRuntime.finishTurn).toHaveBeenCalledWith(permit); + }); + expect(mockGoalRuntime.dispatch).not.toHaveBeenCalled(); + // The graceful stop keeps the user-visible stop message: it is + // the only explanation of a silently stopped autonomous turn. + expect( + vi.mocked(mockClient.sessionUpdate).mock.calls.some(([params]) => { + const update = params.update; + return ( + update.sessionUpdate === 'agent_message_chunk' && + update.content.type === 'text' && + update.content.text.includes('Automatic continuation stopped') + ); + }), + ).toBe(true); + } finally { + if (previousGuardMode === undefined) { + delete process.env[guardModeEnv]; + } else { + process.env[guardModeEnv] = previousGuardMode; + } + } }); it('pauses without counting a Goal turn cancelled before the model request', async () => { @@ -26268,6 +26768,106 @@ describe('Session', () => { await expect(prompt).resolves.toEqual({ stopReason: 'cancelled' }); }); + it('lets cancellation win while a loop-detected Stop continuation is preserved', async () => { + mockConfig.getMaxToolCallsPerTurn = vi.fn().mockReturnValue(1); + mockConfig.isMaxToolCallsPerTurnExplicit = vi.fn().mockReturnValue(true); + mockChat.sendMessageStream = vi + .fn() + .mockResolvedValueOnce(createEmptyStream()) + .mockResolvedValueOnce( + createStreamWithChunks([ + { + type: core.StreamEventType.CHUNK, + value: { + functionCalls: [ + { id: 'loop-1', name: 'read_file', args: { path: 'a' } }, + { id: 'loop-2', name: 'read_file', args: { path: 'b' } }, + ], + }, + }, + ]), + ); + const messageBus = { + request: vi.fn().mockResolvedValue({ + success: true, + output: { decision: 'block', reason: 'continue once' }, + }), + }; + mockConfig.getMessageBus = vi.fn().mockReturnValue(messageBus); + mockConfig.hasHooksForEvent = vi + .fn() + .mockImplementation((name: string) => name === 'Stop'); + let startDrain!: () => void; + const drainStarted = new Promise((resolve) => { + startDrain = resolve; + }); + let releaseDrain!: () => void; + const drainGate = new Promise((resolve) => { + releaseDrain = resolve; + }); + mockClient.extMethod = vi.fn(async () => { + startDrain(); + await drainGate; + return { messages: [] }; + }); + + const prompt = runGuardPrompt(); + await drainStarted; + await session.cancelPendingPrompt(); + releaseDrain(); + + await expect(prompt).resolves.toEqual({ stopReason: 'cancelled' }); + }); + + it('rejects a foreground turn whose Stop continuation trips loop protection', async () => { + // Pins rejectOnLoopDetected=true at the foreground #handleStopHookLoop + // call site: without it this turn would resolve end_turn. + mockConfig.getMaxToolCallsPerTurn = vi.fn().mockReturnValue(1); + mockConfig.isMaxToolCallsPerTurnExplicit = vi.fn().mockReturnValue(true); + mockChat.sendMessageStream = vi + .fn() + .mockResolvedValueOnce(createEmptyStream()) + .mockResolvedValueOnce( + createStreamWithChunks([ + { + type: core.StreamEventType.CHUNK, + value: { + functionCalls: [ + { id: 'loop-1', name: 'read_file', args: { path: 'a' } }, + { id: 'loop-2', name: 'read_file', args: { path: 'b' } }, + ], + }, + }, + ]), + ); + const messageBus = { + request: vi.fn().mockResolvedValue({ + success: true, + output: { decision: 'block', reason: 'continue once' }, + }), + }; + mockConfig.getMessageBus = vi.fn().mockReturnValue(messageBus); + mockConfig.hasHooksForEvent = vi + .fn() + .mockImplementation((name: string) => name === 'Stop'); + mockClient.extMethod = vi.fn(async () => ({ messages: [] })); + + await expect(runGuardPrompt()).rejects.toMatchObject({ + message: LOOP_DETECTED_TURN_ERROR_MESSAGE, + data: expect.objectContaining({ + code: 'LOOP_DETECTED', + loopType: core.LoopType.TURN_TOOL_CALL_CAP, + }), + }); + expect(logLoopDetectedSpy).toHaveBeenCalledWith( + mockConfig, + expect.objectContaining({ + loop_type: core.LoopType.TURN_TOOL_CALL_CAP, + }), + {}, + ); + }); + it('runs exactly two continuations and emits replayable status', async () => { rebuildSessionWithGuard(); installPendingTodoTool(); @@ -31774,6 +32374,207 @@ describe('Session', () => { ).toBe(false); }); + it('keeps a cron turn graceful when its Stop continuation trips loop protection', async () => { + let fireCron!: (job: { + prompt: string; + cronExpr: string; + missed?: boolean; + }) => void; + const scheduler = { + hasPendingWork: true, + enableDurable: vi.fn().mockResolvedValue(undefined), + start: vi.fn( + ( + callback: (job: { + prompt: string; + cronExpr: string; + missed?: boolean; + }) => void, + ) => { + fireCron = callback; + }, + ), + stop: vi.fn(), + list: vi.fn().mockReturnValue([]), + getExitSummary: vi.fn().mockReturnValue(undefined), + }; + mockConfig.isCronEnabled = vi.fn().mockReturnValue(true); + mockConfig.getCronScheduler = vi.fn().mockReturnValue(scheduler); + rebuildSessionWithGuard(); + installPendingTodoTool(); + mockChat.sendMessageStream = vi + .fn() + .mockResolvedValueOnce(createEmptyStream()) + .mockResolvedValueOnce( + createStreamWithChunks([ + { + type: core.StreamEventType.CHUNK, + value: { + functionCalls: [ + { + id: 'cron-todo', + name: core.ToolNames.TODO_WRITE, + args: { todos: pendingTodos }, + }, + ], + }, + }, + ]), + ) + .mockResolvedValueOnce(createEmptyStream()) + .mockResolvedValueOnce( + createStreamWithChunks([ + { + type: core.StreamEventType.CHUNK, + value: { + functionCalls: [ + { + id: 'cron-loop-1', + name: 'read_file', + args: { path: 'a' }, + }, + { + id: 'cron-loop-2', + name: 'read_file', + args: { path: 'b' }, + }, + ], + }, + }, + ]), + ) + .mockResolvedValue(createEmptyStream()); + + await runGuardPrompt(); + // Explicit one-call cap: the cron turn's Stop-continuation batch of + // two calls trips the per-turn cap inside #runStopContinuation, the + // shared path cron and background-notification turns reach through + // #handleStopHookLoop. + mockConfig.getMaxToolCallsPerTurn = vi.fn().mockReturnValue(1); + mockConfig.isMaxToolCallsPerTurnExplicit = vi.fn().mockReturnValue(true); + fireCron({ prompt: 'scheduled work', cronExpr: '* * * * *' }); + + await vi.waitFor(() => { + expect(mockChat.sendMessageStream).toHaveBeenCalledTimes(4); + }); + expect(logLoopDetectedSpy).toHaveBeenCalledWith( + mockConfig, + expect.objectContaining({ + loop_type: core.LoopType.TURN_TOOL_CALL_CAP, + }), + {}, + ); + expect( + vi.mocked(mockClient.sessionUpdate).mock.calls.some(([params]) => { + const update = params.update; + return ( + update.sessionUpdate === 'agent_message_chunk' && + update.content.type === 'text' && + update.content.text.includes('[cron error]') + ); + }), + ).toBe(false); + }); + + it('keeps a background-notification turn graceful when its Stop continuation trips loop protection', async () => { + rebuildSessionWithGuard(); + installPendingTodoTool(); + mockChat.sendMessageStream = vi + .fn() + .mockResolvedValueOnce(createEmptyStream()) + .mockResolvedValueOnce( + createStreamWithChunks([ + { + type: core.StreamEventType.CHUNK, + value: { + functionCalls: [ + { + id: 'notification-todo', + name: core.ToolNames.TODO_WRITE, + args: { todos: pendingTodos }, + }, + ], + }, + }, + ]), + ) + .mockResolvedValueOnce(createEmptyStream()) + .mockResolvedValueOnce( + createStreamWithChunks([ + { + type: core.StreamEventType.CHUNK, + value: { + functionCalls: [ + { + id: 'notification-loop-1', + name: 'read_file', + args: { path: 'a' }, + }, + { + id: 'notification-loop-2', + name: 'read_file', + args: { path: 'b' }, + }, + ], + }, + }, + ]), + ) + .mockResolvedValue(createEmptyStream()); + + await runGuardPrompt(); + // Explicit one-call cap: the notification turn's Stop-continuation + // batch of two calls trips the per-turn cap inside + // #runStopContinuation, pinning the graceful default at the + // background-notification #handleStopHookLoop call site. + mockConfig.getMaxToolCallsPerTurn = vi.fn().mockReturnValue(1); + mockConfig.isMaxToolCallsPerTurnExplicit = vi.fn().mockReturnValue(true); + const callback = + mockBackgroundTaskRegistry.setNotificationCallback.mock.calls.at( + -1, + )?.[0] as ( + displayText: string, + modelText: string, + meta: { agentId: string; status: string }, + ) => void; + + callback('background done', '', { + agentId: 'automatic-agent', + status: 'completed', + }); + + await vi.waitFor(() => { + expect(mockChat.sendMessageStream).toHaveBeenCalledTimes(4); + }); + expect(logLoopDetectedSpy).toHaveBeenCalledWith( + mockConfig, + expect.objectContaining({ + loop_type: core.LoopType.TURN_TOOL_CALL_CAP, + }), + {}, + ); + expect( + vi.mocked(mockClient.sessionUpdate).mock.calls.some(([params]) => { + const update = params.update; + return ( + update.sessionUpdate === 'agent_message_chunk' && + update.content.type === 'text' && + update.content.text.includes('[notification error]') + ); + }), + ).toBe(false); + await vi.waitFor(() => { + expect(mockClient.extNotification).toHaveBeenCalledWith( + '_qwencode/end_turn', + { + sessionId: 'test-session-id', + reason: 'end_turn', + source: 'background_notification', + }, + ); + }); + }); + it('suspends an armed guard when a cron stream aborts', async () => { const scheduler = { hasPendingWork: true, diff --git a/packages/cli/src/acp-integration/session/Session.ts b/packages/cli/src/acp-integration/session/Session.ts index 63beb1bd9d0..711d7f22812 100644 --- a/packages/cli/src/acp-integration/session/Session.ts +++ b/packages/cli/src/acp-integration/session/Session.ts @@ -586,6 +586,7 @@ export type DaemonToolLoopState = { /** Highest repeat count of any single (tool, args) pair this turn. */ maxToolCallKeyRepeat: number; loopDetected: boolean; + loopType?: LoopType; repeatedToolFailureMode: RepeatedToolFailureGuardMode; repeatedToolFailureState: RepeatedToolFailureGuardState; }; @@ -598,6 +599,8 @@ const LOOP_DETECTED_SKIP_MESSAGE = 'Skipped because loop detection stopped the current turn before this tool call could run.'; const LOOP_DETECTED_CONTEXT_MESSAGE = 'System: this turn was terminated because the model exceeded tool-call safety limits. Try a different approach on the next turn.'; +export const LOOP_DETECTED_TURN_ERROR_MESSAGE = + 'Tool-call loop protection stopped this turn. The session is still available; send a more specific instruction to continue.'; const TOOL_EXECUTION_CANCELLED_MESSAGE = 'Tool execution was cancelled.'; const TOOL_POST_EXECUTION_CANCELLED_MESSAGE = 'The tool had already completed; its output was discarded.'; @@ -703,6 +706,7 @@ function recordDaemonLoopDetected( ): true { if (!loopState.loopDetected) { loopState.loopDetected = true; + loopState.loopType = loopType; debugLogger.warn(message); try { logLoopDetected( @@ -720,6 +724,35 @@ function recordDaemonLoopDetected( return true; } +function createLoopDetectedTurnError( + loopState: DaemonToolLoopState, +): RequestError { + return new RequestError(-32603, LOOP_DETECTED_TURN_ERROR_MESSAGE, { + code: 'LOOP_DETECTED', + errorKind: 'loop_detected', + ...(loopState.loopType ? { loopType: loopState.loopType } : {}), + }); +} + +// Cancellation takes precedence when it races a loop-detected stop. +function cancelledOrThrowLoopDetected( + signal: AbortSignal, + loopState: DaemonToolLoopState, +): 'cancelled' { + if (signal.aborted) return 'cancelled'; + throw createLoopDetectedTurnError(loopState); +} + +function isLoopDetectedTurnError(error: unknown): boolean { + if (!(error instanceof RequestError)) return false; + const data = error.data; + return ( + typeof data === 'object' && + data !== null && + (data as { code?: unknown }).code === 'LOOP_DETECTED' + ); +} + function recordDaemonToolCalls( config: Config, promptId: string, @@ -3593,6 +3626,19 @@ export class Session implements SessionContext { ...(channelDelivery ? { channelDelivery: { finalText: '' } } : {}), agentOutput: new AgentOutputMessageCapture(this.config), }; + const channelPromptTurn = + (params as { _meta?: Record })._meta?.[ + CHANNEL_PROMPT_META_KEY + ] === true; + // One server-side channel classification, consumed by both the + // rejection gate below and the guard-mode selection in + // #executePromptInner. Only the authenticated channel-prompt marker + // classifies a turn: the delivery meta is a caller-requested side + // effect (the response is still delivered on end_turn below), and + // letting it classify would let any caller opt its own turn out of + // loop-detected rejection and the repeated-failure guard. The ACP + // boundary strips the channel-prompt key from untrusted callers, so + // both decisions see only trusted values. // Track this prompt's completion for the next prompt to await let resolveCompletion!: () => void; @@ -3600,6 +3646,7 @@ export class Session implements SessionContext { resolveCompletion = resolve; }); + let rejectedByLoopProtection = false; let promptResult: PromptResponse | undefined; let promptFailed = false; try { @@ -3609,7 +3656,18 @@ export class Session implements SessionContext { responseCapture, invocationContext, modelPrompt, + // Channel turns are non-interactive deliveries: like cron, + // background-notification, and goal turns they keep the graceful + // end-turn handling so the collected response text is still + // delivered. Only the authenticated CHANNEL_PROMPT_META_KEY turns + // sent by the channel bridges qualify; the delivery meta alone + // schedules the delivery but keeps the foreground rejection. Goal + // turns bypass the bridge entirely, so a rejection there would + // settle the turn as failed and pause the goal without any + // turn_error ever being published. + !channelPromptTurn && goalTurn === undefined, goalTurn, + channelPromptTurn, ); promptResult = result; releasePendingSend(); @@ -3637,11 +3695,16 @@ export class Session implements SessionContext { errorKind: error.errorKind, }); } + rejectedByLoopProtection = isLoopDetectedTurnError(error); throw error; } finally { const stillOwnsPendingPrompt = this.pendingPrompt === pendingSend; releasePendingSend(); const shouldDrainAutomaticQueues = + // Loop-detected turns resolved end_turn (and drained) before loop + // stops became rejections; keep that invariant on the new path so + // queued cron/notification work is not stranded. + rejectedByLoopProtection || todoStopGuardPreparation.drainSupersededAutomaticQueues || this.todoStopGuardDrainAutomaticQueuesWhenIdle || this.todoStopGuard.blocksUnrelatedAutomaticTurns || @@ -3836,7 +3899,9 @@ export class Session implements SessionContext { responseCapture: AgentResponseCapture, invocationContext?: InvocationContextV1, modelPrompt?: string, + rejectOnLoopDetected = false, goalTurn?: AcpGoalTurn, + channelTurn = false, ): Promise { const sessionId = this.config.getSessionId(); if ( @@ -3860,7 +3925,9 @@ export class Session implements SessionContext { pendingSend, responseCapture, modelPrompt, + rejectOnLoopDetected, goalTurn, + channelTurn, ), ), ); @@ -3874,7 +3941,9 @@ export class Session implements SessionContext { pendingSend: AbortController, responseCapture: AgentResponseCapture, modelPrompt?: string, + rejectOnLoopDetected = false, goalTurn?: AcpGoalTurn, + channelTurn = false, ): Promise { return Storage.runWithRuntimeBaseDir( this.runtimeBaseDir, @@ -4279,9 +4348,7 @@ export class Session implements SessionContext { let nextMessage: Content | null = { role: 'user', parts }; let turnCount = 0; const toolLoopState = createDaemonToolLoopState( - promptMetadata?.[CHANNEL_PROMPT_META_KEY] === true - ? 'off' - : this.repeatedToolFailureGuardMode, + channelTurn ? 'off' : this.repeatedToolFailureGuardMode, ); // conversation_finished must fire on every terminal path of the @@ -4567,13 +4634,17 @@ export class Session implements SessionContext { promptId, toolLoopState, onFullTurnModel, + rejectOnLoopDetected, ); nextMessage = nextAfterTools.message; if (nextAfterTools.stoppedByRepeatedToolFailure) { return { - stopReason: getAbortAwareEndTurnStopReason( - pendingSend.signal, - ), + stopReason: rejectOnLoopDetected + ? cancelledOrThrowLoopDetected( + pendingSend.signal, + toolLoopState, + ) + : getAbortAwareEndTurnStopReason(pendingSend.signal), }; } if (toolRun.loopDetected) { @@ -4583,9 +4654,12 @@ export class Session implements SessionContext { pendingSend.signal, ); return { - stopReason: getAbortAwareEndTurnStopReason( - pendingSend.signal, - ), + stopReason: rejectOnLoopDetected + ? cancelledOrThrowLoopDetected( + pendingSend.signal, + toolLoopState, + ) + : getAbortAwareEndTurnStopReason(pendingSend.signal), }; } } @@ -4606,6 +4680,7 @@ export class Session implements SessionContext { true, fullTurnModelOverride, responseCapture, + rejectOnLoopDetected, ); if (result.stopReason !== 'cancelled') { responseCapture.agentOutput.writeToSpan( @@ -4651,6 +4726,7 @@ export class Session implements SessionContext { allowExternalHooks = true, modelOverride?: string, responseCapture?: AgentResponseCapture, + rejectOnLoopDetected = false, ): Promise<{ stopReason: PromptResponse['stopReason'] }> { const stopHookBlockingCap = this.config.getStopHookBlockingCap(); let stopHookIterationCount = 0; @@ -4716,6 +4792,7 @@ export class Session implements SessionContext { onFullTurnModel, getModelOverride: () => modelOverride, responseCapture, + rejectOnLoopDetected, }, ); if (continuation.kind === 'terminal') { @@ -4815,6 +4892,7 @@ export class Session implements SessionContext { onFullTurnModel, getModelOverride: () => modelOverride, responseCapture, + rejectOnLoopDetected, }, ); if (continuation.kind === 'terminal') { @@ -4949,6 +5027,7 @@ export class Session implements SessionContext { onFullTurnModel, getModelOverride: () => modelOverride, responseCapture, + rejectOnLoopDetected, }, ); if (continuation.supersededAutomaticContinuation && externalReason) { @@ -4974,6 +5053,7 @@ export class Session implements SessionContext { onFullTurnModel?: (model: string) => boolean; getModelOverride?: () => string | undefined; responseCapture?: AgentResponseCapture; + rejectOnLoopDetected?: boolean; } = {}, ): Promise { let nextMessage: Content | null = { role: 'user', parts }; @@ -5526,11 +5606,7 @@ export class Session implements SessionContext { options.onFullTurnModel, ), ); - if ( - toolRun.stopAfterPermissionCancel || - toolRun.loopDetected || - pendingSend.signal.aborted - ) { + if (toolRun.stopAfterPermissionCancel || pendingSend.signal.aborted) { this.todoStopGuard.suspend(); await this.#preserveStoppedToolRun(toolRun, pendingSend.signal); return { @@ -5541,12 +5617,29 @@ export class Session implements SessionContext { : {}), }; } + if (toolRun.loopDetected) { + this.todoStopGuard.suspend(); + await this.#preserveStoppedToolRun(toolRun, pendingSend.signal); + return { + kind: 'terminal', + // Only the foreground chain rejects a loop-detected stop; cron + // and background-notification turns keep the graceful end-turn + // handling they had before loop stops became rejections. + stopReason: options.rejectOnLoopDetected + ? cancelledOrThrowLoopDetected(pendingSend.signal, toolLoopState) + : getAbortAwareEndTurnStopReason(pendingSend.signal), + ...(supersededAutomaticContinuation + ? { supersededAutomaticContinuation: true } + : {}), + }; + } const nextAfterTools = await this.#buildNextMessageAfterToolRun( toolRun, pendingSend.signal, toolPromptId, toolLoopState, options.onFullTurnModel, + options.rejectOnLoopDetected ?? false, ); nextMessage = nextAfterTools.message; if (nextAfterTools.hadMidTurnUserInput) { @@ -5988,6 +6081,7 @@ export class Session implements SessionContext { promptId: string, toolLoopState: DaemonToolLoopState, onFullTurnModel?: (model: string) => boolean, + rejectOnLoopDetected = false, ): Promise { if (toolRun.loopDetected) { debugLogger.debug('Stopping ACP turn after daemon loop detection.'); @@ -6081,14 +6175,19 @@ export class Session implements SessionContext { toolLoopState, { recordToQwenLogger: false }, ); - try { - await this.messageEmitter.emitAgentMessage( - REPEATED_TOOL_FAILURE_STOP_MESSAGE, - ); - } catch (error) { - debugLogger.warn( - `Failed to emit repeated tool failure stop message: ${this.#formatError(error)}`, - ); + if (!rejectOnLoopDetected) { + // Rejecting turns publish the structured turn_error as the + // user-visible explanation; graceful (non-interactive) stops have + // no replacement, so keep the transcript stop message for them. + try { + await this.messageEmitter.emitAgentMessage( + REPEATED_TOOL_FAILURE_STOP_MESSAGE, + ); + } catch (error) { + debugLogger.warn( + `Failed to emit repeated tool failure stop message: ${this.#formatError(error)}`, + ); + } } return { message: null, diff --git a/packages/cli/src/commands/channel/channel-prompt-wire-key.test.ts b/packages/cli/src/commands/channel/channel-prompt-wire-key.test.ts new file mode 100644 index 00000000000..bba61f35de4 --- /dev/null +++ b/packages/cli/src/commands/channel/channel-prompt-wire-key.test.ts @@ -0,0 +1,13 @@ +import { describe, expect, it } from 'vitest'; +import { CHANNEL_PROMPT_META_KEY } from '@qwen-code/channel-base'; +import { CHANNEL_PROMPT_META_KEY as BRIDGE_CHANNEL_PROMPT_META_KEY } from '@qwen-code/acp-bridge/bridgeTypes'; + +// The channel bridges write the channel-turn classification under the +// channel-base key and the daemon-side strip/re-injection reads it under +// the acp-bridge key; the packages have no dependency path between them, +// so pin the wire contract here where both packages are importable. +describe('channel prompt classification wire key', () => { + it('is identical across channel-base and acp-bridge', () => { + expect(CHANNEL_PROMPT_META_KEY).toBe(BRIDGE_CHANNEL_PROMPT_META_KEY); + }); +}); diff --git a/packages/cli/src/serve/routes/session.ts b/packages/cli/src/serve/routes/session.ts index e382d6ec7a1..f9dbed0b88b 100644 --- a/packages/cli/src/serve/routes/session.ts +++ b/packages/cli/src/serve/routes/session.ts @@ -31,7 +31,10 @@ import { type SessionArchiveState, } from '@qwen-code/qwen-code-core'; import type { SessionArtifactInput } from '@qwen-code/acp-bridge/sessionArtifacts'; -import { DAEMON_PROMPT_DISPLAY_TEXT_META_KEY } from '@qwen-code/acp-bridge/bridgeTypes'; +import { + CHANNEL_PROMPT_META_KEY, + DAEMON_PROMPT_DISPLAY_TEXT_META_KEY, +} from '@qwen-code/acp-bridge/bridgeTypes'; import { parseSessionSource } from '@qwen-code/acp-bridge'; import { isReservedLiveSessionSource, @@ -3386,23 +3389,31 @@ export function registerSessionRoutes( forwardedMeta?.[CHANNEL_WORKER_PROMPT_AUTHORIZATION_META_KEY]; const promptDisplayText = forwardedMeta?.[DAEMON_PROMPT_DISPLAY_TEXT_META_KEY]; + const channelPrompt = forwardedMeta?.[CHANNEL_PROMPT_META_KEY]; if (forwardedMeta) { delete forwardedMeta[CHANNEL_WORKER_PROMPT_AUTHORIZATION_META_KEY]; delete forwardedMeta[DAEMON_PROMPT_DISPLAY_TEXT_META_KEY]; + delete forwardedMeta[CHANNEL_PROMPT_META_KEY]; if (Object.keys(forwardedMeta).length > 0) { forwardedBody['_meta'] = forwardedMeta; } else { delete forwardedBody['_meta']; } } + const channelWorkerAuthorized = isChannelWorkerPromptAuthorized( + promptAuthorization, + runtime.workspaceCwd, + ); const trustedPromptDisplayText = - typeof promptDisplayText === 'string' && - isChannelWorkerPromptAuthorized( - promptAuthorization, - runtime.workspaceCwd, - ) + typeof promptDisplayText === 'string' && channelWorkerAuthorized ? promptDisplayText : undefined; + // Channel classification opts the turn out of loop-detected + // rejection, so it rides the same worker authorization as the + // display projection; a forged key from any other caller is dropped + // here and again at the bridge admission strip. + const trustedChannelPrompt = + channelWorkerAuthorized && channelPrompt === true; const lastEventId = ownerBridge.getSessionLastEventId(sessionId); // Epoch token paired with the cursor above: a client that seeds its @@ -3452,6 +3463,7 @@ export function registerSessionRoutes( ...(trustedPromptDisplayText !== undefined ? { promptDisplayText: trustedPromptDisplayText } : {}), + ...(trustedChannelPrompt ? { channelPrompt: true } : {}), ...(delivery !== undefined ? { channelDelivery: { diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index 2b8a22e8785..a8fc7828d0b 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -44,7 +44,10 @@ import { registerChannelWorkerPromptAuthorization, revokeChannelWorkerPromptAuthorization, } from './channel-worker-prompt-authorization.js'; -import { CHANNEL_PROMPT_DISPLAY_TEXT_META_KEY } from '@qwen-code/channel-base'; +import { + CHANNEL_PROMPT_DISPLAY_TEXT_META_KEY, + CHANNEL_PROMPT_META_KEY, +} from '@qwen-code/channel-base'; import { resolveWebShellDir, isDocumentNavigation, @@ -12651,6 +12654,61 @@ describe('createServeApp', () => { } }); + it('accepts channel-prompt classification only from the workspace worker', async () => { + // `qwen.channel.prompt` opts a turn out of loop-detected rejection; + // a forged key from an unauthorized caller must be dropped at the + // route (and again at the bridge admission strip), never reaching + // the trusted prompt context. + const bridge = fakeBridge(); + const app = createServeApp(baseOpts, undefined, { bridge }); + const workspace = realpathSync(process.cwd()); + const token = 'channel-worker-classification-token'; + registerChannelWorkerPromptAuthorization(token, workspace); + try { + const forged = await request(app) + .post('/session/session-A/prompt') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ + prompt: [{ type: 'text', text: 'hi' }], + _meta: { [CHANNEL_PROMPT_META_KEY]: true }, + }); + const forgedWithBadToken = await request(app) + .post('/session/session-A/prompt') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ + prompt: [{ type: 'text', text: 'hi' }], + _meta: { + [CHANNEL_WORKER_PROMPT_AUTHORIZATION_META_KEY]: 'forged', + [CHANNEL_PROMPT_META_KEY]: true, + }, + }); + const trusted = await request(app) + .post('/session/session-A/prompt') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ + prompt: [{ type: 'text', text: 'hi' }], + _meta: { + [CHANNEL_WORKER_PROMPT_AUTHORIZATION_META_KEY]: token, + [CHANNEL_PROMPT_META_KEY]: true, + }, + }); + + expect(forged.status).toBe(202); + expect(forgedWithBadToken.status).toBe(202); + expect(trusted.status).toBe(202); + expect(bridge.promptCalls[0]?.context?.channelPrompt).toBeUndefined(); + expect(bridge.promptCalls[1]?.context?.channelPrompt).toBeUndefined(); + expect(bridge.promptCalls[2]?.context?.channelPrompt).toBe(true); + for (const call of bridge.promptCalls) { + expect(call.req._meta ?? {}).not.toHaveProperty( + CHANNEL_PROMPT_META_KEY, + ); + } + } finally { + revokeChannelWorkerPromptAuthorization(token); + } + }); + it('validates delivery and forwards it only through trusted prompt context', async () => { const bridge = fakeBridge(); const channelDeliveryAuthorizations = diff --git a/packages/sdk-typescript/src/daemon/events.ts b/packages/sdk-typescript/src/daemon/events.ts index 5a3a3ac5e54..3614fe6d15f 100644 --- a/packages/sdk-typescript/src/daemon/events.ts +++ b/packages/sdk-typescript/src/daemon/events.ts @@ -822,6 +822,7 @@ export interface DaemonTurnErrorData { message: string; code?: string; errorKind?: DaemonErrorKind | (string & {}); + loopType?: string; promptId?: string; [key: string]: unknown; } diff --git a/packages/sdk-typescript/src/daemon/types.ts b/packages/sdk-typescript/src/daemon/types.ts index 1222ac68906..9c3fda413c4 100644 --- a/packages/sdk-typescript/src/daemon/types.ts +++ b/packages/sdk-typescript/src/daemon/types.ts @@ -1463,6 +1463,8 @@ export const DAEMON_ERROR_KINDS = [ 'writer_idle_timeout', // The model response stream ended before a complete turn could be read. 'model_stream_interrupted', + // Tool-call loop protection stopped the current turn. + 'loop_detected', ] as const; export type DaemonErrorKind = (typeof DAEMON_ERROR_KINDS)[number]; diff --git a/packages/web-shell/client/App.test.tsx b/packages/web-shell/client/App.test.tsx index 23e13439983..37772c7d407 100644 --- a/packages/web-shell/client/App.test.tsx +++ b/packages/web-shell/client/App.test.tsx @@ -12232,6 +12232,384 @@ describe('App session callbacks', () => { }); }); + it('asks for a new instruction instead of retrying a loop-detected turn', async () => { + const { container, rerender } = renderApp(); + await flush(); + + testState.prompt = 'repeat this'; + await clickSubmit(container); + + act(() => { + testState.blocks = [ + { + kind: 'error', + source: 'turn_error', + id: 'turn-error-loop', + errorKind: 'loop_detected', + text: 'internal fallback', + }, + ]; + rerender(); + }); + + expect(container.querySelector('[data-testid="retry"]')).toBeNull(); + expect(testState.latestChatEditorProps?.disabled).toBe(false); + }); + + it('still reports a loop-detected turn error through turn_complete', async () => { + const onSessionChange = vi.fn(); + const { container, rerender } = renderApp({ onSessionChange }); + await flush(); + + testState.prompt = 'repeat this'; + await clickSubmit(container); + onSessionChange.mockClear(); + + act(() => { + testState.streamingState = 'responding'; + rerender({ onSessionChange }); + }); + act(() => { + testState.blocks = [ + { + kind: 'error', + source: 'turn_error', + id: 'turn-error-loop', + errorKind: 'loop_detected', + text: 'internal fallback', + }, + ]; + testState.streamingState = 'idle'; + rerender({ onSessionChange }); + }); + + expect(onSessionChange).toHaveBeenCalledWith({ + type: 'turn_complete', + sessionId: 'session-1', + error: expect.objectContaining({ + message: 'Turn error (block turn-error-loop)', + }), + }); + expect(container.querySelector('[data-testid="retry"]')).toBeNull(); + }); + + it('reports the turn error through turn_complete across a trailing background notification', async () => { + // turn_complete and the retry decision read the same backward walk, so + // a background-notification user block after the turn error must not + // hide the error from the host while the UI still offers retry. + const onSessionChange = vi.fn(); + const { container, rerender } = renderApp({ onSessionChange }); + await flush(); + + testState.prompt = 'interrupt this stream'; + await clickSubmit(container); + onSessionChange.mockClear(); + + act(() => { + testState.streamingState = 'responding'; + rerender({ onSessionChange }); + }); + act(() => { + testState.blocks = [ + { + kind: 'error', + source: 'turn_error', + id: 'turn-error-with-notification', + errorKind: 'model_stream_interrupted', + text: 'terminated', + }, + { + id: 'background-1', + kind: 'user', + text: 'Background task completed', + meta: { source: 'background_notification' }, + }, + ]; + testState.streamingState = 'idle'; + rerender({ onSessionChange }); + }); + + expect(onSessionChange).toHaveBeenCalledWith({ + type: 'turn_complete', + sessionId: 'session-1', + error: expect.objectContaining({ + message: 'Turn error (block turn-error-with-notification)', + }), + }); + expect(container.querySelector('[data-testid="retry"]')).not.toBeNull(); + }); + + it('does not rearm a retry when the retried turn is loop-stopped', async () => { + // When the retried turn itself is stopped by loop protection, the + // catch path must not arm retry state on the loop error: Ctrl+Y + // calls handleRetry() directly even while the retry button is + // hidden, and resubmitting the stopped prompt tends to re-loop. + const retrySend = deferred(); + mockSessionActions.sendPrompt + .mockResolvedValueOnce(undefined) + .mockReturnValueOnce(retrySend.promise); + const { container, rerender } = renderApp(); + await flush(); + + testState.prompt = 'repeat this'; + await clickSubmit(container); + act(() => { + testState.blocks = [ + { + kind: 'error', + source: 'turn_error', + id: 'turn-error-1', + promptId: 'prompt-1', + }, + ]; + rerender(); + }); + expect(container.querySelector('[data-testid="retry"]')).not.toBeNull(); + + act(() => { + container + .querySelector('[data-testid="retry"]') + ?.click(); + }); + await vi.waitFor(() => { + expect(mockSessionActions.sendPrompt).toHaveBeenCalledTimes(2); + }); + const retryOptions = mockSessionActions.sendPrompt.mock.calls[1]?.[1]; + + // The loop turn_error lands before the rejection settles, so the + // catch walk already sees it when the re-arm runs. + act(() => { + testState.blocks = [ + { + kind: 'error', + source: 'turn_error', + id: 'turn-error-1', + promptId: 'prompt-1', + }, + { + kind: 'error', + source: 'turn_error', + id: 'turn-error-loop', + promptId: 'prompt-2', + errorKind: 'loop_detected', + text: 'internal fallback', + }, + ]; + rerender(); + }); + act(() => { + retryOptions?.onAdmissionStarted?.(); + retryOptions?.onAdmitted?.(); + }); + + await act(async () => { + retrySend.reject( + Object.assign(new Error('loop protection stopped the turn'), { + _daemonTurnError: true, + body: 'LOOP_DETECTED', + }), + ); + await Promise.resolve(); + }); + await flush(); + + expect(container.querySelector('[data-testid="retry"]')).toBeNull(); + await act(async () => { + window.dispatchEvent( + new KeyboardEvent('keydown', { key: 'y', ctrlKey: true }), + ); + await Promise.resolve(); + }); + expect(mockSessionActions.sendPrompt).toHaveBeenCalledTimes(2); + }); + + it('does not reoffer a loop-stopped retry to a later unrelated turn error', async () => { + // The rejection settles before the loop turn_error block commits + // (microtask vs transcript flush), so the catch walk still sees the + // original error. The stashed prompt must not survive the loop stop + // and be consumed by a later unrelated retryable turn error, which + // would resubmit the loop-stopped prompt misattributed to a turn + // the user never submitted. + const retrySend = deferred(); + mockSessionActions.sendPrompt + .mockResolvedValueOnce(undefined) + .mockReturnValueOnce(retrySend.promise) + .mockResolvedValueOnce(undefined); + const { container, rerender } = renderApp(); + await flush(); + + testState.prompt = 'repeat this'; + await clickSubmit(container); + act(() => { + testState.blocks = [ + { + kind: 'error', + source: 'turn_error', + id: 'turn-error-1', + promptId: 'prompt-1', + }, + ]; + rerender(); + }); + expect(container.querySelector('[data-testid="retry"]')).not.toBeNull(); + + act(() => { + container + .querySelector('[data-testid="retry"]') + ?.click(); + }); + await vi.waitFor(() => { + expect(mockSessionActions.sendPrompt).toHaveBeenCalledTimes(2); + }); + const retryOptions = mockSessionActions.sendPrompt.mock.calls[1]?.[1]; + act(() => { + retryOptions?.onAdmissionStarted?.(); + retryOptions?.onAdmitted?.(); + }); + + await act(async () => { + retrySend.reject( + Object.assign(new Error('loop protection stopped the turn'), { + _daemonTurnError: true, + body: 'LOOP_DETECTED', + }), + ); + await Promise.resolve(); + }); + await flush(); + + act(() => { + testState.blocks = [ + { + kind: 'error', + source: 'turn_error', + id: 'turn-error-1', + promptId: 'prompt-1', + }, + { + kind: 'error', + source: 'turn_error', + id: 'turn-error-loop', + promptId: 'prompt-2', + errorKind: 'loop_detected', + text: 'internal fallback', + }, + ]; + rerender(); + }); + await flush(); + expect(container.querySelector('[data-testid="retry"]')).toBeNull(); + + act(() => { + testState.blocks = [ + { + kind: 'error', + source: 'turn_error', + id: 'turn-error-1', + promptId: 'prompt-1', + }, + { + kind: 'error', + source: 'turn_error', + id: 'turn-error-loop', + promptId: 'prompt-2', + errorKind: 'loop_detected', + text: 'internal fallback', + }, + { + kind: 'error', + source: 'turn_error', + id: 'turn-error-3', + promptId: 'prompt-3', + }, + ]; + rerender(); + }); + await flush(); + + expect(container.querySelector('[data-testid="retry"]')).toBeNull(); + await act(async () => { + window.dispatchEvent( + new KeyboardEvent('keydown', { key: 'y', ctrlKey: true }), + ); + await Promise.resolve(); + }); + expect(mockSessionActions.sendPrompt).toHaveBeenCalledTimes(2); + }); + + it('does not report the previous turn error again when a retry settles without content', async () => { + // The retry turn settles while the transcript still ends with the + // original turn error (settle precedes the transcript flush); the + // turn_complete for that turn must not re-report the error the user + // already retried. + const onSessionChange = vi.fn(); + const retrySend = deferred(); + mockSessionActions.sendPrompt + .mockResolvedValueOnce(undefined) + .mockReturnValueOnce(retrySend.promise); + const { container, rerender } = renderApp({ onSessionChange }); + await flush(); + + testState.prompt = 'recover this stream'; + await clickSubmit(container); + onSessionChange.mockClear(); + + act(() => { + testState.streamingState = 'responding'; + rerender({ onSessionChange }); + }); + act(() => { + testState.blocks = [ + { + kind: 'error', + source: 'turn_error', + id: 'turn-error-1', + promptId: 'prompt-1', + }, + ]; + testState.streamingState = 'idle'; + rerender({ onSessionChange }); + }); + expect(onSessionChange).toHaveBeenCalledWith({ + type: 'turn_complete', + sessionId: 'session-1', + error: expect.objectContaining({ + message: 'Turn error (block turn-error-1)', + }), + }); + expect(container.querySelector('[data-testid="retry"]')).not.toBeNull(); + + act(() => { + container + .querySelector('[data-testid="retry"]') + ?.click(); + }); + await vi.waitFor(() => { + expect(mockSessionActions.sendPrompt).toHaveBeenCalledTimes(2); + }); + await act(async () => { + retrySend.resolve(); + await Promise.resolve(); + }); + await flush(); + + act(() => { + testState.streamingState = 'responding'; + rerender({ onSessionChange }); + }); + onSessionChange.mockClear(); + act(() => { + testState.streamingState = 'idle'; + rerender({ onSessionChange }); + }); + + expect(onSessionChange).toHaveBeenCalledWith({ + type: 'turn_complete', + sessionId: 'session-1', + error: undefined, + }); + }); + it.each([ ['a fresh prompt id', 'prompt-2'], ['a reused prompt id', 'prompt-1'], diff --git a/packages/web-shell/client/App.tsx b/packages/web-shell/client/App.tsx index fb637eecb66..66996abf160 100644 --- a/packages/web-shell/client/App.tsx +++ b/packages/web-shell/client/App.tsx @@ -55,6 +55,7 @@ import { WEB_SHELL_SIDE_TASK_SOURCE_TYPE, } from './constants/sessions'; import { extractPendingPermission } from './adapters/transcriptAdapter'; +import { isRetryableTurnErrorKind } from './adapters/transcriptToMessages'; import { MessageList, type MessageListHandle } from './components/MessageList'; import { SubagentDetailsProvider } from './subagentDetailsContext'; import { MonitorDetailsProvider } from './monitorDetailsContext'; @@ -4122,6 +4123,7 @@ export function App({ composerSourceVersionRef.current, ); const retryableTurnErrorIdRef = useRef(null); + const lastTurnErrorIdRef = useRef(null); const retryableTurnErrorIdentityRef = useRef< TranscriptTurnErrorIdentity | undefined >(undefined); @@ -7092,7 +7094,16 @@ export function App({ ]); useEffect(() => { - const retryableTurnError = getRetryableTurnError(blocks); + const lastTurnError = getRetryableTurnError(blocks); + // Loop-detected turn errors still surface through turn_complete below, + // but resubmitting a prompt the daemon stopped for loop protection + // tends to re-loop, so no retry affordance is offered for them. + const retryableTurnError = + lastTurnError && + lastTurnError.kind === 'error' && + isRetryableTurnErrorKind(lastTurnError.errorKind) + ? lastTurnError + : undefined; if (retryableTurnError) { rearmFailedTurnErrorRetry(retryableTurnError, blocks); } @@ -7109,6 +7120,16 @@ export function App({ ) { retriedTurnErrorIdRef.current = retryableTurnError.id; } + // Same walk as the retry decision above, so turn_complete and the + // retry affordance never disagree about whether the current turn has + // a turn error (e.g. across a trailing background notification). An + // error the user already retried stays suppressed, mirroring the + // retry affordance; loop-detected errors are never retried, so they + // always surface. + lastTurnErrorIdRef.current = + lastTurnError && lastTurnError.id !== retriedTurnErrorIdRef.current + ? lastTurnError.id + : null; const canRetry = connected && retryableTurnError !== undefined && @@ -7138,7 +7159,7 @@ export function App({ onStreamingStateChange?.(streamingState); }, [streamingState, onStreamingStateChange]); - // Reads retryableTurnErrorIdRef which is set by the blocks effect above. + // Reads lastTurnErrorIdRef which is set by the blocks effect above. // Declaration order matters: this effect must run after the blocks effect // so that within the same render, the ref is already updated before we read it. const prevStreamingForTurnCompleteRef = useRef(streamingState); @@ -7172,8 +7193,8 @@ export function App({ return; } const turnError = - retryableTurnErrorIdRef.current != null - ? new Error(`Turn error (block ${retryableTurnErrorIdRef.current})`) + lastTurnErrorIdRef.current != null + ? new Error(`Turn error (block ${lastTurnErrorIdRef.current})`) : undefined; if (workspaceCwd) { sessionCatalogController.turnCompleted(workspaceCwd); @@ -9903,17 +9924,27 @@ export function App({ }); } if (isDaemonTurnError(error)) { - failedTurnErrorRetryRef.current = { - errorId: retryErrorId, - text: retryText, - images: retryImages, - inputAnnotations: retryInputAnnotations, - owner: retryOwner, - }; + // A loop-detected rejection ends the retry lineage: the + // retried turn itself was stopped for loop protection, so + // the stashed prompt must not be re-offered — resubmitting + // it tends to re-loop. + if (error.body !== 'LOOP_DETECTED') { + failedTurnErrorRetryRef.current = { + errorId: retryErrorId, + text: retryText, + images: retryImages, + inputAnnotations: retryInputAnnotations, + owner: retryOwner, + }; + } const nextTurnError = getRetryableTurnError( store.getSnapshot().blocks, ); - if (nextTurnError) { + if ( + nextTurnError && + nextTurnError.kind === 'error' && + isRetryableTurnErrorKind(nextTurnError.errorKind) + ) { rearmFailedTurnErrorRetry( nextTurnError, store.getSnapshot().blocks, diff --git a/packages/web-shell/client/adapters/transcriptToMessages.test.ts b/packages/web-shell/client/adapters/transcriptToMessages.test.ts index 83ee4addb67..fd60c467a84 100644 --- a/packages/web-shell/client/adapters/transcriptToMessages.test.ts +++ b/packages/web-shell/client/adapters/transcriptToMessages.test.ts @@ -2733,6 +2733,30 @@ describe('transcriptBlocksToDaemonMessages', () => { ]); }); + it('renders loop detection errors from a structured localized label', () => { + const messages = transcriptBlocksToDaemonMessages( + [ + { + id: 'err-loop', + kind: 'error' as const, + source: 'turn_error' as const, + errorKind: 'loop_detected' as const, + text: 'internal fallback', + clientReceivedAt: 1, + createdAt: 1, + updatedAt: 1, + }, + ], + { labels: { loopDetected: 'Localized loop guidance.' } }, + ); + + expect(messages[0]).toMatchObject({ + content: 'Localized loop guidance.', + retryable: false, + source: 'turn_error', + }); + }); + it('renders model stream interruption errors from structured errorKind labels', () => { const messages = transcriptBlocksToDaemonMessages( [ diff --git a/packages/web-shell/client/adapters/transcriptToMessages.ts b/packages/web-shell/client/adapters/transcriptToMessages.ts index 3b885dbadeb..9b522092ef1 100644 --- a/packages/web-shell/client/adapters/transcriptToMessages.ts +++ b/packages/web-shell/client/adapters/transcriptToMessages.ts @@ -46,6 +46,7 @@ interface TranscriptMessageLabels { branchSuccess?: (name: string) => string; midTurnInserted?: (message: string) => string; modelStreamInterrupted?: string; + loopDetected?: string; } interface TranscriptMessageOptions { @@ -191,10 +192,21 @@ function isUnrecognizedDaemonDebug( ); } +// Resubmitting a prompt the daemon stopped for loop protection tends to +// re-loop, so no retry affordance is offered for these turn errors. +export function isRetryableTurnErrorKind( + errorKind: string | undefined, +): boolean { + return errorKind !== 'loop_detected'; +} + function getErrorDisplayText( block: DaemonStatusTranscriptBlock, labels?: TranscriptMessageLabels, ): string { + if (block.errorKind === 'loop_detected') { + return labels?.loopDetected ?? block.text; + } if ( block.errorKind === 'model_stream_interrupted' || // Older daemons emit this turn_error before they know about errorKind. @@ -775,7 +787,9 @@ export function transcriptBlocksToDaemonMessages( role: 'system', content: getErrorDisplayText(errorBlock, options.labels), variant: 'error', - retryable: errorBlock.source === 'turn_error', + retryable: + errorBlock.source === 'turn_error' && + isRetryableTurnErrorKind(errorKind), timestamp: blockTime, ...(errorBlock.source ? { source: errorBlock.source } : {}), ...getErrorMessageData(errorBlock.data, errorKind), diff --git a/packages/web-shell/client/hooks/useMessages.test.ts b/packages/web-shell/client/hooks/useMessages.test.ts index 6e03fb68ae6..806a97e21fe 100644 --- a/packages/web-shell/client/hooks/useMessages.test.ts +++ b/packages/web-shell/client/hooks/useMessages.test.ts @@ -77,12 +77,22 @@ describe('transcriptBlocksToLocalizedMessages', () => { DaemonStatusTranscriptBlock, 'clientReceivedAt' | 'createdAt' | 'updatedAt' >), + baseBlock({ + id: 'loop', + kind: 'error', + text: 'internal fallback', + errorKind: 'loop_detected', + } as Omit< + DaemonStatusTranscriptBlock, + 'clientReceivedAt' | 'createdAt' | 'updatedAt' + >), ]; expect(transcriptBlocksToLocalizedMessages(blocks, t)).toMatchObject([ { content: 'localized:request.cancelled' }, { content: 'branch.success:review' }, { content: 'localized:error.modelStreamInterrupted' }, + { content: 'localized:error.loopDetected' }, ]); }); }); diff --git a/packages/web-shell/client/hooks/useMessages.ts b/packages/web-shell/client/hooks/useMessages.ts index 64e07d95919..0c7dfd060f5 100644 --- a/packages/web-shell/client/hooks/useMessages.ts +++ b/packages/web-shell/client/hooks/useMessages.ts @@ -32,6 +32,7 @@ export function transcriptBlocksToLocalizedMessages( branchSuccess: (name) => t('branch.success', { name }), midTurnInserted: (message) => t('midTurn.inserted', { message }), modelStreamInterrupted: t('error.modelStreamInterrupted'), + loopDetected: t('error.loopDetected'), }, }); } diff --git a/packages/web-shell/client/i18n.tsx b/packages/web-shell/client/i18n.tsx index 2caba1221aa..546e869cfc9 100644 --- a/packages/web-shell/client/i18n.tsx +++ b/packages/web-shell/client/i18n.tsx @@ -1511,6 +1511,8 @@ const EN: Messages = { 'error.unknown': 'Unknown error', 'error.modelStreamInterrupted': 'Model response stream was interrupted. Please retry.', + 'error.loopDetected': + 'The model got stuck while using tools or reached a safety limit, so this turn was stopped. Your session is still open—try a more specific instruction to continue.', 'shell.command': 'Shell Command', 'compact.enabled': 'Compact mode enabled', 'compact.disabled': 'Compact mode disabled', @@ -4353,6 +4355,8 @@ const ZH: Messages = { 'clear.blocked': '流式输出中无法清屏 — 先按 Esc 取消。', 'error.unknown': '未知错误', 'error.modelStreamInterrupted': '模型响应流已中断，请重试。', + 'error.loopDetected': + '模型在调用工具时反复尝试或达到了安全上限，因此系统停止了本轮操作。会话并未结束，你可以换一个更明确的指令继续。', 'shell.command': 'Shell 命令', 'compact.enabled': '紧凑模式已开启', 'compact.disabled': '紧凑模式已关闭',