diff --git a/packages/cli/src/acp-integration/acpAgent.test.ts b/packages/cli/src/acp-integration/acpAgent.test.ts index 0c3fa65a046..9fcf8d2e613 100644 --- a/packages/cli/src/acp-integration/acpAgent.test.ts +++ b/packages/cli/src/acp-integration/acpAgent.test.ts @@ -421,6 +421,9 @@ vi.mock('@qwen-code/qwen-code-core', async (importOriginal) => ({ Buffer.from(JSON.stringify(state), 'utf8').toString('base64url'), ), SessionTranscriptReader: vi.fn(), + ChatRecordingService: ( + await importOriginal() + ).ChatRecordingService, isReplayTurnStartType: ( await importOriginal() ).isReplayTurnStartType, @@ -1064,11 +1067,13 @@ import { applyProviderInstallPlan, Storage, SessionTranscriptReader, + ChatRecordingService, InvalidSessionTranscriptCursorError, InvalidSessionTranscriptTurnAnchorError, SessionTranscriptSnapshotUnavailableError, SessionTranscriptTooLargeError, SessionTranscriptPageTooLargeError, + SessionWriterUnavailableError, encodeSessionTranscriptCursor, unregisterGoalHook, getActiveGoal, @@ -2192,6 +2197,7 @@ describe('QwenAgent MCP SSE/HTTP support', () => { noteExternalWorkflowDeletion: ReturnType; isIdle: ReturnType; isTurnIdle: ReturnType; + hasActiveTurn: ReturnType; getCreatedAt: ReturnType; getTurnCount: ReturnType; } @@ -4798,6 +4804,7 @@ describe('QwenAgent MCP SSE/HTTP support', () => { releaseTodoStopGuardQueuedPromptWait: vi.fn().mockReturnValue(true), isIdle: vi.fn().mockReturnValue(true), isTurnIdle: vi.fn().mockReturnValue(true), + hasActiveTurn: vi.fn().mockReturnValue(false), getCreatedAt: vi.fn().mockReturnValue(1_700_000_000_000), getTurnCount: vi.fn().mockReturnValue(3), prompt: vi.fn().mockResolvedValue({ stopReason: 'end_turn' }), @@ -17009,7 +17016,7 @@ describe('QwenAgent MCP SSE/HTTP support', () => { await agentPromise; }); - it('flushes the live recording before reading the latest persisted page', async () => { + it('reads the live transcript page through the owner write barrier', async () => { const innerConfig = await setupSessionMocks(VALID_SESSION_ID); const recording = innerConfig.getChatRecordingService(); const readPage = vi.fn().mockResolvedValue({ @@ -17038,7 +17045,8 @@ describe('QwenAgent MCP SSE/HTTP support', () => { }, ); - expect(recording?.flush).toHaveBeenCalledOnce(); + expect(recording?.runWithWriteBarrier).toHaveBeenCalledOnce(); + expect(recording?.flush).not.toHaveBeenCalled(); expect(readPage).toHaveBeenCalledWith(VALID_SESSION_ID, { direction: 'backward', limit: 100, @@ -17050,6 +17058,381 @@ describe('QwenAgent MCP SSE/HTTP support', () => { await agentPromise; }); + it('runs the live transcript read as the write-barrier operation', async () => { + const innerConfig = await setupSessionMocks(VALID_SESSION_ID); + const recording = innerConfig.getChatRecordingService(); + let releaseBarrier!: () => void; + let markBarrierEntered!: () => void; + const barrierEntered = new Promise((resolve) => { + markBarrierEntered = resolve; + }); + const barrierGate = new Promise((resolve) => { + releaseBarrier = resolve; + }); + recording.runWithWriteBarrier.mockImplementation( + async (operation: () => Promise): Promise => { + markBarrierEntered(); + await barrierGate; + return operation(); + }, + ); + + let readStarted = false; + const readPage = vi.fn().mockImplementation(async () => { + readStarted = true; + return { + sessionId: VALID_SESSION_ID, + records: [], + hasMore: false, + startTime: 'start', + lastUpdated: 'end', + }; + }); + vi.mocked(SessionTranscriptReader).mockImplementation( + () => + ({ + readPage, + }) as unknown as InstanceType, + ); + mockHistoryReplayPage.mockResolvedValue({ pendingToolCalls: [] }); + const { agent, agentPromise } = await bootAcpAgent(); + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + + const transcriptPromise = agent.extMethod( + SERVE_STATUS_EXT_METHODS.sessionTranscript, + { + sessionId: VALID_SESSION_ID, + direction: 'backward', + limit: 1, + }, + ); + await barrierEntered; + expect(readStarted).toBe(false); + + releaseBarrier(); + await transcriptPromise; + expect(readPage).toHaveBeenCalledOnce(); + const barrierResult = await recording.runWithWriteBarrier.mock.results[0]! + .value; + expect(barrierResult).toEqual( + expect.objectContaining({ sessionId: VALID_SESSION_ID }), + ); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('does not barrier cursor or record-anchor transcript pages', async () => { + const innerConfig = await setupSessionMocks(VALID_SESSION_ID); + const recording = innerConfig.getChatRecordingService(); + recording.runWithWriteBarrier.mockRejectedValue( + new Error('recorder not accepting writes'), + ); + const readPage = vi.fn().mockResolvedValue({ + sessionId: VALID_SESSION_ID, + records: [], + hasMore: false, + startTime: 'start', + lastUpdated: 'end', + }); + vi.mocked(SessionTranscriptReader).mockImplementation( + () => + ({ + readPage, + }) as unknown as InstanceType, + ); + mockHistoryReplayPage.mockResolvedValue({ pendingToolCalls: [] }); + const { agent, agentPromise } = await bootAcpAgent(); + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + + await expect( + agent.extMethod(SERVE_STATUS_EXT_METHODS.sessionTranscript, { + sessionId: VALID_SESSION_ID, + cursor: 'cursor-1', + limit: 50, + }), + ).resolves.toMatchObject({ hasMore: false }); + await expect( + agent.extMethod(SERVE_STATUS_EXT_METHODS.sessionTranscript, { + sessionId: VALID_SESSION_ID, + beforeRecordId: 'record-1', + limit: 50, + }), + ).resolves.toMatchObject({ hasMore: false }); + await expect( + agent.extMethod(SERVE_STATUS_EXT_METHODS.sessionTranscript, { + sessionId: VALID_SESSION_ID, + atRecordId: 'u1', + snapshot: 'snapshot-1', + }), + ).resolves.toMatchObject({ hasMore: false }); + + expect(recording.runWithWriteBarrier).not.toHaveBeenCalled(); + expect(recording.flush).not.toHaveBeenCalled(); + expect(readPage).toHaveBeenNthCalledWith(1, VALID_SESSION_ID, { + cursor: 'cursor-1', + limit: 50, + maxBytes: 4 * 1024 * 1024, + }); + expect(readPage).toHaveBeenNthCalledWith(2, VALID_SESSION_ID, { + beforeRecordId: 'record-1', + limit: 50, + maxBytes: 4 * 1024 * 1024, + }); + expect(readPage).toHaveBeenNthCalledWith(3, VALID_SESSION_ID, { + atRecordId: 'u1', + snapshot: 'snapshot-1', + maxBytes: 4 * 1024 * 1024, + }); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('falls back to a drained latest transcript page when the recorder is only lifecycle-unavailable', async () => { + const innerConfig = await setupSessionMocks(VALID_SESSION_ID); + const recording = innerConfig.getChatRecordingService(); + Object.assign(recording, { + acceptingWrites: false, + state: 'closing', + }); + recording.runWithWriteBarrier.mockRejectedValue( + new SessionWriterUnavailableError(), + ); + let queuedVisible = false; + let readStarted = false; + let releaseFlush!: () => void; + let markFlushEntered!: () => void; + const flushEntered = new Promise((resolve) => { + markFlushEntered = resolve; + }); + const flushGate = new Promise((resolve) => { + releaseFlush = resolve; + }); + recording.flush.mockImplementation(async () => { + markFlushEntered(); + await flushGate; + queuedVisible = true; + }); + const readPage = vi.fn().mockImplementation(async () => { + readStarted = true; + return { + sessionId: VALID_SESSION_ID, + records: queuedVisible ? [{ uuid: 'queued-tool-result' }] : [], + hasMore: false, + startTime: 'start', + lastUpdated: 'end', + }; + }); + vi.mocked(SessionTranscriptReader).mockImplementation( + () => + ({ + readPage, + }) as unknown as InstanceType, + ); + mockHistoryReplayPage.mockResolvedValue({ pendingToolCalls: [] }); + const { agent, agentPromise } = await bootAcpAgent(); + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + + const transcriptPromise = agent.extMethod( + SERVE_STATUS_EXT_METHODS.sessionTranscript, + { + sessionId: VALID_SESSION_ID, + direction: 'backward', + limit: 1, + }, + ); + await flushEntered; + expect(readStarted).toBe(false); + + releaseFlush(); + await expect(transcriptPromise).resolves.toMatchObject({ hasMore: false }); + + expect(recording.runWithWriteBarrier).toHaveBeenCalledOnce(); + expect(recording.flush).toHaveBeenCalledOnce(); + expect(readPage).toHaveBeenCalledOnce(); + expect(readPage).toHaveBeenCalledWith(VALID_SESSION_ID, { + direction: 'backward', + limit: 1, + maxBytes: 4 * 1024 * 1024, + }); + expect(await readPage.mock.results[0]!.value).toEqual( + expect.objectContaining({ + records: [{ uuid: 'queued-tool-result' }], + }), + ); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('does not retry a latest transcript page when the barriered read itself fails', async () => { + const innerConfig = await setupSessionMocks(VALID_SESSION_ID); + const recording = innerConfig.getChatRecordingService(); + Object.assign(recording, { + acceptingWrites: false, + state: 'closing', + }); + recording.runWithWriteBarrier.mockImplementation( + async (operation: () => Promise): Promise => operation(), + ); + const readPage = vi + .fn() + .mockRejectedValue(new Error('EMFILE: too many open files')); + vi.mocked(SessionTranscriptReader).mockImplementation( + () => + ({ + readPage, + }) as unknown as InstanceType, + ); + mockHistoryReplayPage.mockResolvedValue({ pendingToolCalls: [] }); + const { agent, agentPromise } = await bootAcpAgent(); + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + + await expect( + agent.extMethod(SERVE_STATUS_EXT_METHODS.sessionTranscript, { + sessionId: VALID_SESSION_ID, + direction: 'backward', + limit: 1, + }), + ).rejects.toThrow('EMFILE: too many open files'); + + expect(recording.runWithWriteBarrier).toHaveBeenCalledOnce(); + expect(recording.flush).not.toHaveBeenCalled(); + expect(readPage).toHaveBeenCalledOnce(); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('propagates a latest transcript page refusal when the recorder has a writeFailure', async () => { + const innerConfig = await setupSessionMocks(VALID_SESSION_ID); + const recording = innerConfig.getChatRecordingService(); + const writeFailure = new SessionWriterUnavailableError(); + Object.assign(recording, { + writeFailure, + acceptingWrites: false, + state: 'integrity_failed', + }); + recording.runWithWriteBarrier.mockRejectedValue(writeFailure); + const readPage = vi.fn().mockResolvedValue({ + sessionId: VALID_SESSION_ID, + records: [], + hasMore: false, + startTime: 'start', + lastUpdated: 'end', + }); + vi.mocked(SessionTranscriptReader).mockImplementation( + () => + ({ + readPage, + }) as unknown as InstanceType, + ); + mockHistoryReplayPage.mockResolvedValue({ pendingToolCalls: [] }); + const { agent, agentPromise } = await bootAcpAgent(); + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + + await expect( + agent.extMethod(SERVE_STATUS_EXT_METHODS.sessionTranscript, { + sessionId: VALID_SESSION_ID, + direction: 'backward', + limit: 1, + }), + ).rejects.toMatchObject({ + code: -32023, + data: { errorKind: 'session_writer_unavailable' }, + }); + + expect(recording.runWithWriteBarrier).toHaveBeenCalledOnce(); + expect(recording.flush).not.toHaveBeenCalled(); + expect(readPage).not.toHaveBeenCalled(); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('falls back after a real recorder beginClose and refuses a latched writeFailure', async () => { + const innerConfig = await setupSessionMocks(VALID_SESSION_ID); + const recorder = new ChatRecordingService( + { + getSessionId: () => VALID_SESSION_ID, + getProjectRoot: () => '/tmp', + getCliVersion: () => '1.0.0', + getResumedSessionData: () => undefined, + storage: { getProjectDir: () => '/tmp' }, + } as unknown as Config, + undefined, + false, + ); + innerConfig.getChatRecordingService = vi.fn().mockReturnValue(recorder); + const readPage = vi.fn().mockResolvedValue({ + sessionId: VALID_SESSION_ID, + records: [], + hasMore: false, + startTime: 'start', + lastUpdated: 'end', + }); + vi.mocked(SessionTranscriptReader).mockImplementation( + () => + ({ + readPage, + }) as unknown as InstanceType, + ); + mockHistoryReplayPage.mockResolvedValue({ pendingToolCalls: [] }); + const { agent, agentPromise } = await bootAcpAgent(); + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + + recorder.beginClose(); + await expect( + agent.extMethod(SERVE_STATUS_EXT_METHODS.sessionTranscript, { + sessionId: VALID_SESSION_ID, + direction: 'backward', + limit: 1, + }), + ).resolves.toMatchObject({ hasMore: false }); + expect(readPage).toHaveBeenCalledOnce(); + + const { SessionWriterUnavailableError: RealWriterUnavailable } = + await vi.importActual( + '@qwen-code/qwen-code-core', + ); + const failedRecorder = new ChatRecordingService( + { + getSessionId: () => VALID_SESSION_ID, + getProjectRoot: () => '/tmp', + getCliVersion: () => '1.0.0', + getResumedSessionData: () => undefined, + storage: { getProjectDir: () => '/tmp' }, + } as unknown as Config, + undefined, + false, + ); + await expect( + failedRecorder.runWithWriteBarrier(async () => { + throw new RealWriterUnavailable(); + }), + ).rejects.toBeInstanceOf(RealWriterUnavailable); + innerConfig.getChatRecordingService = vi + .fn() + .mockReturnValue(failedRecorder); + readPage.mockClear(); + + await expect( + agent.extMethod(SERVE_STATUS_EXT_METHODS.sessionTranscript, { + sessionId: VALID_SESSION_ID, + direction: 'backward', + limit: 1, + }), + ).rejects.toMatchObject({ + code: -32023, + data: { errorKind: 'session_writer_unavailable' }, + }); + expect(readPage).not.toHaveBeenCalled(); + + mockConnectionState.resolve(); + await agentPromise; + }); + it('flushes latest but not frozen turn-index pages', async () => { const innerConfig = await setupSessionMocks(VALID_SESSION_ID); const recording = innerConfig.getChatRecordingService(); @@ -17249,7 +17632,129 @@ describe('QwenAgent MCP SSE/HTTP support', () => { await agentPromise; }); - it('does not finalize dangling transcript calls when a prompt settles during the read', async () => { + it('does not finalize dangling transcript calls when a turn settles during the read', async () => { + await setupSessionMocks(VALID_SESSION_ID); + const page = { + sessionId: VALID_SESSION_ID, + records: [], + hasMore: false, + startTime: 'start', + lastUpdated: 'end', + }; + const readPage = vi.fn().mockResolvedValue(page); + vi.mocked(SessionTranscriptReader).mockImplementation( + () => + ({ + readPage, + }) as unknown as InstanceType, + ); + mockHistoryReplayPage.mockResolvedValue({ pendingToolCalls: [] }); + const { agent, agentPromise } = await bootAcpAgent(); + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + lastSessionMock!.hasActiveTurn.mockReturnValue(true); + readPage.mockImplementationOnce(async () => { + lastSessionMock!.hasActiveTurn.mockReturnValue(false); + await new Promise((resolve) => setImmediate(resolve)); + return page; + }); + + await agent.extMethod(SERVE_STATUS_EXT_METHODS.sessionTranscript, { + sessionId: VALID_SESSION_ID, + }); + expect(mockHistoryReplayPage).toHaveBeenLastCalledWith( + expect.anything(), + [], + expect.objectContaining({ finalizeDangling: false }), + ); + + await agent.extMethod(SERVE_STATUS_EXT_METHODS.sessionTranscript, { + sessionId: VALID_SESSION_ID, + }); + expect(mockHistoryReplayPage).toHaveBeenLastCalledWith( + expect.anything(), + [], + expect.objectContaining({ finalizeDangling: true }), + ); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('keeps a dangling transcript call pending when a turn starts during the read', async () => { + await setupSessionMocks(VALID_SESSION_ID); + const page = { + sessionId: VALID_SESSION_ID, + records: [], + hasMore: false, + startTime: 'start', + lastUpdated: 'end', + }; + const readPage = vi.fn().mockResolvedValue(page); + vi.mocked(SessionTranscriptReader).mockImplementation( + () => + ({ + readPage, + }) as unknown as InstanceType, + ); + mockHistoryReplayPage.mockResolvedValue({ pendingToolCalls: [] }); + const { agent, agentPromise } = await bootAcpAgent(); + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + lastSessionMock!.hasActiveTurn.mockReturnValue(false); + readPage.mockImplementationOnce(async () => { + lastSessionMock!.hasActiveTurn.mockReturnValue(true); + await new Promise((resolve) => setImmediate(resolve)); + return page; + }); + + await agent.extMethod(SERVE_STATUS_EXT_METHODS.sessionTranscript, { + sessionId: VALID_SESSION_ID, + }); + expect(mockHistoryReplayPage).toHaveBeenLastCalledWith( + expect.anything(), + [], + expect.objectContaining({ finalizeDangling: false }), + ); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('finalizes dangling transcript calls when the session is closing with no active turn', async () => { + await setupSessionMocks(VALID_SESSION_ID); + const page = { + sessionId: VALID_SESSION_ID, + records: [], + hasMore: false, + startTime: 'start', + lastUpdated: 'end', + }; + const readPage = vi.fn().mockResolvedValue(page); + vi.mocked(SessionTranscriptReader).mockImplementation( + () => + ({ + readPage, + }) as unknown as InstanceType, + ); + mockHistoryReplayPage.mockResolvedValue({ pendingToolCalls: [] }); + const { agent, agentPromise } = await bootAcpAgent(); + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + lastSessionMock!.isTurnIdle.mockReturnValue(false); + lastSessionMock!.hasActiveTurn.mockReturnValue(false); + + await agent.extMethod(SERVE_STATUS_EXT_METHODS.sessionTranscript, { + sessionId: VALID_SESSION_ID, + }); + expect(mockHistoryReplayPage).toHaveBeenLastCalledWith( + expect.anything(), + [], + expect.objectContaining({ finalizeDangling: true }), + ); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('does not finalize dangling transcript calls while a prompt is waiting at admission', async () => { await setupSessionMocks(VALID_SESSION_ID); const page = { sessionId: VALID_SESSION_ID, @@ -17268,6 +17773,9 @@ describe('QwenAgent MCP SSE/HTTP support', () => { mockHistoryReplayPage.mockResolvedValue({ pendingToolCalls: [] }); const { agent, agentPromise } = await bootAcpAgent(); await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + // Admission window: registered in activePromptCalls, Session turn idle. + lastSessionMock!.hasActiveTurn.mockReturnValue(false); + let finishPrompt: ((value: unknown) => void) | undefined; lastSessionMock!.prompt.mockImplementation( () => @@ -17276,7 +17784,37 @@ describe('QwenAgent MCP SSE/HTTP support', () => { }), ); - const prompt = agent.prompt({ sessionId: VALID_SESSION_ID, prompt: [] }); + const gatedPrompt = agent.prompt({ + sessionId: VALID_SESSION_ID, + prompt: [], + }); + await vi.waitFor(() => expect(lastSessionMock!.prompt).toHaveBeenCalled()); + + await agent.extMethod(SERVE_STATUS_EXT_METHODS.sessionTranscript, { + sessionId: VALID_SESSION_ID, + direction: 'backward', + limit: 1, + }); + expect(mockHistoryReplayPage).toHaveBeenLastCalledWith( + expect.anything(), + [], + expect.objectContaining({ finalizeDangling: false }), + ); + + finishPrompt?.({ stopReason: 'end_turn' }); + await gatedPrompt; + + lastSessionMock!.prompt.mockClear(); + lastSessionMock!.prompt.mockImplementation( + () => + new Promise((resolve) => { + finishPrompt = resolve; + }), + ); + const settlingPrompt = agent.prompt({ + sessionId: VALID_SESSION_ID, + prompt: [], + }); await vi.waitFor(() => expect(lastSessionMock!.prompt).toHaveBeenCalled()); readPage.mockImplementationOnce(async () => { finishPrompt?.({ stopReason: 'end_turn' }); @@ -17286,6 +17824,8 @@ describe('QwenAgent MCP SSE/HTTP support', () => { await agent.extMethod(SERVE_STATUS_EXT_METHODS.sessionTranscript, { sessionId: VALID_SESSION_ID, + direction: 'backward', + limit: 1, }); expect(mockHistoryReplayPage).toHaveBeenLastCalledWith( expect.anything(), @@ -17293,9 +17833,45 @@ describe('QwenAgent MCP SSE/HTTP support', () => { expect.objectContaining({ finalizeDangling: false }), ); - await prompt; + await settlingPrompt; + + lastSessionMock!.prompt.mockClear(); + lastSessionMock!.prompt.mockImplementation( + () => + new Promise((resolve) => { + finishPrompt = resolve; + }), + ); + let startedDuringRead: Promise | undefined; + readPage.mockImplementationOnce(async () => { + startedDuringRead = agent.prompt({ + sessionId: VALID_SESSION_ID, + prompt: [], + }); + await vi.waitFor(() => + expect(lastSessionMock!.prompt).toHaveBeenCalled(), + ); + return page; + }); + await agent.extMethod(SERVE_STATUS_EXT_METHODS.sessionTranscript, { sessionId: VALID_SESSION_ID, + direction: 'backward', + limit: 1, + }); + expect(mockHistoryReplayPage).toHaveBeenLastCalledWith( + expect.anything(), + [], + expect.objectContaining({ finalizeDangling: false }), + ); + + finishPrompt?.({ stopReason: 'end_turn' }); + await startedDuringRead; + + await agent.extMethod(SERVE_STATUS_EXT_METHODS.sessionTranscript, { + sessionId: VALID_SESSION_ID, + direction: 'backward', + limit: 1, }); expect(mockHistoryReplayPage).toHaveBeenLastCalledWith( expect.anything(), diff --git a/packages/cli/src/acp-integration/acpAgent.ts b/packages/cli/src/acp-integration/acpAgent.ts index c67219b9df5..c6b053b2a06 100644 --- a/packages/cli/src/acp-integration/acpAgent.ts +++ b/packages/cli/src/acp-integration/acpAgent.ts @@ -756,6 +756,27 @@ function mapSessionWriterRequestError(error: unknown): unknown { : error; } +/** + * A write-barrier refusal is only a lifecycle miss when the recorder has + * stopped accepting writes and has not latched a writeFailure. Do not key + * this on `instanceof SessionWriterUnavailableError`: the barrier rethrows + * writeFailure first, and that failure can itself be that class. + */ +function isWriterLifecycleUnavailable(recording: object): boolean { + const writer = recording as { + writeFailure?: unknown; + acceptingWrites?: boolean; + state?: string; + }; + if (writer.writeFailure != null) { + return false; + } + return ( + writer.acceptingWrites === false || + (writer.state !== undefined && writer.state !== 'active') + ); +} + async function shutdownSessionConfig(config: Config): Promise { await config.shutdown({ shutdownTelemetry: false }); if (config.hasSessionWriteOwnership()) { @@ -4628,12 +4649,30 @@ class QwenAgent implements Agent { * turns, blocks new ones, and reports closing=true — so isTurnIdle() * there is structurally false and would keep genuinely abandoned calls * pending forever. + * + * qwen/status/session/transcript is also ungated and can be served while + * another request holds the close gate, or after dispose before the + * session leaves this.sessions. Pass ignoreClosing so that path samples + * hasActiveTurn() rather than isTurnIdle(): a closing session with no + * active turn must still finalize abandoned trailing calls. loadUpdates + * keeps the isTurnIdle() sample. isTurnIdle() itself is unchanged — it + * remains the busy-check for turn admission. + * + * That transcript path also ANDs the agent-level activePromptCalls + * sample (taken before the read and again at replay). A prompt already + * registered but still waiting at Session admission has no pendingPrompt + * yet, so hasActiveTurn() is false across that window; without the + * extra sample a backward page would finalize a trailing call the + * prompt is about to resume. */ private finalizeDanglingForRestore( session: Session | undefined, turnIdleBeforeRead: boolean, + options?: { ignoreClosing?: boolean }, ): boolean { - const idleAtReplay = session?.isTurnIdle() ?? true; + const idleAtReplay = options?.ignoreClosing + ? !(session?.hasActiveTurn() ?? false) + : (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. @@ -9025,41 +9064,81 @@ class QwenAgent implements Agent { try { const readTranscriptPage = async (settings: LoadedSettings) => { - if (rawDirection === 'backward') { - await this.sessions - .get(sessionId) - ?.getConfig() - .getChatRecordingService() - ?.flush(); - } - const reader = new SessionTranscriptReader(cwd); - const activePromptBeforeRead = - this.activePromptCalls.has(sessionId); - const page = await reader.readPage(sessionId, { - ...(typeof rawCursor === 'string' ? { cursor: rawCursor } : {}), - ...(typeof rawBeforeRecordId === 'string' - ? { beforeRecordId: rawBeforeRecordId } - : {}), - ...(typeof rawAtRecordId === 'string' - ? { atRecordId: rawAtRecordId } - : {}), - ...(typeof rawSnapshot === 'string' - ? { snapshot: rawSnapshot } - : {}), - ...(rawDirection === 'backward' - ? { direction: rawDirection } - : {}), - ...(typeof rawLimit === 'number' ? { limit: rawLimit } : {}), - maxBytes: SESSION_TRANSCRIPT_MAX_PAGE_BYTES, - }); + const liveSession = this.sessions.get(sessionId); + const recording = liveSession + ?.getConfig() + .getChatRecordingService(); + const promptCallBeforeRead = this.activePromptCalls.has(sessionId); + const turnIdleBeforeRead = liveSession + ? !liveSession.hasActiveTurn() + : true; + let readAttempted = false; + const readPersistedPage = async () => { + readAttempted = true; + const reader = new SessionTranscriptReader(cwd); + return await reader.readPage(sessionId, { + ...(typeof rawCursor === 'string' ? { cursor: rawCursor } : {}), + ...(typeof rawBeforeRecordId === 'string' + ? { beforeRecordId: rawBeforeRecordId } + : {}), + ...(typeof rawAtRecordId === 'string' + ? { atRecordId: rawAtRecordId } + : {}), + ...(typeof rawSnapshot === 'string' + ? { snapshot: rawSnapshot } + : {}), + ...(rawDirection === 'backward' + ? { direction: rawDirection } + : {}), + ...(typeof rawLimit === 'number' ? { limit: rawLimit } : {}), + maxBytes: SESSION_TRANSCRIPT_MAX_PAGE_BYTES, + }); + }; + // Barrier the request's backward/latest page so queued + // appends drain before the disk read. Request direction, + // not the resolved page direction, is the gate. + // Cursor/anchor pages never consulted writer health. + // The barrier refuses before touching the tail when the + // recorder is lifecycle-inactive, so that fallback drains + // via flush().catch then reads. A latched writeFailure + // still fails the read. The #9704 dangling placeholder is + // decided below by ANDing the activePromptCalls sample with + // finalizeDanglingForRestore (ignoreClosing active-turn + // sample), not by this drain — tool results are recorded + // only after the batch ends. + const page = + recording !== undefined && rawDirection === 'backward' + ? await recording + .runWithWriteBarrier(readPersistedPage) + .catch(async (error: unknown) => { + if ( + readAttempted || + !isWriterLifecycleUnavailable(recording) + ) { + throw error; + } + const reason = + error instanceof Error ? error.message : String(error); + debugLogger.debug( + `[ACP] sessionTranscript lifecycle fallback session=${sessionId} error=${reason}`, + ); + await recording.flush().catch(() => undefined); + return readPersistedPage(); + }) + : await readPersistedPage(); const config = await this.getTranscriptReplayConfig(cwd, settings); const replay = await replayTranscriptRecordPage({ sessionId, page, config, finalizeDangling: - !activePromptBeforeRead && - !this.activePromptCalls.has(sessionId), + !promptCallBeforeRead && + !this.activePromptCalls.has(sessionId) && + this.finalizeDanglingForRestore( + liveSession, + turnIdleBeforeRead, + { ignoreClosing: true }, + ), encodeCursor: (state) => encodeSessionTranscriptCursor(state, cwd), logger: debugLogger, diff --git a/packages/cli/src/acp-integration/session/Session.test.ts b/packages/cli/src/acp-integration/session/Session.test.ts index a11bbbf4d6a..bc9989dbc97 100644 --- a/packages/cli/src/acp-integration/session/Session.test.ts +++ b/packages/cli/src/acp-integration/session/Session.test.ts @@ -4113,6 +4113,7 @@ describe('Session', () => { it('rejects a prompt while an exclusive history mutation is active', async () => { const releaseMutation = session.beginHistoryMutation(); + expect(session.hasActiveTurn()).toBe(true); expect(session.isIdle()).toBe(false); expect(session.isTurnIdle()).toBe(false); await expect( @@ -4126,6 +4127,65 @@ describe('Session', () => { releaseMutation(); expect(session.isIdle()).toBe(true); expect(session.isTurnIdle()).toBe(true); + expect(session.hasActiveTurn()).toBe(false); + }); + + it('reports no active turn while the close gate is held', () => { + expect(session.hasActiveTurn()).toBe(false); + expect(session.isTurnIdle()).toBe(true); + const releaseClose = session.beginClose(); + expect(session.hasActiveTurn()).toBe(false); + expect(session.isTurnIdle()).toBe(false); + releaseClose(); + expect(session.isTurnIdle()).toBe(true); + }); + + it('reports an active turn while a prompt is in flight', async () => { + let resolveStream!: () => void; + const streamGate = new Promise((resolve) => { + resolveStream = resolve; + }); + mockChat.sendMessageStream = vi.fn().mockImplementation(async () => { + await streamGate; + return createEmptyStream(); + }); + + const prompt = session.prompt({ + sessionId: 'test-session-id', + prompt: [{ type: 'text', text: 'hello' }], + }); + await vi.waitFor(() => expect(session.hasActiveTurn()).toBe(true)); + expect(session.isTurnIdle()).toBe(false); + + resolveStream(); + await prompt; + expect(session.hasActiveTurn()).toBe(false); + expect(session.isTurnIdle()).toBe(true); + }); + + it('reports an active turn from a non-prompt source under the close gate', async () => { + const internals = session as unknown as { + notificationProcessing: boolean; + notificationCompletion: Promise | null; + }; + let resolveNotification!: () => void; + internals.notificationProcessing = true; + internals.notificationCompletion = new Promise((resolve) => { + resolveNotification = resolve; + }); + + expect(session.hasActiveTurn()).toBe(true); + const releaseClose = session.beginClose(); + expect(session.hasActiveTurn()).toBe(true); + expect(session.isTurnIdle()).toBe(false); + + releaseClose(); + resolveNotification(); + internals.notificationProcessing = false; + internals.notificationCompletion = null; + + expect(session.hasActiveTurn()).toBe(false); + expect(session.isTurnIdle()).toBe(true); }); it('rejects a prompt when a history mutation begins during writer admission', async () => { diff --git a/packages/cli/src/acp-integration/session/Session.ts b/packages/cli/src/acp-integration/session/Session.ts index 7774087e445..5ef1ac57dee 100644 --- a/packages/cli/src/acp-integration/session/Session.ts +++ b/packages/cli/src/acp-integration/session/Session.ts @@ -3928,6 +3928,10 @@ export class Session implements SessionContext { return !this.closing && !this.#hasActiveTurn(); } + hasActiveTurn(): boolean { + return this.#hasActiveTurn(); + } + isIdle(): boolean { return this.isTurnIdle() && this.collectActiveWorkHolds().length === 0; }