diff --git a/packages/server/src/server/agent/provider-registry.ts b/packages/server/src/server/agent/provider-registry.ts index 58e45d37c7..7f87628279 100644 --- a/packages/server/src/server/agent/provider-registry.ts +++ b/packages/server/src/server/agent/provider-registry.ts @@ -108,11 +108,12 @@ export interface BuildProviderRegistryOptions { managedProcesses?: ManagedProcessRegistry; isDev?: boolean; ompRuntime?: OmpRuntime; + hermesSharedProcessScope?: object; } interface ProviderClientFactoryOptions extends Pick< BuildProviderRegistryOptions, - "workspaceGitService" | "managedProcesses" | "ompRuntime" + "workspaceGitService" | "managedProcesses" | "ompRuntime" | "hermesSharedProcessScope" > { providerParams?: unknown; customProvider?: { @@ -700,7 +701,7 @@ function buildResolvedBuiltinProviders( runtimeSettings: AgentProviderRuntimeSettingsMap | undefined, options: Pick< BuildProviderRegistryOptions, - "workspaceGitService" | "managedProcesses" | "ompRuntime" + "workspaceGitService" | "managedProcesses" | "ompRuntime" | "hermesSharedProcessScope" >, isDev: boolean, ): Map { @@ -732,6 +733,7 @@ function buildResolvedBuiltinProviders( workspaceGitService: options.workspaceGitService, managedProcesses: options.managedProcesses, ompRuntime: options.ompRuntime, + hermesSharedProcessScope: options.hermesSharedProcessScope, providerParams: override?.params, }), contract: PROVIDER_CONTRACTS[definition.id] ?? UNSUPPORTED_PROVIDER_CONTRACT, @@ -744,7 +746,7 @@ function buildResolvedBuiltinProviders( function addDerivedProviders( resolvedProviders: Map, providerOverrides: Record, - options: Pick, + options: Pick, ): void { for (const [providerId, override] of Object.entries(providerOverrides)) { if (resolvedProviders.has(providerId) || BUILTIN_PROVIDER_IDS.includes(providerId)) { @@ -789,6 +791,8 @@ function addDerivedProviders( providerId, label: override.label ?? providerId, providerParams: override.params, + sharedProcessScope: + providerId === "hermes" ? options.hermesSharedProcessScope : undefined, }; if (providerId === "cursor") { return new CursorACPAgentClient(acpOptions); @@ -840,6 +844,7 @@ function addDerivedProviders( createBaseClient: (logger) => baseFactory(logger, mergedRuntimeSettings, { managedProcesses: options.managedProcesses, + hermesSharedProcessScope: options.hermesSharedProcessScope, providerParams, customProvider: { id: providerId, @@ -865,11 +870,13 @@ export function buildProviderRegistry( workspaceGitService: options?.workspaceGitService, managedProcesses: options?.managedProcesses, ompRuntime: options?.ompRuntime, + hermesSharedProcessScope: options?.hermesSharedProcessScope, }, options?.isDev === true, ); addDerivedProviders(resolvedProviders, providerOverrides, { managedProcesses: options?.managedProcesses, + hermesSharedProcessScope: options?.hermesSharedProcessScope, }); return Object.fromEntries( diff --git a/packages/server/src/server/agent/provider-snapshot-manager.ts b/packages/server/src/server/agent/provider-snapshot-manager.ts index f0985348af..e201f462ae 100644 --- a/packages/server/src/server/agent/provider-snapshot-manager.ts +++ b/packages/server/src/server/agent/provider-snapshot-manager.ts @@ -204,6 +204,7 @@ export class ProviderSnapshotManager { private readonly managedProcesses?: ManagedProcessRegistry; private readonly isDev: boolean; private readonly extraClients: Partial>; + private readonly hermesSharedProcessScope = {}; private runtimeSettings: AgentProviderRuntimeSettingsMap | undefined; private providerOverrides: Record | undefined; private baseProviderOverrides: Record | undefined; @@ -511,6 +512,7 @@ export class ProviderSnapshotManager { workspaceGitService: this.workspaceGitService, managedProcesses: this.managedProcesses, isDev: this.isDev, + hermesSharedProcessScope: this.hermesSharedProcessScope, }); for (const [provider, client] of Object.entries(this.extraClients) as Array< 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 74cd504c28..35a11f7b24 100644 --- a/packages/server/src/server/agent/providers/acp-agent.test.ts +++ b/packages/server/src/server/agent/providers/acp-agent.test.ts @@ -8,6 +8,7 @@ import { RequestError, ndJsonStream, type Agent, + type Client as ACPClient, PermissionOption, PromptResponse, RequestPermissionRequest, @@ -18,6 +19,7 @@ import { import { ACPAgentClient, ACPAgentSession, + type ACPProcessTransport, type SpawnedACPProcess, type SessionStateResponse, buildACPClientCapabilities, @@ -51,6 +53,14 @@ import { buildStringCommandShellInvocation } from "../../../utils/string-command import { asInternals } from "../../test-utils/class-mocks.js"; import * as spawnUtils from "../../../utils/spawn.js"; +function createDeferred() { + let resolve!: (value: T | PromiseLike) => void; + const promise = new Promise((promiseResolve) => { + resolve = promiseResolve; + }); + return { promise, resolve }; +} + describe("buildACPClientCapabilities", () => { test("keeps filesystem and terminal execution with the agent by default", () => { expect(buildACPClientCapabilities()).toEqual({ @@ -84,6 +94,848 @@ describe("buildACPClientCapabilities", () => { }); }); +describe("ACPAgentClient shared process", () => { + test("shares one process across replacement clients in the same daemon scope", async () => { + const sharedProcessScope = {}; + let spawnCount = 0; + let sessionCount = 0; + + class TestSharedACPAgentClient extends ACPAgentClient { + constructor() { + const options = { + provider: "acp", + logger: createTestLogger(), + defaultCommand: ["hermes", "acp"] as [string, ...string[]], + shareProcess: true, + sharedProcessScope, + }; + super(options); + } + + protected override async spawnTransport( + _launchEnv?: Record, + clientFactory?: () => ACPClient, + ): Promise { + spawnCount += 1; + clientFactory?.(); + return { + child: createProbeChildStub(), + connection: { + initialize: vi.fn().mockResolvedValue({ + protocolVersion: PROTOCOL_VERSION, + agentCapabilities: {}, + }), + newSession: vi.fn(async () => ({ sessionId: `session-${++sessionCount}` })), + } as unknown as ClientSideConnection, + stderrChunks: [], + spawnReady: Promise.resolve(), + spawnError: new Promise(() => undefined), + }; + } + } + + const originalClient = new TestSharedACPAgentClient(); + const originalSession = await originalClient.createSession({ + provider: "acp", + cwd: "/tmp/original", + }); + const replacementClient = new TestSharedACPAgentClient(); + const replacementSession = await replacementClient.createSession({ + provider: "acp", + cwd: "/tmp/replacement", + }); + + expect(spawnCount).toBe(1); + expect(originalSession.id).not.toBe(replacementSession.id); + await expect( + replacementClient.createSession( + { provider: "acp", cwd: "/tmp/different-environment" }, + { env: { CUSTOM_VALUE: "different" } }, + ), + ).rejects.toThrow("Shared ACP sessions require the same launch environment"); + expect(spawnCount).toBe(1); + await Promise.all([originalSession.close(), replacementSession.close()]); + }); + + test("coalesces catalog probes and creates ten sessions on one process", async () => { + class TestSharedACPAgentClient extends ACPAgentClient { + readonly newSession = vi.fn(async () => ({ + sessionId: `session-${this.newSession.mock.calls.length}`, + })); + readonly listSessions = vi.fn(async () => ({ sessions: [] })); + readonly launchEnvs: Array | undefined> = []; + spawnCount = 0; + + constructor() { + super({ + provider: "acp", + logger: createTestLogger(), + defaultCommand: ["hermes", "acp"], + shareProcess: true, + }); + } + + protected override async spawnTransport( + launchEnv?: Record, + clientFactory?: () => ACPClient, + ): Promise { + this.spawnCount += 1; + this.launchEnvs.push(launchEnv); + clientFactory?.(); + return { + child: createProbeChildStub(), + connection: { + initialize: vi.fn().mockResolvedValue({ + protocolVersion: PROTOCOL_VERSION, + agentCapabilities: { sessionCapabilities: { list: {} } }, + }), + newSession: this.newSession, + listSessions: this.listSessions, + } as unknown as ClientSideConnection, + stderrChunks: [], + spawnReady: Promise.resolve(), + spawnError: new Promise(() => undefined), + }; + } + + diagnosticRows() { + return this.buildACPProbeDiagnosticRows(); + } + } + + const client = new TestSharedACPAgentClient(); + const catalogs = await Promise.all( + Array.from({ length: 10 }, () => client.fetchCatalog({ scope: "global", force: true })), + ); + expect(catalogs).toHaveLength(10); + expect(client.spawnCount).toBe(1); + expect(client.newSession).toHaveBeenCalledTimes(1); + + const sessions = await Promise.all( + Array.from({ length: 10 }, (_, index) => + client.createSession( + { provider: "acp", cwd: `/tmp/worktree-${index + 1}` }, + { + agentId: `agent-${index + 1}`, + env: { + PASEO_AGENT_ID: `agent-${index + 1}`, + PASEO_AGENT_CWD: `/tmp/worktree-${index + 1}`, + }, + }, + ), + ), + ); + + expect(client.spawnCount).toBe(1); + expect(client.newSession).toHaveBeenCalledTimes(11); + expect(new Set(sessions.map((session) => session.id)).size).toBe(10); + expect(client.launchEnvs).toEqual([undefined]); + await expect( + client.createSession( + { provider: "acp", cwd: "/tmp/different" }, + { + agentId: "different-agent", + env: { + PASEO_AGENT_ID: "different-agent", + PASEO_AGENT_CWD: "/tmp/different", + PASEO_WORKSPACE_ID: "different-workspace", + SHARED_VALUE: "different", + }, + }, + ), + ).rejects.toThrow("Shared ACP sessions require the same launch environment"); + await expect(client.listImportableSessions()).resolves.toEqual([]); + const diagnosticRows = await client.diagnosticRows(); + expect(diagnosticRows).toContainEqual({ label: "ACP initialize", value: "ok (shared)" }); + expect(client.spawnCount).toBe(1); + expect(client.newSession).toHaveBeenCalledTimes(12); + await Promise.all(sessions.map((session) => session.close())); + }); + + test("delivers session updates that arrive before session registration", async () => { + let router: ACPClient | undefined; + class TestSharedACPAgentClient extends ACPAgentClient { + readonly newSession = vi.fn(async () => { + const sessionId = `session-${this.newSession.mock.calls.length}`; + // Simulate an agent that pushes its slash-command batch immediately + // after the session/new response, before the client continuation has + // registered the session with the shared router. + if (router?.sessionUpdate) { + try { + await router.sessionUpdate({ + sessionId, + update: { + sessionUpdate: "available_commands_update", + availableCommands: [ + { + name: "what-did-you-learn", + description: "Run an evidence-based retrospective", + }, + ], + } as SessionUpdate, + }); + } catch { + // The ACP SDK logs and swallows notification handler errors; + // this mirrors that delivery boundary. + } + } + return { sessionId }; + }); + + constructor() { + super({ + provider: "acp", + logger: createTestLogger(), + defaultCommand: ["hermes", "acp"], + shareProcess: true, + }); + } + + protected override async spawnTransport( + launchEnv?: Record, + clientFactory?: () => ACPClient, + ): Promise { + router = clientFactory?.(); + return { + child: createProbeChildStub(), + connection: { + initialize: vi.fn().mockResolvedValue({ + protocolVersion: PROTOCOL_VERSION, + agentCapabilities: {}, + }), + newSession: this.newSession, + } as unknown as ClientSideConnection, + stderrChunks: [], + spawnReady: Promise.resolve(), + spawnError: new Promise(() => undefined), + }; + } + } + + const client = new TestSharedACPAgentClient(); + const session = await client.createSession( + { provider: "acp", cwd: "/tmp/worktree-1" }, + { agentId: "agent-1" }, + ); + + expect(await session.listCommands?.()).toEqual([ + { + name: "what-did-you-learn", + description: "Run an evidence-based retrospective", + argumentHint: "", + kind: "command", + }, + ]); + await session.close(); + }); + + test("preserves caller launch variables while removing agent-scoped values", async () => { + class TestSharedACPAgentClient extends ACPAgentClient { + launchEnv: Record | undefined; + spawnCount = 0; + + constructor() { + super({ + provider: "acp", + logger: createTestLogger(), + defaultCommand: ["hermes", "acp"], + shareProcess: true, + }); + } + + protected override async spawnTransport( + launchEnv?: Record, + clientFactory?: () => ACPClient, + ): Promise { + this.spawnCount += 1; + this.launchEnv = launchEnv; + clientFactory?.(); + return { + child: createProbeChildStub(), + connection: { + initialize: vi.fn().mockResolvedValue({ + protocolVersion: PROTOCOL_VERSION, + agentCapabilities: { sessionCapabilities: { list: {} } }, + }), + newSession: vi.fn().mockResolvedValue({ sessionId: "session-env" }), + listSessions: vi.fn().mockResolvedValue({ sessions: [] }), + } as unknown as ClientSideConnection, + stderrChunks: [], + spawnReady: Promise.resolve(), + spawnError: new Promise(() => undefined), + }; + } + + diagnosticRows() { + return this.buildACPProbeDiagnosticRows(); + } + } + + const client = new TestSharedACPAgentClient(); + const session = await client.createSession( + { provider: "acp", cwd: "/tmp/worktree" }, + { + agentId: "agent-env", + env: { + PASEO_AGENT_ID: "agent-env", + PASEO_AGENT_CWD: "/tmp/worktree", + PASEO_WORKSPACE_ID: "workspace-env", + CUSTOM_ENDPOINT: "https://example.test", + }, + }, + ); + + expect(client.launchEnv).toEqual({ CUSTOM_ENDPOINT: "https://example.test" }); + await expect( + client.fetchCatalog({ scope: "global", force: true, timeoutMs: 100 }), + ).resolves.toEqual({ models: [], modes: [] }); + await expect(client.listImportableSessions()).resolves.toEqual([]); + await expect(client.diagnosticRows()).resolves.toContainEqual({ + label: "ACP initialize", + value: "ok (shared)", + }); + expect(client.spawnCount).toBe(1); + await session.close(); + }); + + test("applies each caller timeout to a coalesced catalog request", async () => { + let resolveNewSession!: (value: { sessionId: string }) => void; + const newSessionStarted = createDeferred(); + + class TestSharedACPAgentClient extends ACPAgentClient { + constructor() { + super({ + provider: "acp", + logger: createTestLogger(), + defaultCommand: ["hermes", "acp"], + shareProcess: true, + }); + } + + protected override async spawnTransport( + _launchEnv?: Record, + clientFactory?: () => ACPClient, + ): Promise { + clientFactory?.(); + return { + child: createProbeChildStub(), + connection: { + initialize: vi.fn().mockResolvedValue({ + protocolVersion: PROTOCOL_VERSION, + agentCapabilities: {}, + }), + newSession: vi.fn( + () => + new Promise<{ sessionId: string }>((resolve) => { + resolveNewSession = resolve; + newSessionStarted.resolve(); + }), + ), + } as unknown as ClientSideConnection, + stderrChunks: [], + spawnReady: Promise.resolve(), + spawnError: new Promise(() => undefined), + }; + } + } + + const client = new TestSharedACPAgentClient(); + const longRequest = client.fetchCatalog({ scope: "global", force: true, timeoutMs: 100 }); + await newSessionStarted.promise; + await expect( + client.fetchCatalog({ scope: "global", force: true, timeoutMs: 5 }), + ).rejects.toThrow("ACP catalog probe timed out after 5ms"); + resolveNewSession({ sessionId: "session-catalog" }); + await expect(longRequest).resolves.toEqual({ models: [], modes: [] }); + }); + + test("does not invalidate active sessions when a catalog probe times out", async () => { + const catalogStarted = createDeferred(); + const catalogResponse = createDeferred<{ sessionId: string }>(); + const terminateProcess: ProcessTerminator = vi.fn(async (child: TreeKillTarget) => { + (child as ChildProcess).emit("exit", null, "SIGTERM"); + return "terminated" as const; + }); + + class TestSharedACPAgentClient extends ACPAgentClient { + spawnCount = 0; + newSessionCount = 0; + + constructor() { + super({ + provider: "acp", + logger: createTestLogger(), + defaultCommand: ["hermes", "acp"], + shareProcess: true, + terminateProcess, + }); + } + + protected override async spawnTransport( + _launchEnv?: Record, + clientFactory?: () => ACPClient, + ): Promise { + this.spawnCount += 1; + clientFactory?.(); + return { + child: createProbeChildStub(), + connection: { + initialize: vi.fn().mockResolvedValue({ + protocolVersion: PROTOCOL_VERSION, + agentCapabilities: {}, + }), + newSession: vi.fn(() => { + this.newSessionCount += 1; + if (this.newSessionCount === 1) { + return Promise.resolve({ sessionId: "session-active" }); + } + catalogStarted.resolve(); + return catalogResponse.promise; + }), + } as unknown as ClientSideConnection, + stderrChunks: [], + spawnReady: Promise.resolve(), + spawnError: new Promise(() => undefined), + }; + } + } + + const client = new TestSharedACPAgentClient(); + const session = await client.createSession( + { provider: "acp", cwd: "/tmp/worktree" }, + { agentId: "agent-active" }, + ); + const catalogRequest = client.fetchCatalog({ scope: "global", force: true, timeoutMs: 5 }); + await catalogStarted.promise; + await expect(catalogRequest).rejects.toThrow("ACP catalog probe timed out after 5ms"); + expect(terminateProcess).not.toHaveBeenCalled(); + expect(client.spawnCount).toBe(1); + + catalogResponse.resolve({ sessionId: "session-catalog" }); + await session.close(); + expect(terminateProcess).not.toHaveBeenCalled(); + }); + + test("terminates and replaces a shared host after a catalog timeout", async () => { + const terminateProcess: ProcessTerminator = vi.fn(async (child: TreeKillTarget) => { + (child as ChildProcess).emit("exit", null, "SIGTERM"); + return "terminated" as const; + }); + + class TestSharedACPAgentClient extends ACPAgentClient { + spawnCount = 0; + + constructor() { + super({ + provider: "acp", + logger: createTestLogger(), + defaultCommand: ["hermes", "acp"], + shareProcess: true, + terminateProcess, + }); + } + + protected override async spawnTransport( + _launchEnv?: Record, + clientFactory?: () => ACPClient, + ): Promise { + this.spawnCount += 1; + clientFactory?.(); + const newSession = + this.spawnCount === 1 + ? vi.fn(() => new Promise(() => undefined)) + : vi.fn().mockResolvedValue({ sessionId: "session-recovered" }); + return { + child: createProbeChildStub(), + connection: { + initialize: vi.fn().mockResolvedValue({ + protocolVersion: PROTOCOL_VERSION, + agentCapabilities: {}, + }), + newSession, + } as unknown as ClientSideConnection, + stderrChunks: [], + spawnReady: Promise.resolve(), + spawnError: new Promise(() => undefined), + }; + } + } + + const client = new TestSharedACPAgentClient(); + await expect( + client.fetchCatalog({ scope: "global", force: true, timeoutMs: 5 }), + ).rejects.toThrow("ACP catalog probe timed out after 5ms"); + await expect( + client.fetchCatalog({ scope: "global", force: true, timeoutMs: 100 }), + ).resolves.toEqual({ models: [], modes: [] }); + expect(client.spawnCount).toBe(2); + expect(terminateProcess).toHaveBeenCalledTimes(1); + }); + + test("bounds shared host initialization with the catalog timeout", async () => { + const terminateProcess: ProcessTerminator = vi.fn(async (child: TreeKillTarget) => { + (child as ChildProcess).emit("exit", null, "SIGTERM"); + return "terminated" as const; + }); + + class TestSharedACPAgentClient extends ACPAgentClient { + spawnCount = 0; + + constructor() { + super({ + provider: "acp", + logger: createTestLogger(), + defaultCommand: ["hermes", "acp"], + shareProcess: true, + terminateProcess, + }); + } + + protected override async spawnTransport( + _launchEnv?: Record, + clientFactory?: () => ACPClient, + ): Promise { + this.spawnCount += 1; + clientFactory?.(); + const initialize = + this.spawnCount === 1 + ? vi.fn(() => new Promise(() => undefined)) + : vi.fn().mockResolvedValue({ + protocolVersion: PROTOCOL_VERSION, + agentCapabilities: {}, + }); + return { + child: createProbeChildStub(), + connection: { + initialize, + newSession: vi.fn().mockResolvedValue({ sessionId: "session-recovered" }), + } as unknown as ClientSideConnection, + stderrChunks: [], + spawnReady: Promise.resolve(), + spawnError: new Promise(() => undefined), + }; + } + } + + const client = new TestSharedACPAgentClient(); + await expect( + client.fetchCatalog({ scope: "global", force: true, timeoutMs: 5 }), + ).rejects.toThrow("ACP initialize timed out after 5ms"); + await expect( + client.fetchCatalog({ scope: "global", force: true, timeoutMs: 100 }), + ).resolves.toEqual({ models: [], modes: [] }); + expect(client.spawnCount).toBe(2); + expect(terminateProcess).toHaveBeenCalledTimes(1); + }); + + test("retains a timed-out initializing host until its process exits", async () => { + const children: ChildProcessWithoutNullStreams[] = []; + const terminateProcess: ProcessTerminator = vi.fn(async () => "kill-timeout" as const); + + class TestSharedACPAgentClient extends ACPAgentClient { + spawnCount = 0; + + constructor() { + super({ + provider: "acp", + logger: createTestLogger(), + defaultCommand: ["hermes", "acp"], + shareProcess: true, + terminateProcess, + }); + } + + protected override async spawnTransport( + _launchEnv?: Record, + clientFactory?: () => ACPClient, + ): Promise { + this.spawnCount += 1; + clientFactory?.(); + const child = createProbeChildStub(); + children.push(child); + return { + child, + connection: { + initialize: + this.spawnCount === 1 + ? vi.fn(() => new Promise(() => undefined)) + : vi.fn().mockResolvedValue({ + protocolVersion: PROTOCOL_VERSION, + agentCapabilities: {}, + }), + newSession: vi.fn().mockResolvedValue({ sessionId: "session-recovered" }), + } as unknown as ClientSideConnection, + stderrChunks: [], + spawnReady: Promise.resolve(), + spawnError: new Promise(() => undefined), + }; + } + } + + const client = new TestSharedACPAgentClient(); + let settled = false; + const timedOutCatalog = client + .fetchCatalog({ scope: "global", force: true, timeoutMs: 5 }) + .then( + () => null, + (error: unknown) => error, + ) + .finally(() => { + settled = true; + }); + const timedOutResult = await timedOutCatalog; + expect(settled).toBe(true); + expect(timedOutResult).toEqual( + expect.objectContaining({ message: "ACP initialize timed out after 5ms" }), + ); + expect(client.spawnCount).toBe(1); + await expect( + client.fetchCatalog({ scope: "global", force: true, timeoutMs: 5 }), + ).rejects.toThrow("ACP initialize timed out after 5ms"); + expect(client.spawnCount).toBe(1); + + children[0]?.emit("exit", null, "SIGKILL"); + }); + + test("applies the diagnostic phase timeout to shared initialization", async () => { + const terminateProcess: ProcessTerminator = vi.fn(async (child: TreeKillTarget) => { + (child as ChildProcess).emit("exit", null, "SIGTERM"); + return "terminated" as const; + }); + + class TestSharedACPAgentClient extends ACPAgentClient { + constructor() { + super({ + provider: "acp", + logger: createTestLogger(), + defaultCommand: ["hermes", "acp"], + shareProcess: true, + terminateProcess, + }); + } + + protected override async spawnTransport( + _launchEnv?: Record, + clientFactory?: () => ACPClient, + ): Promise { + clientFactory?.(); + return { + child: createProbeChildStub(), + connection: { + initialize: vi.fn(() => new Promise(() => undefined)), + } as unknown as ClientSideConnection, + stderrChunks: [], + spawnReady: Promise.resolve(), + spawnError: new Promise(() => undefined), + }; + } + + diagnosticRows() { + return this.buildACPProbeDiagnosticRows({ phaseTimeoutMs: 5 }); + } + } + + const rows = await new TestSharedACPAgentClient().diagnosticRows(); + expect(rows).toContainEqual({ + label: "ACP shared probe", + value: "error: ACP initialize timed out after 5ms", + }); + expect(terminateProcess).toHaveBeenCalledTimes(1); + }); + + test("bounds a management probe joining an already-initializing host", async () => { + const children: ChildProcessWithoutNullStreams[] = []; + const initializeStarted = createDeferred(); + const terminateProcess: ProcessTerminator = vi.fn(async () => "kill-timeout" as const); + + class TestSharedACPAgentClient extends ACPAgentClient { + spawnCount = 0; + + constructor() { + super({ + provider: "acp", + logger: createTestLogger(), + defaultCommand: ["hermes", "acp"], + shareProcess: true, + terminateProcess, + }); + } + + protected override async spawnTransport( + _launchEnv?: Record, + clientFactory?: () => ACPClient, + ): Promise { + this.spawnCount += 1; + clientFactory?.(); + const child = createProbeChildStub(); + children.push(child); + return { + child, + connection: { + initialize: vi.fn(() => { + initializeStarted.resolve(); + return new Promise(() => undefined); + }), + } as unknown as ClientSideConnection, + stderrChunks: [], + spawnReady: Promise.resolve(), + spawnError: new Promise(() => undefined), + }; + } + + diagnosticRows() { + return this.buildACPProbeDiagnosticRows({ phaseTimeoutMs: 5 }); + } + } + + const client = new TestSharedACPAgentClient(); + const initializingCatalog = client + .fetchCatalog({ scope: "global", force: true, timeoutMs: 30 }) + .then( + () => null, + (error: unknown) => error, + ); + await initializeStarted.promise; + await expect(client.diagnosticRows()).resolves.toContainEqual({ + label: "ACP shared probe", + value: "error: ACP initialize timed out after 5ms", + }); + expect(client.spawnCount).toBe(1); + + children[0]?.emit("exit", null, "SIGKILL"); + expect(await initializingCatalog).toEqual( + expect.objectContaining({ message: "ACP initialize timed out after 30ms" }), + ); + }); + + test("waits for idle shutdown before starting a replacement host", async () => { + vi.useFakeTimers(); + let finishTermination!: () => void; + const terminationGate = new Promise((resolve) => { + finishTermination = resolve; + }); + const terminateProcess: ProcessTerminator = vi.fn(async (child: TreeKillTarget) => { + await terminationGate; + (child as ChildProcess).emit("exit", null, "SIGTERM"); + return "terminated" as const; + }); + + class TestSharedACPAgentClient extends ACPAgentClient { + spawnCount = 0; + + constructor() { + super({ + provider: "acp", + logger: createTestLogger(), + defaultCommand: ["hermes", "acp"], + shareProcess: true, + terminateProcess, + }); + } + + protected override async spawnTransport( + _launchEnv?: Record, + clientFactory?: () => ACPClient, + ): Promise { + this.spawnCount += 1; + clientFactory?.(); + return { + child: createProbeChildStub(), + connection: { + initialize: vi.fn().mockResolvedValue({ + protocolVersion: PROTOCOL_VERSION, + agentCapabilities: {}, + }), + newSession: vi.fn().mockResolvedValue({ sessionId: `session-${this.spawnCount}` }), + } as unknown as ClientSideConnection, + stderrChunks: [], + spawnReady: Promise.resolve(), + spawnError: new Promise(() => undefined), + }; + } + } + + try { + const client = new TestSharedACPAgentClient(); + const first = await client.createSession({ provider: "acp", cwd: "/tmp/first" }); + await first.close(); + await vi.advanceTimersByTimeAsync(1_000); + + const secondPromise = client.createSession({ provider: "acp", cwd: "/tmp/second" }); + await Promise.resolve(); + expect(client.spawnCount).toBe(1); + + finishTermination(); + const second = await secondPromise; + expect(client.spawnCount).toBe(2); + await second.close(); + } finally { + vi.useRealTimers(); + } + }); + + test("does not replace a host until a timed-out process actually exits", async () => { + vi.useFakeTimers(); + const children: ChildProcessWithoutNullStreams[] = []; + const terminateProcess: ProcessTerminator = vi.fn(async () => "kill-timeout" as const); + + class TestSharedACPAgentClient extends ACPAgentClient { + spawnCount = 0; + + constructor() { + super({ + provider: "acp", + logger: createTestLogger(), + defaultCommand: ["hermes", "acp"], + shareProcess: true, + terminateProcess, + }); + } + + protected override async spawnTransport( + _launchEnv?: Record, + clientFactory?: () => ACPClient, + ): Promise { + this.spawnCount += 1; + clientFactory?.(); + const child = createProbeChildStub(); + children.push(child); + return { + child, + connection: { + initialize: vi.fn().mockResolvedValue({ + protocolVersion: PROTOCOL_VERSION, + agentCapabilities: {}, + }), + newSession: vi.fn().mockResolvedValue({ sessionId: `session-${this.spawnCount}` }), + } as unknown as ClientSideConnection, + stderrChunks: [], + spawnReady: Promise.resolve(), + spawnError: new Promise(() => undefined), + }; + } + } + + try { + const client = new TestSharedACPAgentClient(); + const first = await client.createSession({ provider: "acp", cwd: "/tmp/first" }); + await first.close(); + await vi.advanceTimersByTimeAsync(1_000); + + const secondPromise = client.createSession({ provider: "acp", cwd: "/tmp/second" }); + await Promise.resolve(); + expect(client.spawnCount).toBe(1); + + children[0]?.emit("exit", null, "SIGKILL"); + const second = await secondPromise; + expect(client.spawnCount).toBe(2); + await second.close(); + } finally { + vi.useRealTimers(); + } + }); +}); + interface ACPSessionInternals { sessionId: string | null; connection: { prompt: (...args: unknown[]) => Promise }; @@ -1169,6 +2021,32 @@ describe("ACPAgentSession Zed parity", () => { }); }); + test("cancels pending permissions when the shared process exits", async () => { + const session = createSessionWithConfig({ provider: "cursor-acp" }); + asInternals(session).sessionId = "session-1"; + + const permission = session.requestPermission({ + sessionId: "session-1", + toolCall: { + toolCallId: "tool-crash", + title: "Edit file", + kind: "edit", + status: "pending", + }, + options: [{ optionId: "allow-once", name: "Allow", kind: "allow_once" }], + } satisfies RequestPermissionRequest); + await Promise.resolve(); + const [pending] = session.getPendingPermissions(); + expect(pending).toBeDefined(); + + session.handleSharedProcessExit(null, "SIGKILL", "crashed"); + await expect(permission).resolves.toEqual({ outcome: { outcome: "cancelled" } }); + expect(session.getPendingPermissions()).toEqual([]); + await expect(session.respondToPermission(pending!.id, { behavior: "allow" })).rejects.toThrow( + "No pending permission request", + ); + }); + test("preserves ACP chooser actions and returns the selected option", async () => { const session = createSessionWithConfig({ provider: "kimi-acp", diff --git a/packages/server/src/server/agent/providers/acp-agent.ts b/packages/server/src/server/agent/providers/acp-agent.ts index 1a43ed5538..cde9f7c760 100644 --- a/packages/server/src/server/agent/providers/acp-agent.ts +++ b/packages/server/src/server/agent/providers/acp-agent.ts @@ -140,6 +140,25 @@ function isRecord(value: unknown): value is Record { return value != null && typeof value === "object" && !Array.isArray(value); } +function createSharedACPLaunchEnv( + launchEnv?: Record, +): Record | undefined { + if (!launchEnv) { + return undefined; + } + const sharedEntries = Object.entries(launchEnv).filter( + ([key]) => + key !== "PASEO_AGENT_ID" && key !== "PASEO_AGENT_CWD" && key !== "PASEO_WORKSPACE_ID", + ); + return sharedEntries.length > 0 ? Object.fromEntries(sharedEntries) : undefined; +} + +function stableEnvKey(env?: Record): string { + return JSON.stringify( + Object.entries(env ?? {}).sort(([left], [right]) => left.localeCompare(right)), + ); +} + function isACPError(value: unknown): value is ACPError { return isRecord(value) && typeof value.message === "string" && typeof value.code === "number"; } @@ -272,8 +291,12 @@ export function buildACPClientCapabilities( // sign-in URL in the browser) when probing an ACP agent for models/modes. // NO_BROWSER is honored by Gemini CLI; other ACP agents ignore it. const PROBE_ENV: Record = { NO_BROWSER: "true" }; +const ACP_CATALOG_TIMEOUT_MS = 60_000; +const MAX_PENDING_SESSION_UPDATES_PER_SESSION = 100; const ACP_DIAGNOSTIC_PHASE_TIMEOUT_MS = 20_000; +type ACPFetchCatalogOptions = FetchCatalogOptions & { timeoutMs?: number }; + function summarizeMalformedACPStdoutError(error: unknown): { type: string; message: string } { return { type: error instanceof Error ? error.name : typeof error, @@ -435,6 +458,37 @@ interface ACPAgentClientOptions { waitForInitialCommands?: boolean; initialCommandsWaitTimeoutMs?: number; terminateProcess?: ProcessTerminator; + shareProcess?: boolean; + sharedProcessScope?: object; +} + +interface ACPSharedProcessState { + host: Promise | null; + launchEnvKey: string | null; + initializationAbort: (() => Promise) | null; +} + +const SHARED_PROCESS_STATES = new WeakMap(); + +function createACPSharedProcessState(): ACPSharedProcessState { + return { + host: null, + launchEnvKey: null, + initializationAbort: null, + }; +} + +function resolveACPSharedProcessState(scope?: object): ACPSharedProcessState { + if (!scope) { + return createACPSharedProcessState(); + } + const existing = SHARED_PROCESS_STATES.get(scope); + if (existing) { + return existing; + } + const state = createACPSharedProcessState(); + SHARED_PROCESS_STATES.set(scope, state); + return state; } interface ACPAgentSessionOptions { @@ -468,6 +522,32 @@ interface ACPAgentSessionOptions { waitForInitialCommands?: boolean; initialCommandsWaitTimeoutMs?: number; terminateProcess?: ProcessTerminator; + sharedProcess?: ACPSharedProcessLease; +} + +interface ACPSharedRouteTarget extends Pick< + ACPAgentSession, + | "requestPermission" + | "sessionUpdate" + | "readTextFile" + | "writeTextFile" + | "createTerminal" + | "terminalOutput" + | "waitForTerminalExit" + | "releaseTerminal" + | "killTerminal" + | "extNotification" +> { + handleSharedProcessExit?: ACPAgentSession["handleSharedProcessExit"]; +} + +interface ACPSharedProcessLease { + connection: ClientSideConnection; + initialize: InitializeResponse; + register(sessionId: string, client: ACPSharedRouteTarget): void; + hasOtherReferences(): boolean; + release(sessionId: string | null): void; + invalidate(reason: Error): Promise; } export interface SpawnedACPProcess { @@ -481,7 +561,7 @@ type UninitializedACPProcess = Omit & { initialize?: InitializeResponse; }; -interface ACPProcessTransport { +export interface ACPProcessTransport { child: ChildProcessWithoutNullStreams; connection: ClientSideConnection; stderrChunks: string[]; @@ -823,6 +903,10 @@ export class ACPAgentClient implements AgentClient { private readonly initialCommandsWaitTimeoutMs: number; private readonly extensionCommandsParser?: ACPExtensionCommandsParser; protected readonly terminateProcess: ProcessTerminator; + private readonly shareProcess: boolean; + private readonly sharedProcessState: ACPSharedProcessState; + private sharedCatalog: ProviderCatalog | null = null; + private sharedCatalogRequest: Promise | null = null; constructor(options: ACPAgentClientOptions) { this.provider = options.provider; @@ -850,6 +934,8 @@ export class ACPAgentClient implements AgentClient { this.waitForInitialCommands = options.waitForInitialCommands ?? false; this.initialCommandsWaitTimeoutMs = options.initialCommandsWaitTimeoutMs ?? 1500; this.extensionCommandsParser = options.extensionCommandsParser; + this.shareProcess = options.shareProcess ?? false; + this.sharedProcessState = resolveACPSharedProcessState(options.sharedProcessScope); } async createSession( @@ -857,6 +943,9 @@ export class ACPAgentClient implements AgentClient { launchContext?: AgentLaunchContext, ): Promise { this.assertProvider(config); + const sharedProcess = this.shareProcess + ? await this.acquireSharedProcess(launchContext?.env) + : undefined; const session = new ACPAgentSession( { ...config, provider: this.provider }, { @@ -882,6 +971,7 @@ export class ACPAgentClient implements AgentClient { extensionCommandsParser: this.extensionCommandsParser, waitForInitialCommands: this.waitForInitialCommands, initialCommandsWaitTimeoutMs: this.initialCommandsWaitTimeoutMs, + sharedProcess, }, ); await session.initializeNewSession(); @@ -909,6 +999,9 @@ export class ACPAgentClient implements AgentClient { provider: this.provider, cwd, }; + const sharedProcess = this.shareProcess + ? await this.acquireSharedProcess(launchContext?.env) + : undefined; const session = new ACPAgentSession(mergedConfig, { provider: this.provider, logger: this.logger, @@ -933,15 +1026,49 @@ export class ACPAgentClient implements AgentClient { extensionCommandsParser: this.extensionCommandsParser, waitForInitialCommands: this.waitForInitialCommands, initialCommandsWaitTimeoutMs: this.initialCommandsWaitTimeoutMs, + sharedProcess, }); await session.initializeResumedSession(); return session; } async fetchCatalog( - options: FetchCatalogOptions, + options: ACPFetchCatalogOptions, context?: ProviderRefreshContext, ): Promise { + if (this.shareProcess) { + if (!options.force && this.sharedCatalog) { + return this.sharedCatalog; + } + if (this.sharedCatalogRequest) { + const timeoutMs = options.timeoutMs ?? ACP_CATALOG_TIMEOUT_MS; + return runProviderRefreshActivity(context, "shared-catalog", () => + raceProviderRefreshAbort( + context?.signal, + withTimeout( + this.sharedCatalogRequest!, + timeoutMs, + `ACP catalog probe timed out after ${timeoutMs}ms`, + ), + ), + ); + } + const request = this.fetchCatalogFromSharedProcess(options).then((catalog) => { + this.sharedCatalog = catalog; + return catalog; + }); + this.sharedCatalogRequest = request; + try { + return await runProviderRefreshActivity(context, "shared-catalog", () => + raceProviderRefreshAbort(context?.signal, request), + ); + } finally { + if (this.sharedCatalogRequest === request) { + this.sharedCatalogRequest = null; + } + } + } + const cwd = options.scope === "global" ? homedir() : options.cwd; let probe: UninitializedACPProcess | null = null; let closePromise: Promise | null = null; @@ -1018,6 +1145,99 @@ export class ACPAgentClient implements AgentClient { } } + private async fetchCatalogFromSharedProcess( + options: ACPFetchCatalogOptions, + ): Promise { + const startedAt = Date.now(); + const cwd = options.scope === "global" ? homedir() : options.cwd; + const timeoutMs = options.timeoutMs ?? ACP_CATALOG_TIMEOUT_MS; + const sharedProcess = await this.acquireSharedProcess(undefined, timeoutMs, true); + const remainingTimeoutMs = Math.max(1, timeoutMs - (Date.now() - startedAt)); + let requestSettled = false; + const request = (async () => { + let sessionId: string | null = null; + const unsupported = async (): Promise => { + throw new Error("ACP catalog probe does not support client callbacks"); + }; + const routeTarget: ACPSharedRouteTarget = { + requestPermission: unsupported, + sessionUpdate: async () => {}, + readTextFile: unsupported, + writeTextFile: unsupported, + createTerminal: unsupported, + terminalOutput: unsupported, + waitForTerminalExit: unsupported, + releaseTerminal: unsupported, + killTerminal: unsupported, + extNotification: async () => {}, + }; + + try { + const response = await this.runACPRequest(() => + sharedProcess.connection.newSession({ cwd, mcpServers: [] }), + ); + const createdSessionId = response.sessionId; + sessionId = createdSessionId; + sharedProcess.register(createdSessionId, routeTarget); + const transformed = this.transformSessionResponse(response); + const derivedModels = deriveModelDefinitionsFromACP( + this.provider, + transformed.models, + transformed.configOptions, + ); + const models = this.catalogModelResolver + ? await this.catalogModelResolver({ + connection: sharedProcess.connection, + sessionId: createdSessionId, + models: derivedModels, + configOptions: transformed.configOptions, + runRequest: (catalogRequest) => this.runACPRequest(catalogRequest), + transformConfigOptions: (configOptions) => + this.configOptionsTransformer + ? this.configOptionsTransformer(configOptions) + : configOptions, + logger: this.logger, + provider: this.provider, + }) + : derivedModels; + const modeInfo = deriveModesFromACP( + this.defaultModes, + transformed.modes, + transformed.configOptions, + ); + return { + models: this.modelTransformer ? this.modelTransformer(models) : models, + modes: modeInfo.modes, + }; + } finally { + if (sessionId && sharedProcess.initialize.agentCapabilities?.sessionCapabilities?.close) { + try { + await sharedProcess.connection.unstable_closeSession({ sessionId }); + } catch (error) { + this.logger.debug({ err: error }, "ACP shared catalog session close failed"); + } + } + sharedProcess.release(sessionId); + } + })().finally(() => { + requestSettled = true; + }); + + const timeoutMessage = `ACP catalog probe timed out after ${timeoutMs}ms`; + try { + return await withTimeout(request, remainingTimeoutMs, timeoutMessage); + } catch (error) { + if (!requestSettled) { + const hasOtherReferences = sharedProcess.hasOtherReferences(); + sharedProcess.release(null); + if (!hasOtherReferences) { + await sharedProcess.invalidate(new Error(timeoutMessage)); + } + } + throw error; + } + } + async listFeatures(config: AgentSessionConfig): Promise { const autoAcceptFeature = buildACPAutoAcceptFeature(config); if (this.configFeatureOptions.length === 0) { @@ -1046,7 +1266,7 @@ export class ACPAgentClient implements AgentClient { async listImportableSessions( options?: ListImportableSessionsOptions, ): Promise { - const probe = await this.spawnProcess(PROBE_ENV); + const probe = await this.acquireSessionListingConnection(); try { if (!probe.initialize.agentCapabilities?.sessionCapabilities?.list) { return []; @@ -1081,8 +1301,33 @@ export class ACPAgentClient implements AgentClient { return typeof options?.limit === "number" ? sessions.slice(0, options.limit) : sessions; } finally { - await this.closeProbe(probe); + await probe.release(); + } + } + + private async acquireSessionListingConnection(): Promise<{ + connection: ClientSideConnection; + initialize: InitializeResponse; + release: () => Promise; + }> { + if (this.shareProcess) { + const sharedProcess = await this.acquireSharedProcess( + undefined, + ACP_CATALOG_TIMEOUT_MS, + true, + ); + return { + connection: sharedProcess.connection, + initialize: sharedProcess.initialize, + release: async () => sharedProcess.release(null), + }; } + const probe = await this.spawnProcess(PROBE_ENV); + return { + connection: probe.connection, + initialize: probe.initialize, + release: async () => this.closeProbe(probe), + }; } async importSession(input: ImportProviderSessionInput, context: ImportProviderSessionContext) { @@ -1131,7 +1376,10 @@ export class ACPAgentClient implements AgentClient { } } - protected async spawnTransport(launchEnv?: Record): Promise { + protected async spawnTransport( + launchEnv?: Record, + clientFactory: () => ACPClient = () => this.buildProbeClient(), + ): Promise { const { command, args } = await this.resolveLaunchCommand(); const child = spawnProcess(command, args, { cwd: process.cwd(), @@ -1165,7 +1413,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(clientFactory, stream); return { child, @@ -1211,6 +1459,128 @@ export class ACPAgentClient implements AgentClient { } } + private async acquireSharedProcess( + launchEnv?: Record, + initializeTimeoutMs = ACP_CATALOG_TIMEOUT_MS, + reuseActiveLaunchEnvironment = false, + ): Promise { + const sharedLaunchEnv = createSharedACPLaunchEnv(launchEnv); + const requestedLaunchEnvKey = stableEnvKey(sharedLaunchEnv); + for (;;) { + const launchEnvKey = + reuseActiveLaunchEnvironment && + this.sharedProcessState.host && + this.sharedProcessState.launchEnvKey + ? this.sharedProcessState.launchEnvKey + : requestedLaunchEnvKey; + if (this.sharedProcessState.host && this.sharedProcessState.launchEnvKey !== launchEnvKey) { + const existingHost = await this.awaitSharedProcessHost( + this.sharedProcessState.host, + initializeTimeoutMs, + ); + if (existingHost.isStopped()) { + await existingHost.whenStopped(); + continue; + } + if (!existingHost.hasReferences()) { + await existingHost.retire(); + continue; + } + throw new Error( + "Shared ACP sessions require the same launch environment; refusing to reuse a process with different environment values", + ); + } + if (!this.sharedProcessState.host) { + let hostPromise: Promise; + hostPromise = this.createSharedProcessHost(sharedLaunchEnv, initializeTimeoutMs, () => { + if (this.sharedProcessState.host === hostPromise) { + this.sharedProcessState.host = null; + this.sharedProcessState.launchEnvKey = null; + } + }); + this.sharedProcessState.host = hostPromise; + this.sharedProcessState.launchEnvKey = launchEnvKey; + void hostPromise.catch(() => { + if (this.sharedProcessState.host === hostPromise) { + this.sharedProcessState.host = null; + this.sharedProcessState.launchEnvKey = null; + } + }); + } + const host = await this.awaitSharedProcessHost( + this.sharedProcessState.host, + initializeTimeoutMs, + ); + if (host.isStopped()) { + await host.whenStopped(); + continue; + } + return host.acquire(); + } + } + + private async awaitSharedProcessHost( + hostPromise: Promise, + timeoutMs: number, + ): Promise { + try { + return await withTimeout( + hostPromise, + timeoutMs, + `ACP initialize timed out after ${timeoutMs}ms`, + ); + } catch (error) { + if ( + this.sharedProcessState.host === hostPromise && + this.sharedProcessState.initializationAbort + ) { + const abort = this.sharedProcessState.initializationAbort; + void abort().then(() => { + if (this.sharedProcessState.host === hostPromise) { + this.sharedProcessState.host = null; + this.sharedProcessState.launchEnvKey = null; + } + return undefined; + }); + } + throw error; + } + } + + private async createSharedProcessHost( + launchEnv?: Record, + initializeTimeoutMs = ACP_CATALOG_TIMEOUT_MS, + onStopped: () => void = () => {}, + ): Promise { + const router = new ACPSharedClientRouter(); + const transport = await this.spawnTransport(launchEnv, () => router); + let cleanupPromise: Promise | null = null; + const abortInitialization = (): Promise => { + cleanupPromise ??= terminateChildProcess(transport.child, 2_000, this.terminateProcess, true); + return cleanupPromise; + }; + this.sharedProcessState.initializationAbort = abortInitialization; + try { + const initialize = await this.initializeTransport(transport, initializeTimeoutMs); + return new ACPSharedProcessHost({ + child: transport.child, + connection: transport.connection, + initialize, + stderrChunks: transport.stderrChunks, + router, + terminateProcess: this.terminateProcess, + onStopped, + }); + } catch (error) { + await abortInitialization(); + throw error; + } finally { + if (this.sharedProcessState.initializationAbort === abortInitialization) { + this.sharedProcessState.initializationAbort = null; + } + } + } + protected buildProbeClient(): ACPClient { return { async requestPermission(): Promise { @@ -1256,6 +1626,10 @@ export class ACPAgentClient implements AgentClient { phaseTimeoutMs?: number; } = {}, ): Promise { + if (this.shareProcess) { + return this.buildSharedACPProbeDiagnosticRows(options); + } + const rows: DiagnosticEntry[] = []; const phaseTimeoutMs = options.phaseTimeoutMs ?? ACP_DIAGNOSTIC_PHASE_TIMEOUT_MS; const cwd = options.cwd ?? homedir(); @@ -1358,6 +1732,49 @@ export class ACPAgentClient implements AgentClient { } } + private async buildSharedACPProbeDiagnosticRows( + options: { + cwd?: string; + phaseTimeoutMs?: number; + } = {}, + ): Promise { + const rows: DiagnosticEntry[] = []; + const startedAt = Date.now(); + let sharedProcess: ACPSharedProcessLease | null = null; + try { + sharedProcess = await this.acquireSharedProcess( + undefined, + options.phaseTimeoutMs ?? ACP_DIAGNOSTIC_PHASE_TIMEOUT_MS, + true, + ); + rows.push( + { label: "ACP spawn", value: `ok (${formatDurationMs(startedAt)}; shared)` }, + { label: "ACP initialize", value: "ok (shared)" }, + ); + const catalogStartedAt = Date.now(); + const catalog = await this.fetchCatalogFromSharedProcess({ + scope: "workspace", + cwd: options.cwd ?? homedir(), + force: true, + timeoutMs: options.phaseTimeoutMs ?? ACP_DIAGNOSTIC_PHASE_TIMEOUT_MS, + }); + rows.push({ + label: "ACP session/new", + value: `ok (${formatDurationMs(catalogStartedAt)}; models=${catalog.models.length}; modes=${catalog.modes.length}; shared)`, + }); + return rows; + } catch (error) { + rows.push({ + label: "ACP shared probe", + value: `error: ${toDiagnosticErrorMessage(error)}`, + }); + return rows; + } finally { + sharedProcess?.release(null); + rows.push({ label: "ACP cleanup", value: "ok (shared lease released)" }); + } + } + protected async resolveLaunchCommand(): Promise<{ command: string; args: string[] }> { const prefix = await resolveProviderLaunch({ commandConfig: this.runtimeSettings?.command, @@ -1434,6 +1851,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { private readonly terminalEntries = new Map(); private readonly persistedHistory: AgentTimelineItem[] = []; private readonly initialHandle?: AgentPersistenceHandle; + private readonly sharedProcess?: ACPSharedProcessLease; private readonly config: AgentSessionConfig; private child: ChildProcessWithoutNullStreams | null = null; @@ -1486,6 +1904,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { this.agentId = options.agentId; this.launchEnv = options.launchEnv; this.initialHandle = options.handle; + this.sharedProcess = options.sharedProcess; this.config = { ...config, provider: options.provider }; this.currentMode = config.modeId ?? null; this.currentModel = config.model ?? null; @@ -1502,10 +1921,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { async initializeNewSession(): Promise { try { - const spawned = await this.spawnProcess(); - this.child = spawned.child; - this.connection = spawned.connection; - this.agentCapabilities = spawned.initialize.agentCapabilities ?? null; + await this.attachProcess(); const response = await this.runACPRequest(() => this.connection!.newSession({ @@ -1514,6 +1930,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { }), ); this.sessionId = response.sessionId; + this.sharedProcess?.register(response.sessionId, this); this.bootstrapThreadEventPending = true; this.applySessionState(response); await this.applyConfiguredOverrides(); @@ -1536,11 +1953,9 @@ export class ACPAgentSession implements AgentSession, ACPClient { throw new Error("Resume requested without persistence handle"); } - const spawned = await this.spawnProcess(); - this.child = spawned.child; - this.connection = spawned.connection; - this.agentCapabilities = spawned.initialize.agentCapabilities ?? null; + await this.attachProcess(); this.sessionId = handle.sessionId; + this.sharedProcess?.register(handle.sessionId, this); this.bootstrapThreadEventPending = true; const sessionCapabilities = this.agentCapabilities?.sessionCapabilities; @@ -1588,6 +2003,18 @@ export class ACPAgentSession implements AgentSession, ACPClient { throw error; } + private async attachProcess(): Promise { + if (this.sharedProcess) { + this.connection = this.sharedProcess.connection; + this.agentCapabilities = this.sharedProcess.initialize.agentCapabilities ?? null; + return; + } + const spawned = await this.spawnProcess(); + this.child = spawned.child; + this.connection = spawned.connection; + this.agentCapabilities = spawned.initialize.agentCapabilities ?? null; + } + async run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise { const result = await runProviderTurn({ prompt, @@ -2223,6 +2650,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { if (this.child) { await this.terminateProcess(this.child, { gracefulTimeoutMs: 2_000, forceTimeoutMs: 2_000 }); } + this.sharedProcess?.release(this.sessionId); this.subscribers.clear(); this.connection = null; @@ -2230,6 +2658,31 @@ export class ACPAgentSession implements AgentSession, ACPClient { this.activeForegroundTurnId = null; } + handleSharedProcessExit( + code: number | null, + signal: NodeJS.Signals | null, + diagnostic?: string, + ): void { + if (this.closed) { + return; + } + this.connection = null; + for (const pending of this.pendingPermissions.values()) { + pending.resolve({ outcome: { outcome: "cancelled" } }); + } + this.pendingPermissions.clear(); + if (this.activeForegroundTurnId) { + this.synthesizeCanceledToolCalls(); + this.finishTurn({ + type: "turn_failed", + provider: this.provider, + error: `Shared ACP agent exited unexpectedly (${code ?? "null"}${signal ? `, ${signal}` : ""})`, + diagnostic: diagnostic || undefined, + turnId: this.activeForegroundTurnId, + }); + } + } + async requestPermission(params: RequestPermissionRequest): Promise { const canAutoAccept = isACPAutoAcceptEnabled(this.config) && !isACPChooserRequest(params.options); @@ -2996,6 +3449,262 @@ export class ACPAgentSession implements AgentSession, ACPClient { } } +class ACPSharedClientRouter implements ACPClient { + private readonly sessions = new Map(); + private readonly pendingSessionUpdates = new Map(); + + register(sessionId: string, client: ACPSharedRouteTarget): void { + this.sessions.set(sessionId, client); + const pending = this.pendingSessionUpdates.get(sessionId); + if (pending) { + this.pendingSessionUpdates.delete(sessionId); + for (const params of pending) { + void client.sessionUpdate(params); + } + } + } + + unregister(sessionId: string | null): void { + if (sessionId) { + this.sessions.delete(sessionId); + this.pendingSessionUpdates.delete(sessionId); + } + } + + notifyProcessExit(code: number | null, signal: NodeJS.Signals | null, diagnostic?: string): void { + for (const session of this.sessions.values()) { + session.handleSharedProcessExit?.(code, signal, diagnostic); + } + this.sessions.clear(); + this.pendingSessionUpdates.clear(); + } + + async requestPermission(params: RequestPermissionRequest): Promise { + return this.session(params.sessionId).requestPermission(params); + } + + async sessionUpdate(params: SessionNotification): Promise { + const session = this.sessions.get(params.sessionId); + if (!session) { + // Agents may push session-scoped notifications (for example + // `available_commands_update`) immediately after the session/new + // response, before the client has registered the session with this + // router. Buffer them instead of dropping; register() replays them in + // arrival order once the session is known. + const pending = this.pendingSessionUpdates.get(params.sessionId) ?? []; + if (pending.length < MAX_PENDING_SESSION_UPDATES_PER_SESSION) { + pending.push(params); + this.pendingSessionUpdates.set(params.sessionId, pending); + } + return; + } + return session.sessionUpdate(params); + } + + async readTextFile(params: ReadTextFileRequest): Promise<{ content: string }> { + return this.session(params.sessionId).readTextFile(params); + } + + async writeTextFile(params: WriteTextFileRequest): Promise> { + return this.session(params.sessionId).writeTextFile(params); + } + + async createTerminal(params: CreateTerminalRequest): Promise<{ terminalId: string }> { + return this.session(params.sessionId).createTerminal(params); + } + + async terminalOutput(params: TerminalOutputRequest): Promise { + return this.session(params.sessionId).terminalOutput(params); + } + + async waitForTerminalExit(params: WaitForTerminalExitRequest): Promise { + return this.session(params.sessionId).waitForTerminalExit(params); + } + + async releaseTerminal(params: { sessionId: string; terminalId: string }): Promise { + return this.session(params.sessionId).releaseTerminal(params); + } + + async killTerminal(params: KillTerminalRequest): Promise> { + return this.session(params.sessionId).killTerminal(params); + } + + async extNotification(method: string, params: Record): Promise { + const sessionId = typeof params.sessionId === "string" ? params.sessionId : null; + if (sessionId) { + await this.session(sessionId).extNotification(method, params); + return; + } + await Promise.all( + Array.from(this.sessions.values(), (session) => session.extNotification(method, params)), + ); + } + + private session(sessionId: string): ACPSharedRouteTarget { + const session = this.sessions.get(sessionId); + if (!session) { + throw new Error(`ACP shared process sent a request for unknown session '${sessionId}'`); + } + return session; + } +} + +class ACPSharedProcessHost { + private readonly child: ChildProcessWithoutNullStreams; + private readonly connection: ClientSideConnection; + private readonly initialize: InitializeResponse; + private readonly stderrChunks: string[]; + private readonly router: ACPSharedClientRouter; + private readonly terminateProcess: ProcessTerminator; + private readonly onStopped: () => void; + private readonly stoppedPromise: Promise; + private resolveStopped!: () => void; + private references = 0; + private stopTimer: ReturnType | null = null; + private stopped = false; + private stopFinalized = false; + + constructor(options: { + child: ChildProcessWithoutNullStreams; + connection: ClientSideConnection; + initialize: InitializeResponse; + stderrChunks: string[]; + router: ACPSharedClientRouter; + terminateProcess: ProcessTerminator; + onStopped: () => void; + }) { + this.child = options.child; + this.connection = options.connection; + this.initialize = options.initialize; + this.stderrChunks = options.stderrChunks; + this.router = options.router; + this.terminateProcess = options.terminateProcess; + this.onStopped = options.onStopped; + this.stoppedPromise = new Promise((resolve) => { + this.resolveStopped = resolve; + }); + this.child.once("exit", (code, signal) => { + this.handleExit(code, signal); + }); + } + + isStopped(): boolean { + return this.stopped; + } + + hasReferences(): boolean { + return this.references > 0; + } + + whenStopped(): Promise { + return this.stoppedPromise; + } + + retire(): Promise { + return this.stop(); + } + + acquire(): ACPSharedProcessLease { + if (this.stopped) { + throw new Error("ACP shared process is no longer available"); + } + this.references += 1; + if (this.stopTimer) { + clearTimeout(this.stopTimer); + this.stopTimer = null; + } + let released = false; + return { + connection: this.connection, + initialize: this.initialize, + register: (sessionId, client) => { + this.router.register(sessionId, client); + }, + hasOtherReferences: () => this.references > 1, + invalidate: (reason) => this.invalidate(reason), + release: (sessionId) => { + if (released) { + return; + } + released = true; + this.router.unregister(sessionId); + this.references = Math.max(0, this.references - 1); + if (this.references === 0) { + this.scheduleStop(); + } + }, + }; + } + + private scheduleStop(): void { + this.stopTimer = setTimeout(() => { + this.stopTimer = null; + void this.stop(); + }, 1_000); + this.stopTimer.unref?.(); + } + + private async stop(): Promise { + if (this.stopped || this.references > 0) { + return; + } + this.stopped = true; + await this.terminateAndFinalizeWhenExited(); + } + + private async invalidate(reason: Error): Promise { + if (this.stopped) { + return this.stoppedPromise; + } + this.stopped = true; + if (this.stopTimer) { + clearTimeout(this.stopTimer); + this.stopTimer = null; + } + this.router.notifyProcessExit(null, null, reason.message); + await this.terminateAndFinalizeWhenExited(); + } + + private async terminateAndFinalizeWhenExited(): Promise { + try { + const result = await this.terminateProcess(this.child, { + gracefulTimeoutMs: 2_000, + forceTimeoutMs: 2_000, + }); + if (result !== "kill-timeout") { + this.finalizeStop(); + } + } catch { + if (typeof this.child.exitCode === "number" || this.child.signalCode != null) { + this.finalizeStop(); + } + } + } + + private handleExit(code: number | null, signal: NodeJS.Signals | null): void { + if (this.stopped) { + this.finalizeStop(); + return; + } + this.stopped = true; + if (this.stopTimer) { + clearTimeout(this.stopTimer); + this.stopTimer = null; + } + this.router.notifyProcessExit(code, signal, this.stderrChunks.join("").trim() || undefined); + this.finalizeStop(); + } + + private finalizeStop(): void { + if (this.stopFinalized) { + return; + } + this.stopFinalized = true; + this.onStopped(); + this.resolveStopped(); + } +} + export function findSelectConfigOption({ configOptions, category, @@ -3674,12 +4383,36 @@ async function terminateChildProcess( child: ChildProcessWithoutNullStreams, timeoutMs: number, terminate: ProcessTerminator, + waitForExitOnTimeout = false, ): Promise { + const exited = waitForExitOnTimeout + ? new Promise((resolve) => { + child.once("exit", () => resolve()); + }) + : null; + let result: Awaited> | null = null; + let terminationError: unknown; try { - await terminate(child, { gracefulTimeoutMs: timeoutMs, forceTimeoutMs: timeoutMs }); + result = await terminate(child, { + gracefulTimeoutMs: timeoutMs, + forceTimeoutMs: timeoutMs, + }); + } catch (error) { + terminationError = error; } finally { child.stdin.destroy(); child.stdout.destroy(); child.stderr.destroy(); } + if ( + waitForExitOnTimeout && + (result === "kill-timeout" || terminationError !== undefined) && + typeof child.exitCode !== "number" && + child.signalCode == null + ) { + await exited; + } + if (terminationError !== undefined && !waitForExitOnTimeout) { + throw terminationError; + } } diff --git a/packages/server/src/server/agent/providers/generic-acp-agent.test.ts b/packages/server/src/server/agent/providers/generic-acp-agent.test.ts index 19a1e9f8ee..3be8f04192 100644 --- a/packages/server/src/server/agent/providers/generic-acp-agent.test.ts +++ b/packages/server/src/server/agent/providers/generic-acp-agent.test.ts @@ -41,7 +41,7 @@ describe("GenericACPAgentClient", () => { }); void _client; - expect(mockState.superConstructorOptions).toEqual([ + expect(mockState.superConstructorOptions).toMatchObject([ { provider: "acp", logger: expect.any(Object), @@ -62,10 +62,37 @@ describe("GenericACPAgentClient", () => { supportsRewindFiles: false, supportsRewindBoth: false, }, + shareProcess: false, }, ]); }); + test("shares the ACP process only for the built-in Hermes provider", () => { + const sharedProcessScope = {}; + const hermesOptions = { + logger: createTestLogger(), + command: ["hermes", "acp"] as [string, ...string[]], + providerId: "hermes", + sharedProcessScope, + }; + const hermesClient = new GenericACPAgentClient({ + ...hermesOptions, + }); + const otherClient = new GenericACPAgentClient({ + logger: createTestLogger(), + command: ["other-acp"], + providerId: "other", + }); + void hermesClient; + void otherClient; + + expect(mockState.superConstructorOptions.at(-2)).toMatchObject({ + shareProcess: true, + sharedProcessScope, + }); + expect(mockState.superConstructorOptions.at(-1)).toMatchObject({ shareProcess: false }); + }); + test("uses provider params to report MCP support", () => { const _client = new GenericACPAgentClient({ logger: createTestLogger(), 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..9a797025b4 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; + sharedProcessScope?: object; } export class GenericACPAgentClient extends ACPAgentClient { @@ -76,6 +77,8 @@ export class GenericACPAgentClient extends ACPAgentClient { configFeatureOptions: options.configFeatureOptions, extensionCommandsParser: options.extensionCommandsParser, catalogModelResolver: options.catalogModelResolver, + shareProcess: options.providerId === "hermes", + sharedProcessScope: options.sharedProcessScope, }); this.command = options.command;