diff --git a/libs/domains/agenstra/backend/feature-agent-manager/src/lib/providers/acp/acp-session.service.spec.ts b/libs/domains/agenstra/backend/feature-agent-manager/src/lib/providers/acp/acp-session.service.spec.ts index f3a66b17b..3743dfc1d 100644 --- a/libs/domains/agenstra/backend/feature-agent-manager/src/lib/providers/acp/acp-session.service.spec.ts +++ b/libs/domains/agenstra/backend/feature-agent-manager/src/lib/providers/acp/acp-session.service.spec.ts @@ -1,10 +1,13 @@ import { Test, TestingModule } from '@nestjs/testing'; import { AgentsRepository } from '../../repositories/agents.repository'; +import type { AgentResponseObject } from '../agent-provider.interface'; -import { AcpClientHostFactory } from './acp-client-host'; +import { AcpClientHostFactory, type AcpPromptEventSink } from './acp-client-host'; +import type { AcpLaunchSpec, AcpSessionKey } from './acp-launch-spec.types'; import { AcpNotificationMapper } from './acp-notification-mapper'; import { AcpSessionService } from './acp-session.service'; +import { CURSOR_ACP_STALE_AUTH_TEXT, CURSOR_ACP_STALE_AUTH_USER_MESSAGE } from './acp-stale-auth'; import { DockerAcpTransportFactory } from './docker-acp-transport'; type CreateOrLoad = ( @@ -16,30 +19,62 @@ type CreateOrLoad = ( knownSessionId?: string, ) => Promise; +type RunPrompt = ( + key: AcpSessionKey, + launchSpec: AcpLaunchSpec, + message: string, + options: undefined, + sink: AcpPromptEventSink, +) => Promise<{ acpSessionId: string }>; + describe('AcpSessionService', () => { let service: AcpSessionService; + let agentsRepository: { + findPersistedAcpSessionId: jest.Mock; + saveAcpSession: jest.Mock; + clearAcpSession: jest.Mock; + }; + let mapper: { mapSessionUpdate: jest.Mock; buildFinalResult: jest.Mock }; const loadSession = jest.fn(); const newSession = jest.fn(); + const sessionKey: AcpSessionKey = { + agentId: 'agent-1', + containerId: 'container-1', + }; + const launchSpec = { + cwd: '/app', + supportsLoadSession: true, + executable: 'cursor-agent', + args: ['acp'], + } as AcpLaunchSpec; + beforeEach(async () => { jest.clearAllMocks(); newSession.mockResolvedValue({ sessionId: 'sess-new' }); loadSession.mockResolvedValue({}); + agentsRepository = { + findPersistedAcpSessionId: jest.fn(), + saveAcpSession: jest.fn(), + clearAcpSession: jest.fn().mockResolvedValue(undefined), + }; + mapper = { + mapSessionUpdate: jest.fn(), + buildFinalResult: jest.fn((text: string, sessionId?: string) => ({ + type: 'result', + subtype: 'success', + result: text, + ...(sessionId ? { session_id: sessionId } : {}), + })), + }; const module: TestingModule = await Test.createTestingModule({ providers: [ AcpSessionService, { provide: DockerAcpTransportFactory, useValue: { connect: jest.fn() } }, { provide: AcpClientHostFactory, useValue: { create: jest.fn() } }, - { provide: AcpNotificationMapper, useValue: { mapSessionUpdate: jest.fn(), buildFinalResult: jest.fn() } }, - { - provide: AgentsRepository, - useValue: { - findPersistedAcpSessionId: jest.fn(), - saveAcpSession: jest.fn(), - clearAcpSession: jest.fn(), - }, - }, + { provide: AcpNotificationMapper, useValue: mapper }, + { provide: AgentsRepository, useValue: agentsRepository }, ], }).compile(); @@ -49,10 +84,20 @@ describe('AcpSessionService', () => { const createOrLoad = (): CreateOrLoad => (service as unknown as { createOrLoadSession: CreateOrLoad }).createOrLoadSession.bind(service); + const collectStream = async (): Promise => { + const events: AgentResponseObject[] = []; + + for await (const event of service.promptStream(sessionKey, launchSpec, 'hi')) { + events.push(event); + } + + return events; + }; + it('createOrLoadSession loads a known agent-issued session id', async () => { - const launchSpec = { cwd: '/app', supportsLoadSession: true }; + const spec = { cwd: '/app', supportsLoadSession: true }; - await expect(createOrLoad()({ loadSession, newSession }, launchSpec, 'sess-old')).resolves.toBe('sess-old'); + await expect(createOrLoad()({ loadSession, newSession }, spec, 'sess-old')).resolves.toBe('sess-old'); expect(loadSession).toHaveBeenCalledWith({ sessionId: 'sess-old', cwd: '/app', @@ -63,17 +108,63 @@ describe('AcpSessionService', () => { it('createOrLoadSession falls back to newSession when loadSession fails', async () => { loadSession.mockRejectedValueOnce(new Error('gone')); - const launchSpec = { cwd: '/app', supportsLoadSession: true }; + const spec = { cwd: '/app', supportsLoadSession: true }; - await expect(createOrLoad()({ loadSession, newSession }, launchSpec, 'sess-old')).resolves.toBe('sess-new'); + await expect(createOrLoad()({ loadSession, newSession }, spec, 'sess-old')).resolves.toBe('sess-new'); expect(newSession).toHaveBeenCalled(); }); it('createOrLoadSession skips load when no known id', async () => { - const launchSpec = { cwd: '/app', supportsLoadSession: true }; + const spec = { cwd: '/app', supportsLoadSession: true }; - await expect(createOrLoad()({ loadSession, newSession }, launchSpec)).resolves.toBe('sess-new'); + await expect(createOrLoad()({ loadSession, newSession }, spec)).resolves.toBe('sess-new'); expect(loadSession).not.toHaveBeenCalled(); expect(newSession).toHaveBeenCalled(); }); + + it('promptStream restarts ACP and retries once after Cursor stale-auth reply', async () => { + let calls = 0; + const runPrompt = jest + .spyOn(service as unknown as { runPrompt: RunPrompt }, 'runPrompt') + .mockImplementation(async (_key, _launchSpec, _message, _options, sink) => { + calls += 1; + + if (calls === 1) { + sink.onResponses([{ type: 'delta', delta: CURSOR_ACP_STALE_AUTH_TEXT }]); + + return { acpSessionId: 'sess-stale' }; + } + + sink.onResponses([{ type: 'delta', delta: 'Recovered answer' }]); + + return { acpSessionId: 'sess-fresh' }; + }); + const closeSession = jest.spyOn(service, 'closeSession').mockResolvedValue(undefined); + + const events = await collectStream(); + + expect(runPrompt).toHaveBeenCalledTimes(2); + expect(closeSession).toHaveBeenCalledWith(sessionKey); + expect(agentsRepository.clearAcpSession).toHaveBeenCalledWith('agent-1', undefined); + expect(events).toEqual([ + { type: 'thinking', phase: 'running' }, + { type: 'delta', delta: 'Recovered answer' }, + { type: 'result', subtype: 'success', result: 'Recovered answer', session_id: 'sess-fresh' }, + ]); + expect(events.some((e) => e.type === 'delta' && e.delta === CURSOR_ACP_STALE_AUTH_TEXT)).toBe(false); + }); + + it('promptStream throws when stale-auth persists after retry', async () => { + jest + .spyOn(service as unknown as { runPrompt: RunPrompt }, 'runPrompt') + .mockImplementation(async (_k, _l, _m, _o, sink) => { + sink.onResponses([{ type: 'delta', delta: CURSOR_ACP_STALE_AUTH_TEXT }]); + + return { acpSessionId: 'sess-stale' }; + }); + jest.spyOn(service, 'closeSession').mockResolvedValue(undefined); + + await expect(collectStream()).rejects.toThrow(CURSOR_ACP_STALE_AUTH_USER_MESSAGE); + expect(agentsRepository.clearAcpSession).toHaveBeenCalled(); + }); }); diff --git a/libs/domains/agenstra/backend/feature-agent-manager/src/lib/providers/acp/acp-session.service.ts b/libs/domains/agenstra/backend/feature-agent-manager/src/lib/providers/acp/acp-session.service.ts index b38ee0ac0..60379f5a4 100644 --- a/libs/domains/agenstra/backend/feature-agent-manager/src/lib/providers/acp/acp-session.service.ts +++ b/libs/domains/agenstra/backend/feature-agent-manager/src/lib/providers/acp/acp-session.service.ts @@ -12,6 +12,11 @@ import { } from './acp-client-host'; import type { AcpLaunchSpec, AcpSessionKey } from './acp-launch-spec.types'; import { AcpNotificationMapper, createAcpToolCallState } from './acp-notification-mapper'; +import { + CURSOR_ACP_STALE_AUTH_USER_MESSAGE, + isCursorAcpStaleAuthPrefix, + isCursorAcpStaleAuthResponse, +} from './acp-stale-auth'; import type { AcpTransport } from './acp-transport.interface'; import { DockerAcpTransportFactory } from './docker-acp-transport'; @@ -22,6 +27,12 @@ interface ManagedAcpSession { bindings: AcpClientHostBindings; } +interface PromptAttemptMeta { + acpSessionId: string | undefined; + aggregatedText: string; + staleAuth: boolean; +} + @Injectable() export class AcpSessionService { private readonly logger = new Logger(AcpSessionService.name); @@ -68,12 +79,86 @@ export class AcpSessionService { launchSpec: AcpLaunchSpec, message: string, options?: AgentProviderOptions, + ): AsyncIterable { + yield* this.promptStreamWithStaleAuthRetry(key, launchSpec, message, options, false); + } + + async prompt( + key: AcpSessionKey, + launchSpec: AcpLaunchSpec, + message: string, + options?: AgentProviderOptions, + ): Promise { + const parts: string[] = []; + + for await (const obj of this.promptStream(key, launchSpec, message, options)) { + parts.push(JSON.stringify(obj)); + } + + return parts.join('\n'); + } + + private async *promptStreamWithStaleAuthRetry( + key: AcpSessionKey, + launchSpec: AcpLaunchSpec, + message: string, + options: AgentProviderOptions | undefined, + isRetry: boolean, + ): AsyncIterable { + const meta: PromptAttemptMeta = { + acpSessionId: undefined, + aggregatedText: '', + staleAuth: false, + }; + + yield* this.iteratePromptAttempt(key, launchSpec, message, options, meta); + + if (meta.staleAuth) { + await this.resetSessionAfterStaleAuth(key); + + if (isRetry) { + this.logger.warn( + `Cursor ACP stale auth persisted after restart for agent ${key.agentId}` + + (key.resumeSessionSuffix ? ` (suffix ${key.resumeSessionSuffix})` : ''), + ); + throw new Error(CURSOR_ACP_STALE_AUTH_USER_MESSAGE); + } + + this.logger.warn( + `Cursor ACP stale auth detected for agent ${key.agentId}; restarting ACP session and retrying once` + + (key.resumeSessionSuffix ? ` (suffix ${key.resumeSessionSuffix})` : ''), + ); + + // Keep reconnect UX out of the assistant transcript (thinking only). + yield { type: 'thinking', phase: 'running' }; + yield* this.promptStreamWithStaleAuthRetry(key, launchSpec, message, options, true); + + return; + } + + if (meta.aggregatedText.trim()) { + yield this.mapper.buildFinalResult(meta.aggregatedText, meta.acpSessionId); + } + } + + /** + * Streams one ACP prompt. Short replies that still look like the Cursor stale-auth line + * are held until the turn ends so we can discard them on restart/retry without flashing UI. + */ + private async *iteratePromptAttempt( + key: AcpSessionKey, + launchSpec: AcpLaunchSpec, + message: string, + options: AgentProviderOptions | undefined, + meta: PromptAttemptMeta, ): AsyncIterable { const queue: AgentResponseObject[] = []; let aggregatedText = ''; let acpSessionId: string | undefined; let done = false; let promptError: unknown | null = null; + const held: AgentResponseObject[] = []; + let holdingForStaleAuth = true; const notify = (() => { let resolve: (() => void) | null = null; @@ -117,11 +202,50 @@ export class AcpSessionService { notify.signal(); }); + const shouldHoldEvent = (obj: AgentResponseObject): boolean => { + if (!holdingForStaleAuth) { + return false; + } + + if ( + obj.type === 'tool_call' || + obj.type === 'tool_result' || + obj.type === 'interaction_query' || + obj.type === 'interactionQuery' || + obj.type === 'question' + ) { + holdingForStaleAuth = false; + + return false; + } + + if (!isCursorAcpStaleAuthPrefix(aggregatedText)) { + holdingForStaleAuth = false; + + return false; + } + + return true; + }; + while (!done || queue.length > 0) { const item = queue.shift(); if (item) { - yield item; + if (shouldHoldEvent(item)) { + held.push(item); + } else { + if (held.length > 0) { + for (const heldItem of held) { + yield heldItem; + } + + held.length = 0; + } + + yield item; + } + continue; } @@ -136,24 +260,23 @@ export class AcpSessionService { throw promptError; } - if (aggregatedText.trim()) { - yield this.mapper.buildFinalResult(aggregatedText, acpSessionId); - } - } + meta.acpSessionId = acpSessionId; + meta.aggregatedText = aggregatedText; + meta.staleAuth = isCursorAcpStaleAuthResponse(aggregatedText); - async prompt( - key: AcpSessionKey, - launchSpec: AcpLaunchSpec, - message: string, - options?: AgentProviderOptions, - ): Promise { - const parts: string[] = []; + if (meta.staleAuth) { + // Discard held sign-in deltas; caller will restart/retry. + return; + } - for await (const obj of this.promptStream(key, launchSpec, message, options)) { - parts.push(JSON.stringify(obj)); + for (const heldItem of held) { + yield heldItem; } + } - return parts.join('\n'); + private async resetSessionAfterStaleAuth(key: AcpSessionKey): Promise { + await this.closeSession(key); + await this.agentsRepository.clearAcpSession(key.agentId, key.resumeSessionSuffix); } private async runPrompt( diff --git a/libs/domains/agenstra/backend/feature-agent-manager/src/lib/providers/acp/acp-stale-auth.spec.ts b/libs/domains/agenstra/backend/feature-agent-manager/src/lib/providers/acp/acp-stale-auth.spec.ts new file mode 100644 index 000000000..d2b60230f --- /dev/null +++ b/libs/domains/agenstra/backend/feature-agent-manager/src/lib/providers/acp/acp-stale-auth.spec.ts @@ -0,0 +1,33 @@ +import { CURSOR_ACP_STALE_AUTH_TEXT, isCursorAcpStaleAuthPrefix, isCursorAcpStaleAuthResponse } from './acp-stale-auth'; + +describe('acp-stale-auth', () => { + describe('isCursorAcpStaleAuthResponse', () => { + it('matches the exact Cursor sign-in line', () => { + expect(isCursorAcpStaleAuthResponse(CURSOR_ACP_STALE_AUTH_TEXT)).toBe(true); + }); + + it('matches case and whitespace variants', () => { + expect(isCursorAcpStaleAuthResponse(' please SIGN IN to continue ')).toBe(true); + expect(isCursorAcpStaleAuthResponse('Please sign in to continue.')).toBe(true); + }); + + it('rejects empty or unrelated replies', () => { + expect(isCursorAcpStaleAuthResponse('')).toBe(false); + expect(isCursorAcpStaleAuthResponse('Please sign in to continue and then run tests')).toBe(false); + expect(isCursorAcpStaleAuthResponse('Unauthorized. Please login first.')).toBe(false); + }); + }); + + describe('isCursorAcpStaleAuthPrefix', () => { + it('holds empty and partial prefixes', () => { + expect(isCursorAcpStaleAuthPrefix('')).toBe(true); + expect(isCursorAcpStaleAuthPrefix('Please')).toBe(true); + expect(isCursorAcpStaleAuthPrefix('Please sign in to continue')).toBe(true); + }); + + it('releases once text diverges from the sign-in line', () => { + expect(isCursorAcpStaleAuthPrefix('Please deploy')).toBe(false); + expect(isCursorAcpStaleAuthPrefix('Hello')).toBe(false); + }); + }); +}); diff --git a/libs/domains/agenstra/backend/feature-agent-manager/src/lib/providers/acp/acp-stale-auth.ts b/libs/domains/agenstra/backend/feature-agent-manager/src/lib/providers/acp/acp-stale-auth.ts new file mode 100644 index 000000000..7841b8efb --- /dev/null +++ b/libs/domains/agenstra/backend/feature-agent-manager/src/lib/providers/acp/acp-stale-auth.ts @@ -0,0 +1,43 @@ +/** + * Cursor ACP long-lived processes sometimes emit this plain assistant text when auth + * goes stale, instead of a structured JSON-RPC error. See: + * https://forum.cursor.com/t/cursor-agent-acp-live-process-returns-sign-in-prompt-as-assistant-content-after-auth-state-goes-stale/163787 + */ +export const CURSOR_ACP_STALE_AUTH_TEXT = 'Please sign in to continue'; + +export const CURSOR_ACP_STALE_AUTH_USER_MESSAGE = + 'Cursor agent session expired. Please try sending your message again.'; + +function normalizeAcpAuthText(text: string): string { + return text + .trim() + .toLowerCase() + .replace(/[.!?]+$/u, '') + .replace(/\s+/g, ' ') + .trim(); +} + +/** True when the full assistant reply is only the Cursor stale-auth sign-in line. */ +export function isCursorAcpStaleAuthResponse(text: string): boolean { + const normalized = normalizeAcpAuthText(text); + + if (!normalized) { + return false; + } + + return normalized === normalizeAcpAuthText(CURSOR_ACP_STALE_AUTH_TEXT); +} + +/** + * True while streamed text could still become the stale-auth reply (or is empty). + * Used to hold deltas so we do not flash the sign-in line before a restart/retry. + */ +export function isCursorAcpStaleAuthPrefix(text: string): boolean { + const normalized = normalizeAcpAuthText(text); + + if (!normalized) { + return true; + } + + return normalizeAcpAuthText(CURSOR_ACP_STALE_AUTH_TEXT).startsWith(normalized); +}