fix(agenstra): acp session management

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<string>;
 
+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<AgentResponseObject[]> => {
+    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();
+  });
 });

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<AgentResponseObject> {
+    yield* this.promptStreamWithStaleAuthRetry(key, launchSpec, message, options, false);
+  }
+
+  async prompt(
+    key: AcpSessionKey,
+    launchSpec: AcpLaunchSpec,
+    message: string,
+    options?: AgentProviderOptions,
+  ): Promise<string> {
+    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<AgentResponseObject> {
+    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<AgentResponseObject> {
     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<string> {
-    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<void> {
+    await this.closeSession(key);
+    await this.agentsRepository.clearAcpSession(key.agentId, key.resumeSessionSuffix);
   }
 
   private async runPrompt(