diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 2c2faa3b92..f2c8fc7d05 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -10157,7 +10157,7 @@ test("listImportableSessions returns healthy rows alongside thrown and timed-out }); const resultPromise = manager.listImportableSessions(); - await vi.advanceTimersByTimeAsync(8_000); + await vi.advanceTimersByTimeAsync(90_000); await expect(resultPromise).resolves.toEqual({ sessions: [ @@ -10175,7 +10175,7 @@ test("listImportableSessions returns healthy rows alongside thrown and timed-out { provider: "codex", message: "codex listing failed" }, { provider: "pi", - message: "Timed out listing importable sessions for provider 'pi' after 8000ms", + message: "Timed out listing importable sessions for provider 'pi' after 90000ms", }, ], }); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 2a1b2d4e59..edda3204e4 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -87,7 +87,7 @@ import { withTimeout } from "../../utils/promise-timeout.js"; const RELOAD_SESSION_CLOSE_TIMEOUT_MS = 3_000; const INTERRUPT_SESSION_TIMEOUT_MS = 2_000; -const IMPORTABLE_SESSION_LIST_TIMEOUT_MS = 8_000; +const IMPORTABLE_SESSION_LIST_TIMEOUT_MS = 90_000; const STORED_AGENT_CAPABILITIES: AgentCapabilityFlags = { supportsStreaming: false, supportsSessionPersistence: true, diff --git a/packages/server/src/server/agent/providers/acp-agent.test.ts b/packages/server/src/server/agent/providers/acp-agent.test.ts index ac4b407cf4..1c77a6a25d 100644 --- a/packages/server/src/server/agent/providers/acp-agent.test.ts +++ b/packages/server/src/server/agent/providers/acp-agent.test.ts @@ -45,7 +45,11 @@ import { GenericACPAgentClient } from "./generic-acp-agent.js"; import { parseKiroExtensionCommands } from "./kiro-acp-agent.js"; import { transformPiModels } from "./pi/agent.js"; import type { AgentStreamEvent } from "../agent-sdk-types.js"; -import type { AgentCapabilityFlags, AgentPersistenceHandle } from "../agent-sdk-types.js"; +import type { + AgentCapabilityFlags, + AgentPersistenceHandle, + ProviderRefreshContext, +} from "../agent-sdk-types.js"; import { createTestLogger } from "../../../test-utils/test-logger.js"; import { buildStringCommandShellInvocation } from "../../../utils/string-command-shell.js"; import { asInternals } from "../../test-utils/class-mocks.js"; @@ -3516,6 +3520,246 @@ describe("ACPAgentClient probe cleanup", () => { expect(child.stdout.destroyed).toBe(true); expect(child.stderr.destroyed).toBe(true); }); + + test("closes the native catalog probe session before terminating its process", async () => { + const terminator = new FakeTerminator(); + const child = createProbeChildStub(); + const closeSession = vi.fn().mockImplementation(async () => { + expect(terminator.terminated).toEqual([]); + return {}; + }); + + class TestACPAgentClient extends ACPAgentClient { + protected override async spawnProcess(): Promise { + return { + child, + connection: { + newSession: vi.fn().mockResolvedValue({ + sessionId: "catalog-probe-session", + modes: null, + models: null, + configOptions: [], + }), + unstable_closeSession: closeSession, + }, + initialize: { agentCapabilities: { sessionCapabilities: { close: {} } } }, + } as unknown as SpawnedACPProcess; + } + } + + const client = new TestACPAgentClient({ + provider: "claude-acp", + logger: createTestLogger(), + defaultCommand: ["claude", "--acp"], + defaultModes: [], + terminateProcess: terminator.terminate, + }); + + await client.fetchCatalog({ scope: "workspace", cwd: "/tmp/acp-models", force: false }); + + expect(closeSession).toHaveBeenCalledWith({ sessionId: "catalog-probe-session" }); + expect(terminator.terminated).toContain(child); + }); + + test("closes a catalog probe session that finishes being created after refresh aborts", async () => { + const terminator = new FakeTerminator(); + const child = createProbeChildStub(); + const abort = new AbortController(); + let resolveNewSession!: (value: { + sessionId: string; + modes: null; + models: null; + configOptions: []; + }) => void; + const newSession = vi.fn( + () => + new Promise<{ + sessionId: string; + modes: null; + models: null; + configOptions: []; + }>((resolve) => { + resolveNewSession = resolve; + }), + ); + const closeSession = vi.fn().mockResolvedValue({}); + + class TestACPAgentClient extends ACPAgentClient { + protected override async spawnProcess(): Promise { + return { + child, + connection: { newSession, unstable_closeSession: closeSession }, + initialize: { agentCapabilities: { sessionCapabilities: { close: {} } } }, + } as unknown as SpawnedACPProcess; + } + } + + const client = new TestACPAgentClient({ + provider: "claude-acp", + logger: createTestLogger(), + defaultCommand: ["claude", "--acp"], + defaultModes: [], + terminateProcess: terminator.terminate, + }); + const context: ProviderRefreshContext = { + signal: abort.signal, + runActivity: async (_name, operation) => await operation(), + }; + const catalog = client.fetchCatalog( + { scope: "workspace", cwd: "/tmp/acp-models", force: false }, + context, + ); + await vi.waitFor(() => expect(newSession).toHaveBeenCalledOnce()); + + abort.abort(new Error("refresh cancelled")); + resolveNewSession({ + sessionId: "late-catalog-probe-session", + modes: null, + models: null, + configOptions: [], + }); + + await expect(catalog).rejects.toThrow("refresh cancelled"); + expect(closeSession).toHaveBeenCalledWith({ sessionId: "late-catalog-probe-session" }); + expect(terminator.terminated).toContain(child); + }); + + test("closes the native feature probe session before terminating its process", async () => { + const terminator = new FakeTerminator(); + const child = createProbeChildStub(); + const closeSession = vi.fn().mockResolvedValue({}); + + class TestACPAgentClient extends ACPAgentClient { + protected override async spawnProcess(): Promise { + return { + child, + connection: { + newSession: vi.fn().mockResolvedValue({ + sessionId: "feature-probe-session", + modes: null, + models: null, + configOptions: [copilotAgentConfigOption("Probe Agent")], + }), + unstable_closeSession: closeSession, + }, + initialize: { agentCapabilities: { sessionCapabilities: { close: {} } } }, + } as unknown as SpawnedACPProcess; + } + } + + const client = new TestACPAgentClient({ + provider: "copilot", + logger: createTestLogger(), + defaultCommand: ["copilot", "--acp"], + defaultModes: [], + configFeatureOptions: [COPILOT_AGENT_FEATURE_OPTION], + terminateProcess: terminator.terminate, + }); + + await client.listFeatures({ provider: "copilot", cwd: "/tmp/acp-features" }); + + expect(closeSession).toHaveBeenCalledWith({ sessionId: "feature-probe-session" }); + expect(terminator.terminated).toContain(child); + }); + + test("still terminates the probe when native session close fails", async () => { + const terminator = new FakeTerminator(); + const child = createProbeChildStub(); + + class TestACPAgentClient extends ACPAgentClient { + protected override async spawnProcess(): Promise { + return { + child, + connection: { + newSession: vi.fn().mockResolvedValue({ + sessionId: "catalog-probe-session", + modes: null, + models: null, + configOptions: [], + }), + unstable_closeSession: vi.fn().mockRejectedValue(new Error("close failed")), + }, + initialize: { agentCapabilities: { sessionCapabilities: { close: {} } } }, + } as unknown as SpawnedACPProcess; + } + } + + const client = new TestACPAgentClient({ + provider: "claude-acp", + logger: createTestLogger(), + defaultCommand: ["claude", "--acp"], + defaultModes: [], + terminateProcess: terminator.terminate, + }); + + await client.fetchCatalog({ scope: "workspace", cwd: "/tmp/acp-models", force: false }); + + expect(terminator.terminated).toContain(child); + }); + + test("keeps a session visible and cleans up after its history load times out", async () => { + vi.useFakeTimers(); + try { + const terminator = new FakeTerminator(); + const child = createProbeChildStub(); + const loadSession = vi.fn(async () => await new Promise(() => undefined)); + + class TestACPAgentClient extends ACPAgentClient { + protected override async spawnProcess(): Promise { + return { + child, + connection: { + listSessions: vi.fn().mockResolvedValue({ + sessions: [ + { + sessionId: "slow-session", + cwd: "/tmp/acp-import", + title: "Slow but real", + updatedAt: "2026-09-04T20:00:00.000Z", + }, + ], + nextCursor: null, + }), + loadSession, + unstable_closeSession: vi.fn().mockResolvedValue({}), + }, + initialize: { + agentCapabilities: { + loadSession: true, + sessionCapabilities: { list: {}, close: {} }, + }, + }, + } as unknown as SpawnedACPProcess; + } + } + + const client = new TestACPAgentClient({ + provider: "claude-acp", + logger: createTestLogger(), + defaultCommand: ["claude", "--acp"], + defaultModes: [], + terminateProcess: terminator.terminate, + }); + const listing = client.listImportableSessions({ limit: 1 }); + await vi.waitFor(() => expect(loadSession).toHaveBeenCalledOnce()); + + await vi.advanceTimersByTimeAsync(30_000); + + await expect(listing).resolves.toEqual([ + { + providerHandleId: "slow-session", + cwd: "/tmp/acp-import", + title: "Slow but real", + firstPromptPreview: null, + lastPromptPreview: null, + lastActivityAt: new Date("2026-09-04T20:00:00.000Z"), + }, + ]); + expect(terminator.terminated).toContain(child); + } finally { + vi.useRealTimers(); + } + }); }); describe("ACP session/load invariant — cwd and mcpServers always passed", () => { diff --git a/packages/server/src/server/agent/providers/acp-agent.ts b/packages/server/src/server/agent/providers/acp-agent.ts index fab367e90a..0adb620136 100644 --- a/packages/server/src/server/agent/providers/acp-agent.ts +++ b/packages/server/src/server/agent/providers/acp-agent.ts @@ -277,6 +277,9 @@ export function buildACPClientCapabilities( // NO_BROWSER is honored by Gemini CLI; other ACP agents ignore it. const PROBE_ENV: Record = { NO_BROWSER: "true" }; const ACP_DIAGNOSTIC_PHASE_TIMEOUT_MS = 20_000; +const ACP_PROBE_CLOSE_TIMEOUT_MS = 2_000; +const ACP_IMPORT_HISTORY_LOAD_TIMEOUT_MS = 30_000; +const ACP_IMPORT_HISTORY_BUDGET_MS = 60_000; function summarizeMalformedACPStdoutError(error: unknown): { type: string; message: string } { return { @@ -439,6 +442,7 @@ interface ACPAgentClientOptions { waitForInitialCommands?: boolean; initialCommandsWaitTimeoutMs?: number; terminateProcess?: ProcessTerminator; + now?: () => number; } interface ACPAgentSessionOptions { @@ -493,6 +497,78 @@ interface ACPProcessTransport { spawnError: Promise; } +interface ACPImportPromptPreviews { + firstPromptPreview: string | null; + lastPromptPreview: string | null; + hasConversation: boolean; +} + +interface ACPImportPromptCacheEntry extends ACPImportPromptPreviews { + updatedAt: string | null; +} + +class ACPImportHistoryCollector { + private readonly prompts = new Map< + string, + { + completed: string[]; + current: { messageId: string | null; text: string } | null; + hasConversation: boolean; + } + >(); + + begin(sessionId: string): void { + this.prompts.set(sessionId, { completed: [], current: null, hasConversation: false }); + } + + accept(params: SessionNotification): void { + const state = this.prompts.get(params.sessionId); + if (!state) return; + const update = params.update; + if (isACPConversationUpdate(update)) state.hasConversation = true; + if (update.sessionUpdate !== "user_message_chunk") { + this.flushCurrent(state); + return; + } + + const text = contentBlockToText(update.content); + if (!text) return; + const messageId = update.messageId ?? null; + if (state.current?.messageId && messageId !== state.current.messageId) { + this.flushCurrent(state); + } + state.current ??= { messageId, text: "" }; + state.current.text += text; + } + + finish(sessionId: string): ACPImportPromptPreviews { + const state = this.prompts.get(sessionId); + if (!state) { + return { firstPromptPreview: null, lastPromptPreview: null, hasConversation: false }; + } + this.flushCurrent(state); + this.prompts.delete(sessionId); + return { + firstPromptPreview: normalizeACPImportPromptPreview(state.completed[0] ?? null), + lastPromptPreview: normalizeACPImportPromptPreview(state.completed.at(-1) ?? null), + hasConversation: state.hasConversation, + }; + } + + discard(sessionId: string): void { + this.prompts.delete(sessionId); + } + + private flushCurrent(state: { + completed: string[]; + current: { messageId: string | null; text: string } | null; + hasConversation: boolean; + }): void { + if (state.current?.text.trim()) state.completed.push(state.current.text); + state.current = null; + } +} + export interface ACPToolSnapshot { toolCallId: string; title: string; @@ -826,6 +902,8 @@ export class ACPAgentClient implements AgentClient { private readonly waitForInitialCommands: boolean; private readonly initialCommandsWaitTimeoutMs: number; private readonly extensionCommandsParser?: ACPExtensionCommandsParser; + private readonly importPromptCache = new Map(); + private readonly now: () => number; protected readonly terminateProcess: ProcessTerminator; constructor(options: ACPAgentClientOptions) { @@ -854,6 +932,7 @@ export class ACPAgentClient implements AgentClient { this.waitForInitialCommands = options.waitForInitialCommands ?? false; this.initialCommandsWaitTimeoutMs = options.initialCommandsWaitTimeoutMs ?? 1500; this.extensionCommandsParser = options.extensionCommandsParser; + this.now = options.now ?? Date.now; } async createSession( @@ -948,10 +1027,30 @@ export class ACPAgentClient implements AgentClient { ): Promise { const cwd = options.scope === "global" ? homedir() : options.cwd; let probe: UninitializedACPProcess | null = null; + let probeSessionId: string | null = null; + let probeSessionPromise: Promise | null = null; let closePromise: Promise | null = null; const closeProbe = (): Promise => { if (!probe) return Promise.resolve(); - closePromise ??= this.closeProbe(probe); + closePromise ??= (async () => { + if (!probeSessionId && probeSessionPromise) { + try { + probeSessionId = ( + await withTimeout( + probeSessionPromise, + ACP_PROBE_CLOSE_TIMEOUT_MS, + `ACP probe session/new cleanup timed out after ${ACP_PROBE_CLOSE_TIMEOUT_MS}ms`, + ) + ).sessionId; + } catch (error) { + this.logger.debug( + { err: error }, + "ACP probe session/new did not finish during cleanup", + ); + } + } + await this.closeProbe(probe, probeSessionId); + })(); return closePromise; }; const handleAbort = () => void closeProbe().catch(() => undefined); @@ -970,17 +1069,16 @@ export class ACPAgentClient implements AgentClient { ), ); probe = initializedProbe; - const response = await runProviderRefreshActivity(context, "session/new", () => - raceProviderRefreshAbort( - context?.signal, - this.runACPRequest(() => - initializedProbe.connection.newSession({ - cwd, - mcpServers: [], - }), - ), - ), + probeSessionPromise = this.runACPRequest(() => + initializedProbe.connection.newSession({ + cwd, + mcpServers: [], + }), ); + const response = await runProviderRefreshActivity(context, "session/new", () => + raceProviderRefreshAbort(context?.signal, probeSessionPromise!), + ); + probeSessionId = response.sessionId; const transformed = this.transformSessionResponse(response); const derivedModels = deriveModelDefinitionsFromACP( this.provider, @@ -1030,6 +1128,7 @@ export class ACPAgentClient implements AgentClient { this.assertProvider(config); const probe = await this.spawnProcess(PROBE_ENV); + let probeSessionId: string | null = null; try { const response = await this.runACPRequest(() => probe.connection.newSession({ @@ -1037,53 +1136,154 @@ export class ACPAgentClient implements AgentClient { mcpServers: [], }), ); + probeSessionId = response.sessionId; const transformed = this.transformSessionResponse(response); return [ autoAcceptFeature, ...deriveFeaturesFromACP(transformed.configOptions, this.configFeatureOptions), ]; } finally { - await this.closeProbe(probe); + await this.closeProbe(probe, probeSessionId); } } async listImportableSessions( options?: ListImportableSessionsOptions, ): Promise { - const probe = await this.spawnProcess(PROBE_ENV); + const history = new ACPImportHistoryCollector(); + const probe = await this.spawnProcess(PROBE_ENV, { + client: this.buildProbeClient((params) => history.accept(params)), + }); try { if (!probe.initialize.agentCapabilities?.sessionCapabilities?.list) { return []; } const sessions: ImportableProviderSession[] = []; - const scanLimit = Math.min(options?.scanLimit ?? options?.limit ?? 500, 500); + const resultLimit = options?.limit ?? Number.POSITIVE_INFINITY; + const scanLimit = Math.min(options?.scanLimit ?? 500, 500); + const canLoadHistory = probe.initialize.agentCapabilities.loadSession === true; + const historyDeadline = this.now() + ACP_IMPORT_HISTORY_BUDGET_MS; + let scanned = 0; let cursor: string | null | undefined; for (;;) { const page: ListSessionsResponse = await this.runACPRequest(() => probe.connection.listSessions(acpSessionListRequest(cursor, options?.cwd)), ); for (const session of page.sessions) { - sessions.push({ - providerHandleId: session.sessionId, - cwd: session.cwd, - title: session.title ?? null, - firstPromptPreview: null, - lastPromptPreview: null, - lastActivityAt: session.updatedAt ? new Date(session.updatedAt) : new Date(0), - }); + if (scanned >= scanLimit || sessions.length >= resultLimit) break; + scanned += 1; + const descriptor = await this.describeImportableSession( + probe, + history, + session, + canLoadHistory, + historyDeadline, + ); + if (descriptor) sessions.push(descriptor); } cursor = page.nextCursor ?? null; if (!cursor) break; - if (sessions.length >= scanLimit) break; + if (scanned >= scanLimit || sessions.length >= resultLimit) break; } - return typeof options?.limit === "number" ? sessions.slice(0, options.limit) : sessions; + return sessions; } finally { await this.closeProbe(probe); } } + private async describeImportableSession( + probe: SpawnedACPProcess, + history: ACPImportHistoryCollector, + session: ListSessionsResponse["sessions"][number], + canLoadHistory: boolean, + historyDeadline: number, + ): Promise { + const loadedPreviews = canLoadHistory + ? await this.loadImportPromptPreviews(probe, history, session, historyDeadline) + : null; + if (loadedPreviews?.hasConversation === false) return null; + return { + providerHandleId: session.sessionId, + cwd: session.cwd, + title: session.title ?? null, + firstPromptPreview: loadedPreviews?.firstPromptPreview ?? null, + lastPromptPreview: loadedPreviews?.lastPromptPreview ?? null, + lastActivityAt: session.updatedAt ? new Date(session.updatedAt) : new Date(0), + }; + } + + private async loadImportPromptPreviews( + probe: SpawnedACPProcess, + history: ACPImportHistoryCollector, + session: ListSessionsResponse["sessions"][number], + historyDeadline: number, + ): Promise { + const updatedAt = session.updatedAt ?? null; + const cached = this.importPromptCache.get(session.sessionId); + if (updatedAt !== null && cached?.updatedAt === updatedAt) { + return { + firstPromptPreview: cached.firstPromptPreview, + lastPromptPreview: cached.lastPromptPreview, + hasConversation: cached.hasConversation, + }; + } + + const remainingMs = historyDeadline - this.now(); + if (remainingMs <= 0) return null; + const loadTimeoutMs = Math.min(ACP_IMPORT_HISTORY_LOAD_TIMEOUT_MS, remainingMs); + + history.begin(session.sessionId); + try { + await withTimeout( + this.runACPRequest(() => + probe.connection.loadSession({ + sessionId: session.sessionId, + cwd: session.cwd, + mcpServers: [], + }), + ), + loadTimeoutMs, + `ACP import history load timed out after ${loadTimeoutMs}ms`, + ); + const previews = history.finish(session.sessionId); + if (updatedAt !== null) { + this.importPromptCache.set(session.sessionId, { updatedAt, ...previews }); + } + return previews; + } catch (error) { + history.discard(session.sessionId); + this.logger.debug( + { err: error, sessionId: session.sessionId }, + "ACP import history load failed; keeping the session visible", + ); + return null; + } finally { + await this.closeLoadedSession(probe, session.sessionId, historyDeadline); + } + } + + private async closeLoadedSession( + probe: SpawnedACPProcess, + sessionId: string, + historyDeadline: number, + ): Promise { + if (!probe.initialize.agentCapabilities?.sessionCapabilities?.close) return; + const remainingMs = historyDeadline - this.now(); + if (remainingMs <= 0) return; + const closeTimeoutMs = Math.min(ACP_PROBE_CLOSE_TIMEOUT_MS, remainingMs); + try { + await withTimeout( + probe.connection.unstable_closeSession({ sessionId }), + closeTimeoutMs, + `ACP loaded session/close timed out after ${closeTimeoutMs}ms`, + ); + } catch (error) { + this.logger.debug({ err: error, sessionId }, "ACP loaded session close failed"); + } + } + async importSession(input: ImportProviderSessionInput, context: ImportProviderSessionContext) { return importSessionFromPersistence({ provider: this.provider, @@ -1107,9 +1307,10 @@ export class ACPAgentClient implements AgentClient { options?: { initializeTimeoutMs?: number; onSpawned?: (probe: UninitializedACPProcess) => void; + client?: ACPClient; }, ): Promise { - const transport = await this.spawnTransport(launchEnv); + const transport = await this.spawnTransport(launchEnv, options?.client); const probe: UninitializedACPProcess = { child: transport.child, connection: transport.connection, @@ -1130,7 +1331,10 @@ export class ACPAgentClient implements AgentClient { } } - protected async spawnTransport(launchEnv?: Record): Promise { + protected async spawnTransport( + launchEnv?: Record, + client: ACPClient = this.buildProbeClient(), + ): Promise { const { command, args } = await this.resolveLaunchCommand(); const child = spawnProcess(command, args, { cwd: process.cwd(), @@ -1164,7 +1368,7 @@ export class ACPAgentClient implements AgentClient { Readable.toWeb(child.stdout), { logger: this.logger, provider: this.provider }, ); - const connection = new ClientSideConnection(() => this.buildProbeClient(), stream); + const connection = new ClientSideConnection(() => client, stream); return { child, @@ -1210,12 +1414,16 @@ export class ACPAgentClient implements AgentClient { } } - protected buildProbeClient(): ACPClient { + protected buildProbeClient( + onSessionUpdate?: (params: SessionNotification) => void | Promise, + ): ACPClient { return { async requestPermission(): Promise { return { outcome: { outcome: "cancelled" } }; }, - async sessionUpdate(): Promise {}, + async sessionUpdate(params): Promise { + await onSessionUpdate?.(params); + }, async readTextFile(params: ReadTextFileRequest) { const content = await fs.readFile(params.path, "utf8"); return { content }; @@ -1231,11 +1439,20 @@ export class ACPAgentClient implements AgentClient { }; } - protected async closeProbe(probe: UninitializedACPProcess): Promise { + protected async closeProbe( + probe: UninitializedACPProcess, + sessionId: string | null = null, + ): Promise { try { - if (probe.initialize?.agentCapabilities?.sessionCapabilities?.close) { - // No active session to close here; ignore capability. + if (sessionId && probe.initialize?.agentCapabilities?.sessionCapabilities?.close) { + await withTimeout( + probe.connection.unstable_closeSession({ sessionId }), + ACP_PROBE_CLOSE_TIMEOUT_MS, + `ACP probe session/close timed out after ${ACP_PROBE_CLOSE_TIMEOUT_MS}ms`, + ); } + } catch (error) { + this.logger.debug({ err: error, sessionId }, "ACP probe closeSession failed during cleanup"); } finally { await terminateChildProcess(probe.child, 2_000, this.terminateProcess); } @@ -1259,6 +1476,8 @@ export class ACPAgentClient implements AgentClient { const phaseTimeoutMs = options.phaseTimeoutMs ?? ACP_DIAGNOSTIC_PHASE_TIMEOUT_MS; const cwd = options.cwd ?? homedir(); let transport: ACPProcessTransport | null = null; + let initialize: InitializeResponse | null = null; + let probeSessionId: string | null = null; try { const spawnStartedAt = Date.now(); @@ -1284,7 +1503,7 @@ export class ACPAgentClient implements AgentClient { const initializeStartedAt = Date.now(); try { - await this.initializeTransport(activeTransport, phaseTimeoutMs); + initialize = await this.initializeTransport(activeTransport, phaseTimeoutMs); rows.push({ label: "ACP initialize", value: `ok (${formatDurationMs(initializeStartedAt)})`, @@ -1310,6 +1529,7 @@ export class ACPAgentClient implements AgentClient { phaseTimeoutMs, `ACP session/new timed out after ${phaseTimeoutMs}ms`, ); + probeSessionId = response.sessionId; const transformed = this.transformSessionResponse(response); const models = deriveModelDefinitionsFromACP( this.provider, @@ -1342,7 +1562,15 @@ export class ACPAgentClient implements AgentClient { if (transport) { const cleanupStartedAt = Date.now(); try { - await terminateChildProcess(transport.child, 2_000, this.terminateProcess); + await this.closeProbe( + { + child: transport.child, + connection: transport.connection, + stderrChunks: transport.stderrChunks, + ...(initialize ? { initialize } : {}), + }, + probeSessionId, + ); rows.push({ label: "ACP cleanup", value: `ok (${formatDurationMs(cleanupStartedAt)})`, @@ -3225,6 +3453,23 @@ function contentBlockToText(content: ContentBlock): string { } } +function normalizeACPImportPromptPreview(text: string | null): string | null { + const normalized = text?.trim().replace(/\s+/g, " ") ?? ""; + if (!normalized) return null; + return normalized.length > 160 ? normalized.slice(0, 160) : normalized; +} + +function isACPConversationUpdate(update: SessionUpdate): boolean { + return ( + update.sessionUpdate === "user_message_chunk" || + update.sessionUpdate === "agent_message_chunk" || + update.sessionUpdate === "agent_thought_chunk" || + update.sessionUpdate === "tool_call" || + update.sessionUpdate === "tool_call_update" || + update.sessionUpdate === "plan" + ); +} + function coalesceDefined(next: T | undefined, previous: T | undefined, fallback: T): T { if (next !== undefined) { return next; diff --git a/packages/server/src/server/agent/providers/generic-acp-agent.diagnostic.test.ts b/packages/server/src/server/agent/providers/generic-acp-agent.diagnostic.test.ts index 5bac343fdc..b37e1a9a54 100644 --- a/packages/server/src/server/agent/providers/generic-acp-agent.diagnostic.test.ts +++ b/packages/server/src/server/agent/providers/generic-acp-agent.diagnostic.test.ts @@ -17,7 +17,132 @@ function parseInitializeTrace(content: string): Array<{ clientCapabilities: unkn .map((line) => JSON.parse(line) as { clientCapabilities: unknown }); } +function historyBudgetClock(): () => number { + let calls = 0; + return () => (calls++ < 2 ? 0 : 60_001); +} + describe("GenericACPAgentClient diagnostics", () => { + test("filters empty ACP sessions by replaying history and returns prompt previews", async () => { + await withFakeACPAgent("history-list", async (scriptPath, mode, testDir) => { + const closeTracePath = path.join(testDir, "session-close.jsonl"); + const client = new GenericACPAgentClient({ + logger: createTestLogger(), + command: [process.execPath, scriptPath, mode, "", "", closeTracePath, testDir], + }); + + await expect(client.listImportableSessions({ limit: 1 })).resolves.toEqual([ + { + providerHandleId: "conversation-session", + cwd: testDir, + title: "Conversation", + firstPromptPreview: "First prompt", + lastPromptPreview: "Last prompt", + lastActivityAt: new Date("2026-09-04T20:00:00.000Z"), + }, + ]); + await expect(readFile(closeTracePath, "utf8")).resolves.toBe( + `${JSON.stringify({ sessionId: "empty-session" })}\n${JSON.stringify({ sessionId: "conversation-session" })}\n`, + ); + }); + }); + + test("keeps a listed session visible when its ACP history cannot be loaded", async () => { + await withFakeACPAgent("history-load-failure", async (scriptPath, mode, testDir) => { + const client = new GenericACPAgentClient({ + logger: createTestLogger(), + command: [process.execPath, scriptPath, mode, "", "", "", testDir], + }); + + await expect(client.listImportableSessions({ limit: 1 })).resolves.toEqual([ + { + providerHandleId: "stale-session", + cwd: testDir, + title: "Could still be real", + firstPromptPreview: null, + lastPromptPreview: null, + lastActivityAt: new Date("2026-09-04T20:00:00.000Z"), + }, + ]); + }); + }); + + test("keeps sessions with replayed conversation activity but no user text preview", async () => { + await withFakeACPAgent("history-list", async (scriptPath, mode, testDir) => { + const client = new GenericACPAgentClient({ + logger: createTestLogger(), + command: [process.execPath, scriptPath, mode, "", "", "", testDir], + }); + + const result = await client.listImportableSessions({ limit: 2 }); + + expect(result).toMatchObject([ + { providerHandleId: "conversation-session" }, + { providerHandleId: "assistant-only-session" }, + ]); + expect(result[1]).toMatchObject({ + firstPromptPreview: null, + lastPromptPreview: null, + }); + }); + }); + + test("reuses ACP prompt previews while the listed session timestamp is unchanged", async () => { + await withFakeACPAgent("history-list", async (scriptPath, mode, testDir) => { + const loadTracePath = path.join(testDir, "session-load.jsonl"); + const client = new GenericACPAgentClient({ + logger: createTestLogger(), + command: [process.execPath, scriptPath, mode, "", "", "", testDir, loadTracePath], + }); + + const first = await client.listImportableSessions({ limit: 1 }); + const second = await client.listImportableSessions({ limit: 1 }); + + expect(second).toEqual(first); + await expect(readFile(loadTracePath, "utf8")).resolves.toBe( + `${JSON.stringify({ sessionId: "empty-session" })}\n${JSON.stringify({ sessionId: "conversation-session" })}\n`, + ); + }); + }); + + test("bounds history loads by scanLimit when every inspected session is empty", async () => { + await withFakeACPAgent("history-list", async (scriptPath, mode, testDir) => { + const closeTracePath = path.join(testDir, "session-close.jsonl"); + const client = new GenericACPAgentClient({ + logger: createTestLogger(), + command: [process.execPath, scriptPath, mode, "", "", closeTracePath, testDir], + }); + + await expect(client.listImportableSessions({ limit: 1, scanLimit: 1 })).resolves.toEqual([]); + await expect(readFile(closeTracePath, "utf8")).resolves.toBe( + `${JSON.stringify({ sessionId: "empty-session" })}\n`, + ); + }); + }); + + test("bounds the total time spent loading inaccessible session histories", async () => { + await withFakeACPAgent("history-load-failures", async (scriptPath, mode, testDir) => { + const loadTracePath = path.join(testDir, "session-load.jsonl"); + const client = new GenericACPAgentClient({ + logger: createTestLogger(), + command: [process.execPath, scriptPath, mode, "", "", "", testDir, loadTracePath], + now: historyBudgetClock(), + }); + + const sessions = await client.listImportableSessions({ limit: 4 }); + + expect(sessions).toMatchObject([ + { providerHandleId: "failed-session-1" }, + { providerHandleId: "failed-session-2" }, + { providerHandleId: "failed-session-3" }, + { providerHandleId: "failed-session-4" }, + ]); + await expect(readFile(loadTracePath, "utf8")).resolves.toBe( + `${JSON.stringify({ sessionId: "failed-session-1" })}\n`, + ); + }); + }); + test("probes npx-backed agent packages instead of npx itself", () => { expect(buildVersionProbeCommand(["npx", "-y", "@google/gemini-cli@0.41.1", "--acp"])).toEqual({ command: "npx", @@ -57,6 +182,25 @@ describe("GenericACPAgentClient diagnostics", () => { }); }); + test("closes the native diagnostic probe session", async () => { + await withFakeACPAgent("success", async (scriptPath, mode, testDir) => { + const closeTracePath = path.join(testDir, "session-close.jsonl"); + const client = new GenericACPAgentClient({ + logger: createTestLogger(), + command: [process.execPath, scriptPath, mode, "", "", closeTracePath], + providerId: "custom-acp", + label: "Custom ACP", + diagnosticPhaseTimeoutMs: TEST_ACP_TIMEOUT_MS, + }); + + await client.getDiagnostic(); + + await expect(readFile(closeTracePath, "utf8")).resolves.toContain( + JSON.stringify({ sessionId: "session-1" }), + ); + }); + }); + test("reports a hung ACP session/new phase without failing the diagnostic", async () => { await withFakeACPAgent("hang-session", async (scriptPath, mode) => { const client = new GenericACPAgentClient({ @@ -181,7 +325,12 @@ describe("GenericACPAgentClient diagnostics", () => { }); async function withFakeACPAgent( - mode: "success" | "hang-session", + mode: + | "success" + | "hang-session" + | "history-list" + | "history-load-failure" + | "history-load-failures", run: (scriptPath: string, mode: string, testDir: string) => Promise, ): Promise { await withTempDir("paseo-acp-diagnostic-", async (testDir) => { @@ -230,6 +379,9 @@ const readline = require("node:readline"); const mode = process.argv[2]; const pidPath = process.argv[3]; const initializeTracePath = process.argv[4]; +const closeTracePath = process.argv[5]; +const sessionCwd = process.argv[6]; +const loadTracePath = process.argv[7]; if (pidPath) { fs.writeFileSync(pidPath, String(process.pid)); } @@ -239,6 +391,10 @@ function send(id, result) { process.stdout.write(JSON.stringify({ jsonrpc: "2.0", id, result }) + "\\n"); } +function sendError(id, code, message) { + process.stdout.write(JSON.stringify({ jsonrpc: "2.0", id, error: { code, message } }) + "\\n"); +} + rl.on("line", (line) => { const message = JSON.parse(line); if (message.method === "initialize") { @@ -250,7 +406,13 @@ rl.on("line", (line) => { } send(message.id, { protocolVersion: message.params?.protocolVersion ?? 1, - agentCapabilities: {}, + agentCapabilities: { + loadSession: + mode === "history-list" || + mode === "history-load-failure" || + mode === "history-load-failures", + sessionCapabilities: { close: {}, list: {} }, + }, }); return; } @@ -272,6 +434,122 @@ rl.on("line", (line) => { }, configOptions: [], }); + return; + } + + if (message.method === "session/close") { + if (closeTracePath) { + fs.appendFileSync(closeTracePath, JSON.stringify(message.params) + "\\n"); + } + send(message.id, {}); + return; + } + + if (message.method === "session/list" && mode === "history-list") { + send(message.id, { + sessions: [ + { + sessionId: "empty-session", + cwd: sessionCwd, + title: "Empty", + updatedAt: "2026-09-04T21:00:00.000Z", + }, + { + sessionId: "conversation-session", + cwd: sessionCwd, + title: "Conversation", + updatedAt: "2026-09-04T20:00:00.000Z", + }, + { + sessionId: "assistant-only-session", + cwd: sessionCwd, + title: "Recovered conversation", + updatedAt: "2026-09-04T19:00:00.000Z", + }, + ], + nextCursor: null, + }); + return; + } + + if (message.method === "session/list" && mode === "history-load-failure") { + send(message.id, { + sessions: [{ + sessionId: "stale-session", + cwd: sessionCwd, + title: "Could still be real", + updatedAt: "2026-09-04T20:00:00.000Z", + }], + nextCursor: null, + }); + return; + } + + if (message.method === "session/list" && mode === "history-load-failures") { + send(message.id, { + sessions: [1, 2, 3, 4].map((number) => ({ + sessionId: "failed-session-" + number, + cwd: sessionCwd, + title: "Could still be real " + number, + updatedAt: "2026-09-04T20:00:00.000Z", + })), + nextCursor: null, + }); + return; + } + + if (message.method === "session/load" && mode === "history-list") { + if (loadTracePath) { + fs.appendFileSync( + loadTracePath, + JSON.stringify({ sessionId: message.params.sessionId }) + "\\n", + ); + } + if (message.params.sessionId === "conversation-session") { + for (const [messageId, text] of [["user-1", "First prompt"], ["user-2", "Last prompt"]]) { + process.stdout.write(JSON.stringify({ + jsonrpc: "2.0", + method: "session/update", + params: { + sessionId: message.params.sessionId, + update: { + sessionUpdate: "user_message_chunk", + messageId, + content: { type: "text", text }, + }, + }, + }) + "\\n"); + } + } + if (message.params.sessionId === "assistant-only-session") { + process.stdout.write(JSON.stringify({ + jsonrpc: "2.0", + method: "session/update", + params: { + sessionId: message.params.sessionId, + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "Recovered response" }, + }, + }, + }) + "\\n"); + } + send(message.id, { modes: null, models: null, configOptions: [] }); + return; + } + + if ( + message.method === "session/load" && + (mode === "history-load-failure" || mode === "history-load-failures") + ) { + if (loadTracePath) { + fs.appendFileSync( + loadTracePath, + JSON.stringify({ sessionId: message.params.sessionId }) + "\\n", + ); + } + sendError(message.id, -32602, "stale session"); + return; } }); `; diff --git a/packages/server/src/server/agent/providers/generic-acp-agent.ts b/packages/server/src/server/agent/providers/generic-acp-agent.ts index 095d4e10a8..3fcabb596e 100644 --- a/packages/server/src/server/agent/providers/generic-acp-agent.ts +++ b/packages/server/src/server/agent/providers/generic-acp-agent.ts @@ -51,6 +51,7 @@ interface GenericACPAgentClientOptions { configFeatureOptions?: ACPConfigFeatureOption[]; extensionCommandsParser?: ACPExtensionCommandsParser; catalogModelResolver?: ACPCatalogModelResolver; + now?: () => number; } export class GenericACPAgentClient extends ACPAgentClient { @@ -76,6 +77,7 @@ export class GenericACPAgentClient extends ACPAgentClient { configFeatureOptions: options.configFeatureOptions, extensionCommandsParser: options.extensionCommandsParser, catalogModelResolver: options.catalogModelResolver, + now: options.now, }); this.command = options.command;