diff --git a/packages/cli/src/acp-integration/acpAgent.test.ts b/packages/cli/src/acp-integration/acpAgent.test.ts index 6ad752811cf..09a36eb6269 100644 --- a/packages/cli/src/acp-integration/acpAgent.test.ts +++ b/packages/cli/src/acp-integration/acpAgent.test.ts @@ -11645,11 +11645,14 @@ describe('QwenAgent MCP SSE/HTTP support', () => { const VALID_SESSION_ID = '12345678-1234-1234-1234-1234567890ab'; - function mockSessionServiceLoad(result: unknown) { + function mockSessionServiceLoad(result: unknown, onRead?: () => void) { vi.mocked(SessionService).mockImplementation( () => ({ - loadSession: vi.fn().mockResolvedValue(result), + loadSession: vi.fn().mockImplementation(async () => { + onRead?.(); + return result; + }), }) as unknown as InstanceType, ); } @@ -11736,6 +11739,9 @@ describe('QwenAgent MCP SSE/HTTP support', () => { expect.anything(), expect.anything(), gaps, + // Non-live loadUpdates replays have no live stream to deliver a + // trailing result, so dangling calls must still finalize. + expect.objectContaining({ finalizeDangling: true }), ); mockConnectionState.resolve(); @@ -12052,6 +12058,202 @@ describe('QwenAgent MCP SSE/HTTP support', () => { await agentPromise; }); + it('live session load finalizes dangling calls because the restore gate drains active turns', async () => { + const innerConfig = await setupSessionMocks(VALID_SESSION_ID); + innerConfig.getSessionRuntimeBaseDir = vi + .fn() + .mockReturnValue('/tmp/qwen-runtime-test'); + vi.mocked(SessionService).mockImplementation( + () => + ({ + readLiveRestoreProjection: vi.fn().mockResolvedValue({ + replay: { + records: [{ role: 'user' }], + gaps: [], + }, + }), + }) as unknown as InstanceType, + ); + mockHistoryReplay.mockResolvedValue(undefined); + const { agent, agentPromise } = await bootAcpAgent(); + const loadSession = (params: Record) => + ( + agent as unknown as { + loadSession: (p: Record) => Promise; + } + ).loadSession(params); + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + // The restore gate drains active turns and blocks new ones before the + // replay runs, so the live path must finalize regardless of turn state; + // sampling isTurnIdle() under the gate is structurally false and would + // keep genuinely abandoned calls pending forever. + lastSessionMock!.isTurnIdle.mockReturnValue(false); + + await loadSession({ + cwd: '/tmp', + sessionId: VALID_SESSION_ID, + mcpServers: [], + }); + expect(mockHistoryReplay).toHaveBeenLastCalledWith( + expect.anything(), + expect.anything(), + expect.anything(), + expect.objectContaining({ finalizeDangling: true }), + ); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('qwen/session/loadUpdates keeps dangling transcript calls in flight while a prompt is active', async () => { + const innerConfig = await setupSessionMocks(VALID_SESSION_ID); + innerConfig.getSessionRuntimeBaseDir = vi + .fn() + .mockReturnValue('/tmp/qwen-runtime-test'); + mockSessionServiceLoad({ + conversation: { + messages: [{ role: 'user' }], + startTime: 'start', + lastUpdated: 'end', + }, + }); + mockHistoryReplay.mockResolvedValue(undefined); + const { agent, agentPromise } = await bootAcpAgent(); + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + lastSessionMock!.isTurnIdle.mockReturnValue(false); + let finishPrompt: ((value: unknown) => void) | undefined; + lastSessionMock!.prompt.mockImplementation( + () => + new Promise((resolve) => { + finishPrompt = resolve; + }), + ); + + const prompt = agent.prompt({ sessionId: VALID_SESSION_ID, prompt: [] }); + await vi.waitFor(() => expect(lastSessionMock!.prompt).toHaveBeenCalled()); + + await agent.extMethod('qwen/session/loadUpdates', { + sessionId: VALID_SESSION_ID, + cwd: '/tmp', + }); + expect(mockHistoryReplay).toHaveBeenLastCalledWith( + expect.anything(), + expect.anything(), + undefined, + expect.objectContaining({ finalizeDangling: false }), + ); + // The diagnostic must be a single interpolated string: debugLogger does + // no printf substitution, so %s placeholders would ship unexpanded. + expect(mockDebugLogger.debug).toHaveBeenCalledWith( + expect.stringContaining( + '[ACP] restore replay finalizeDangling=false (idleBeforeRead=false, idleAtReplay=false)', + ), + ); + + finishPrompt?.({ stopReason: 'end_turn' }); + await prompt; + lastSessionMock!.isTurnIdle.mockReturnValue(true); + + await agent.extMethod('qwen/session/loadUpdates', { + sessionId: VALID_SESSION_ID, + cwd: '/tmp', + }); + expect(mockHistoryReplay).toHaveBeenLastCalledWith( + expect.anything(), + expect.anything(), + undefined, + expect.objectContaining({ finalizeDangling: true }), + ); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('qwen/session/loadUpdates keeps a dangling call pending when a turn settles during the read', async () => { + const innerConfig = await setupSessionMocks(VALID_SESSION_ID); + innerConfig.getSessionRuntimeBaseDir = vi + .fn() + .mockReturnValue('/tmp/qwen-runtime-test'); + // Anchor the settle flip to the read boundary: the session is active + // before the read and the mocked sessionService.loadSession flips it to + // idle inside the read window (mirrors the sessionTranscript test that + // toggles state inside readPage.mockImplementationOnce). Keying the + // isTurnIdle mock by call order instead would let a mutation that moves + // the before-read sample across the read survive. + mockSessionServiceLoad( + { + conversation: { + messages: [{ role: 'user' }], + startTime: 'start', + lastUpdated: 'end', + }, + }, + () => lastSessionMock!.isTurnIdle.mockReturnValue(true), + ); + mockHistoryReplay.mockResolvedValue(undefined); + const { agent, agentPromise } = await bootAcpAgent(); + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + // Active before the read, idle by replay time: the before-read sample + // alone must keep the trailing call pending. + lastSessionMock!.isTurnIdle.mockReturnValue(false); + + await agent.extMethod('qwen/session/loadUpdates', { + sessionId: VALID_SESSION_ID, + cwd: '/tmp', + }); + expect(mockHistoryReplay).toHaveBeenLastCalledWith( + expect.anything(), + expect.anything(), + undefined, + expect.objectContaining({ finalizeDangling: false }), + ); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('qwen/session/loadUpdates keeps a dangling call pending when a turn starts during the read', async () => { + const innerConfig = await setupSessionMocks(VALID_SESSION_ID); + innerConfig.getSessionRuntimeBaseDir = vi + .fn() + .mockReturnValue('/tmp/qwen-runtime-test'); + // Anchor the start flip to the read boundary: the session is idle + // before the read and the mocked sessionService.loadSession flips it to + // active inside the read window. Keying the isTurnIdle mock by call + // order instead would let a mutation that moves the replay-time sample + // across the read survive. + mockSessionServiceLoad( + { + conversation: { + messages: [{ role: 'user' }], + startTime: 'start', + lastUpdated: 'end', + }, + }, + () => lastSessionMock!.isTurnIdle.mockReturnValue(false), + ); + mockHistoryReplay.mockResolvedValue(undefined); + const { agent, agentPromise } = await bootAcpAgent(); + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + // Idle before the read, active by replay time: the replay-time sample + // alone must keep the trailing call pending. + lastSessionMock!.isTurnIdle.mockReturnValue(true); + + await agent.extMethod('qwen/session/loadUpdates', { + sessionId: VALID_SESSION_ID, + cwd: '/tmp', + }); + expect(mockHistoryReplay).toHaveBeenLastCalledWith( + expect.anything(), + expect.anything(), + undefined, + expect.objectContaining({ finalizeDangling: false }), + ); + + mockConnectionState.resolve(); + await agentPromise; + }); + it('disposes a pending transcript config superseded by newer settings', async () => { const oldSettings = makeCoreSettings('English'); const newSettings = makeCoreSettings('Japanese'); @@ -16908,6 +17110,7 @@ describe('QwenAgent loadSession / unstable_resumeSession', () => { waitForActiveTurnsToSettle: vi.fn().mockResolvedValue(undefined), cancelPendingPrompt: vi.fn().mockResolvedValue(undefined), assertCanStartTurn: vi.fn().mockResolvedValue(undefined), + isTurnIdle: vi.fn().mockReturnValue(true), sendUpdate: opts.recoveredGoalSendError ? vi.fn().mockRejectedValue(opts.recoveredGoalSendError) : vi.fn().mockResolvedValue(undefined), @@ -18028,6 +18231,7 @@ describe('QwenAgent loadSession / unstable_resumeSession', () => { assertCanStartTurn: vi.fn().mockResolvedValue(undefined), beginClose: vi.fn().mockReturnValue(vi.fn()), waitForActiveTurnsToSettle: vi.fn().mockResolvedValue(undefined), + isTurnIdle: vi.fn().mockReturnValue(true), sendUpdate: vi.fn().mockResolvedValue(undefined), clearActiveTodoPlanRevision: vi.fn(), }; @@ -19040,6 +19244,7 @@ describe('QwenAgent loadSession / unstable_resumeSession', () => { waitForCloseGateToRelease: vi.fn().mockResolvedValue(undefined), cancelPendingPrompt: vi.fn().mockResolvedValue(undefined), waitForActiveTurnsToSettle: vi.fn().mockResolvedValue(undefined), + isTurnIdle: vi.fn().mockReturnValue(true), dispose: replacementDispose, } as unknown as InstanceType; const sessions = ( @@ -19355,6 +19560,7 @@ describe('QwenAgent loadSession / unstable_resumeSession', () => { }); expect(replayOptions).toEqual({ + finalizeDangling: true, goalBootstrap: { goalStatus: { kind: 'set', @@ -19440,7 +19646,7 @@ describe('QwenAgent loadSession / unstable_resumeSession', () => { }, }); - expect(replayOptions).toBeUndefined(); + expect(replayOptions).toEqual({ finalizeDangling: true }); mockConnectionState.resolve(); await agentPromise; @@ -19512,7 +19718,7 @@ describe('QwenAgent loadSession / unstable_resumeSession', () => { }, }); - expect(replayOptions).toBeUndefined(); + expect(replayOptions).toEqual({ finalizeDangling: true }); mockConnectionState.resolve(); await agentPromise; diff --git a/packages/cli/src/acp-integration/acpAgent.ts b/packages/cli/src/acp-integration/acpAgent.ts index 45ff95d5d62..0bca0627615 100644 --- a/packages/cli/src/acp-integration/acpAgent.ts +++ b/packages/cli/src/acp-integration/acpAgent.ts @@ -4979,6 +4979,33 @@ class QwenAgent implements Agent { return runWithAcpRuntimeOutputDir(settings, cwd, operation); } + /** + * Whether an ungated restore replay (qwen/session/loadUpdates) may + * finalize dangling tool calls. A session with an active turn — a client + * prompt or an autonomous goal/cron/notification turn — may still owe the + * trailing call's result, so the replay keeps it pending and lets the + * live stream deliver it (#9704). Samples the turn state before the + * transcript read and again at replay time so a turn that starts or + * settles inside the read window is seen. Not for the live loadSession + * path: that restore runs under the close gate, which drains active + * turns, blocks new ones, and reports closing=true — so isTurnIdle() + * there is structurally false and would keep genuinely abandoned calls + * pending forever. + */ + private finalizeDanglingForRestore( + session: Session | undefined, + turnIdleBeforeRead: boolean, + ): boolean { + const idleAtReplay = session?.isTurnIdle() ?? true; + const finalize = turnIdleBeforeRead && idleAtReplay; + // Template literal, not printf-style placeholders: createDebugLogger's + // formatArgs does no util.format substitution, it space-joins the args. + debugLogger.debug( + `[ACP] restore replay finalizeDangling=${finalize} (idleBeforeRead=${turnIdleBeforeRead}, idleAtReplay=${idleAtReplay}) session=${session?.getId() ?? '(non-live)'}`, + ); + return finalize; + } + private async assertLiveSessionScope( config: Config, settings: LoadedSettings, @@ -5405,6 +5432,13 @@ class QwenAgent implements Agent { replayState: replayPage.replay, goalBootstrap: replayGoalBootstrap(projection), suppressRestoreAskUserQuestion, + // The restore gate already drained active turns and blocks + // new ones (and a drain timeout rejects before replay), so + // a trailing unmatched call here is genuinely abandoned — + // finalize it. The turn-activity guard cannot be sampled + // under the gate: isTurnIdle() is structurally false while + // the close gate is held (#9704). + finalizeDangling: true, ...(restoreOptions.replay.kind === 'recent' ? { limits: { @@ -11876,6 +11910,7 @@ class QwenAgent implements Agent { } const liveSession = this.sessions.get(sessionId); + const turnIdleBeforeRead = liveSession?.isTurnIdle() ?? true; let replayConfig = this.config; let sessionData: ResumedSessionData | undefined; if (liveSession) { @@ -11917,6 +11952,13 @@ class QwenAgent implements Agent { // Read-only history dump never re-hangs the question. Skip // finalize only on load/resume that will actually restore. suppressRestoreAskUserQuestion: true, + // Ungated read: unlike the live loadSession restore (whose gate + // drains turns), a turn may still be running here, so guard on + // turn activity instead of finalizing unconditionally (#9704). + finalizeDangling: this.finalizeDanglingForRestore( + liveSession, + turnIdleBeforeRead, + ), }); return { diff --git a/packages/cli/src/acp-integration/session/history-replay-page.test.ts b/packages/cli/src/acp-integration/session/history-replay-page.test.ts index 512326844e6..d03d0bcdce2 100644 --- a/packages/cli/src/acp-integration/session/history-replay-page.test.ts +++ b/packages/cli/src/acp-integration/session/history-replay-page.test.ts @@ -286,6 +286,38 @@ describe('history replay page', () => { ).toBe(false); }); + it('finalizes a dangling tool call as failed by default', async () => { + const result = await collectHistoryReplayUpdates({ + sessionId: SESSION_ID, + records: [userRecord(), toolCallRecord()], + cumulativeUsage: createReplayCumulativeUsage(), + }); + + expect(result.replayError).toBeUndefined(); + expect(result.updates).toContainEqual( + expect.objectContaining({ + sessionUpdate: 'tool_call_update', + status: 'failed', + }), + ); + }); + + it('keeps a dangling tool call in flight when finalizeDangling is false', async () => { + const result = await collectHistoryReplayUpdates({ + sessionId: SESSION_ID, + records: [userRecord(), toolCallRecord()], + cumulativeUsage: createReplayCumulativeUsage(), + finalizeDangling: false, + }); + + expect(result.replayError).toBeUndefined(); + expect( + result.updates.some( + (update) => update.sessionUpdate === 'tool_call_update', + ), + ).toBe(false); + }); + it('bounds textual tool results collected for bulk replay', async () => { const source = 'x'.repeat(499_999); const result = await collectHistoryReplayUpdates({ diff --git a/packages/cli/src/acp-integration/session/history-replay-page.ts b/packages/cli/src/acp-integration/session/history-replay-page.ts index 37955947240..5102c27c18d 100644 --- a/packages/cli/src/acp-integration/session/history-replay-page.ts +++ b/packages/cli/src/acp-integration/session/history-replay-page.ts @@ -254,6 +254,7 @@ export async function collectHistoryReplayUpdates({ goalBootstrap, limits, suppressRestoreAskUserQuestion, + finalizeDangling, }: { sessionId: string; config?: Config; @@ -270,6 +271,14 @@ export async function collectHistoryReplayUpdates({ * so the replayed card doesn't spin forever with no restore prompt coming. */ suppressRestoreAskUserQuestion?: boolean; + /** + * Ungated live-session loads (qwen/session/loadUpdates) while the + * session still has an active turn must not finalize dangling calls: a + * trailing unmatched call is in-flight, not abandoned, and its result + * arrives through the live stream (#9704). The gated live loadSession + * path and non-live loads pass true. + */ + finalizeDangling?: boolean; }): Promise<{ updates: SessionUpdate[]; replayError?: string }> { const updates: SessionUpdate[] = []; try { @@ -299,6 +308,7 @@ export async function collectHistoryReplayUpdates({ ...(initial.goalCause ? { initialGoalCause: initial.goalCause } : {}), ...(goalBootstrap ? { goalBootstrap } : {}), ...(skipFinalizeCallIds ? { skipFinalizeCallIds } : {}), + ...(finalizeDangling === undefined ? {} : { finalizeDangling }), }); } catch (error) { if (error instanceof HistoryReplayLimitError) throw error; diff --git a/packages/cli/src/acp-integration/session/history-replayer.test.ts b/packages/cli/src/acp-integration/session/history-replayer.test.ts index ee0f9a212e5..9e5995d196e 100644 --- a/packages/cli/src/acp-integration/session/history-replayer.test.ts +++ b/packages/cli/src/acp-integration/session/history-replayer.test.ts @@ -433,6 +433,34 @@ describe('HistoryReplayer', () => { ]); }); + it('keeps a dangling call in flight when finalizeDangling is false', async () => { + const record: ChatRecord = { + ...createAssistantRecord(''), + message: { + role: 'model', + parts: [ + { + functionCall: { + id: 'call-inflight', + name: 'run_shell_command', + args: { command: 'sleep 10' }, + }, + }, + ], + }, + }; + + await replayer.replay([record], undefined, { finalizeDangling: false }); + + const updates = sentUpdates(); + expect(updates.map((update) => update['sessionUpdate'])).toEqual([ + 'tool_call', + ]); + expect(replayer.getPendingToolCalls()).toEqual([ + expect.objectContaining({ callId: 'call-inflight' }), + ]); + }); + it('should carry dangling function calls across replay pages', async () => { const record: ChatRecord = { ...createAssistantRecord(''), diff --git a/packages/cli/src/acp-integration/session/history-replayer.ts b/packages/cli/src/acp-integration/session/history-replayer.ts index d3690d7b7ea..05d938220ba 100644 --- a/packages/cli/src/acp-integration/session/history-replayer.ts +++ b/packages/cli/src/acp-integration/session/history-replayer.ts @@ -91,6 +91,7 @@ export class HistoryReplayer { initialGoalCause?: GoalStateCause; goalBootstrap?: HistoryReplayGoalBootstrap; skipFinalizeCallIds?: ReadonlySet; + finalizeDangling?: boolean; } = {}, ): Promise { try { @@ -108,7 +109,7 @@ export class HistoryReplayer { await this.sendUpdate(update); } await this.replayPage(records, { - finalizeDangling: true, + finalizeDangling: options.finalizeDangling ?? true, gaps, ...(options.skipFinalizeCallIds ? { skipFinalizeCallIds: options.skipFinalizeCallIds }