From 48d3bcc6baf5d77127264b281e852704684e8562 Mon Sep 17 00:00:00 2001 From: 7Sageer Date: Fri, 30 Oct 2026 11:40:50 +0800 Subject: [PATCH 1/2] fix: queue ACP requests until startup completes --- .changeset/fix-acp-startup-race.md | 5 ++ packages/acp-server/src/server.ts | 56 ++++++++++++--------- packages/acp-server/src/start.ts | 16 +++--- packages/acp-server/test/initialize.test.ts | 34 +++++++++++++ 4 files changed, 80 insertions(+), 31 deletions(-) create mode 100644 .changeset/fix-acp-startup-race.md diff --git a/.changeset/fix-acp-startup-race.md b/.changeset/fix-acp-startup-race.md new file mode 100644 index 00000000000..7c26ceb11f3 --- /dev/null +++ b/.changeset/fix-acp-startup-race.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Fix ACP initialization failing when clients send the first request during startup. diff --git a/packages/acp-server/src/server.ts b/packages/acp-server/src/server.ts index 51033750461..2c2e1ffd299 100644 --- a/packages/acp-server/src/server.ts +++ b/packages/acp-server/src/server.ts @@ -691,33 +691,41 @@ function parseSetSessionModelParams(params: unknown): SetSessionModelParams { * Build the ACP agent app (SDK `agent()` builder) that routes every inbound * method to the {@link AcpServer} returned by `getServer`. * - * `getServer` is dereferenced lazily per request because the app must be - * connected before the outbound client surface (and thus the server) exists - * — see `./start`. Handlers for unregistered methods never arise here: the - * connection layer answers unknown requests with `method_not_found` (-32601) - * and silently drops unknown notifications, matching the legacy - * `extMethod` / `extNotification` fallbacks. + * `getServer` is awaited per request because the app must be connected before + * the outbound client surface (and thus the server) exists, and the environment + * provider attaches asynchronously — see `./start`. Handlers for unregistered + * methods never arise here: the connection layer answers unknown requests with + * `method_not_found` (-32601) and silently drops unknown notifications, matching + * the legacy `extMethod` / `extNotification` fallbacks. */ -export function createAcpAgentApp(getServer: () => AcpServer): AgentApp { +export function createAcpAgentApp(getServer: () => AcpServer | Promise): AgentApp { return agent({ name: 'kimi-code-acp' }) - .onRequest(methods.agent.initialize, (ctx) => getServer().initialize(ctx.params)) - .onRequest(methods.agent.authenticate, (ctx) => getServer().authenticate(ctx.params)) - .onRequest(methods.agent.logout, (ctx) => getServer().logout(ctx.params)) - .onRequest(methods.agent.session.new, (ctx) => getServer().newSession(ctx.params)) - .onRequest(methods.agent.session.load, (ctx) => getServer().loadSession(ctx.params)) - .onRequest(methods.agent.session.resume, (ctx) => getServer().resumeSession(ctx.params)) - .onRequest(methods.agent.session.list, (ctx) => getServer().listSessions(ctx.params)) - .onRequest(methods.agent.session.close, (ctx) => getServer().closeSession(ctx.params)) - .onRequest(methods.agent.session.delete, (ctx) => getServer().deleteSession(ctx.params)) - .onRequest(methods.agent.session.fork, (ctx) => getServer().unstable_forkSession(ctx.params)) - .onRequest(methods.agent.session.setMode, (ctx) => getServer().setSessionMode(ctx.params)) - .onRequest(methods.agent.session.setConfigOption, (ctx) => - getServer().setSessionConfigOption(ctx.params), + .onRequest(methods.agent.initialize, async (ctx) => (await getServer()).initialize(ctx.params)) + .onRequest(methods.agent.authenticate, async (ctx) => (await getServer()).authenticate(ctx.params)) + .onRequest(methods.agent.logout, async (ctx) => (await getServer()).logout(ctx.params)) + .onRequest(methods.agent.session.new, async (ctx) => (await getServer()).newSession(ctx.params)) + .onRequest(methods.agent.session.load, async (ctx) => (await getServer()).loadSession(ctx.params)) + .onRequest(methods.agent.session.resume, async (ctx) => (await getServer()).resumeSession(ctx.params)) + .onRequest(methods.agent.session.list, async (ctx) => (await getServer()).listSessions(ctx.params)) + .onRequest(methods.agent.session.close, async (ctx) => (await getServer()).closeSession(ctx.params)) + .onRequest(methods.agent.session.delete, async (ctx) => (await getServer()).deleteSession(ctx.params)) + .onRequest(methods.agent.session.fork, async (ctx) => + (await getServer()).unstable_forkSession(ctx.params), ) - .onRequest(methods.agent.session.prompt, (ctx) => getServer().prompt(ctx.params, ctx.signal)) - .onNotification(methods.agent.session.cancel, (ctx) => getServer().cancel(ctx.params)) - .onRequest(SET_SESSION_MODEL_METHOD, parseSetSessionModelParams, (ctx) => - getServer().setSessionModel(ctx.params), + .onRequest(methods.agent.session.setMode, async (ctx) => + (await getServer()).setSessionMode(ctx.params), + ) + .onRequest(methods.agent.session.setConfigOption, async (ctx) => + (await getServer()).setSessionConfigOption(ctx.params), + ) + .onRequest(methods.agent.session.prompt, async (ctx) => + (await getServer()).prompt(ctx.params, ctx.signal), + ) + .onNotification(methods.agent.session.cancel, async (ctx) => + (await getServer()).cancel(ctx.params), + ) + .onRequest(SET_SESSION_MODEL_METHOD, parseSetSessionModelParams, async (ctx) => + (await getServer()).setSessionModel(ctx.params), ); } diff --git a/packages/acp-server/src/start.ts b/packages/acp-server/src/start.ts index b8a644ae6df..617bdb060dd 100644 --- a/packages/acp-server/src/start.ts +++ b/packages/acp-server/src/start.ts @@ -130,12 +130,13 @@ export async function runAcpServerWithStream( // Route every inbound ACP method to the `AcpServer`. The app must be // connected before the outbound client surface (`conn.client`) — and thus - // the server — exists, so handlers dereference `server` lazily. No inbound - // message can be dispatched before this synchronous block yields (the - // connection's reader only runs on later microtasks), so `server` is always - // assigned by the time a handler fires. - let server: AcpServer; - const app = createAcpAgentApp(() => server); + // the server — exists. Handlers wait for `serverReady` so requests received + // while the environment provider is attaching run only after construction. + let resolveServer!: (server: AcpServer) => void; + const serverReady = new Promise((resolve) => { + resolveServer = resolve; + }); + const app = createAcpAgentApp(() => serverReady); const conn = app.connect(stream); const client = acpClientFromContext(conn.client); // Bind the process-wide ACP client connection before any session performs @@ -144,7 +145,7 @@ export async function runAcpServerWithStream( acpConnection.bind(client); const acpEnvironmentProvider = new AcpEnvironmentProviderFactory(acpConnection, core.accessor.get(IHostEnvironment), core.accessor.get(IHostProcessService)); const acpProviderRegistration = await core.accessor.get(IEnvironmentService).addProvider(acpEnvironmentProvider); - server = new AcpServer(client, klient, acpConnection, { + const server = new AcpServer(client, klient, acpConnection, { agentInfo: opts.agentInfo, disableAuth: opts.disableAuth, terminalAuthEnv: opts.terminalAuthEnv, @@ -176,6 +177,7 @@ export async function runAcpServerWithStream( : sessionMediaOriginalsDir(handle.accessor.get(ISessionContext).sessionDir); }, }); + resolveServer(server); let closePromise: Promise | undefined; const close = async (): Promise => { diff --git a/packages/acp-server/test/initialize.test.ts b/packages/acp-server/test/initialize.test.ts index 1254db30bd4..5d0cb3d4dde 100644 --- a/packages/acp-server/test/initialize.test.ts +++ b/packages/acp-server/test/initialize.test.ts @@ -98,6 +98,40 @@ describe('acp-server initialize handshake', () => { 30_000, ); + it( + 'answers an initialize request sent before startup completes', + async () => { + const homeDir = await mkdtemp(join(tmpdir(), 'acp-server-early-init-')); + const toAgent = new PassThrough(); + const toClient = new PassThrough(); + try { + const stream = ndJsonStream(Writable.toWeb(toClient), Readable.toWeb(toAgent)); + const serverPromise = runAcpServerWithStream(stream, { homeDir }); + toAgent.write( + `${JSON.stringify({ + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { protocolVersion: 1, clientCapabilities: {} }, + })}\n`, + ); + + const response = await readOneMessage(toClient); + const server = await serverPromise; + await server.close(); + toAgent.end(); + toClient.end(); + + expect(response.id).toBe(1); + expect(response.error).toBeUndefined(); + expect(response.result).toMatchObject({ protocolVersion: 1 }); + } finally { + await rm(homeDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 }); + } + }, + 30_000, + ); + it( 'negotiates down to the highest supported version when the client advertises a newer one', async () => { From 425121eb1e356145b2064f2205e95796a4fcea5a Mon Sep 17 00:00:00 2001 From: 7Sageer Date: Fri, 30 Oct 2026 11:47:31 +0800 Subject: [PATCH 2/2] fix: reject queued ACP requests on startup failure --- packages/acp-server/src/start.ts | 13 ++++++- packages/acp-server/test/initialize.test.ts | 42 +++++++++++++++++++++ 2 files changed, 53 insertions(+), 2 deletions(-) diff --git a/packages/acp-server/src/start.ts b/packages/acp-server/src/start.ts index 617bdb060dd..d5545269646 100644 --- a/packages/acp-server/src/start.ts +++ b/packages/acp-server/src/start.ts @@ -133,9 +133,12 @@ export async function runAcpServerWithStream( // the server — exists. Handlers wait for `serverReady` so requests received // while the environment provider is attaching run only after construction. let resolveServer!: (server: AcpServer) => void; - const serverReady = new Promise((resolve) => { + let rejectServer!: (error: unknown) => void; + const serverReady = new Promise((resolve, reject) => { resolveServer = resolve; + rejectServer = reject; }); + void serverReady.catch(() => {}); const app = createAcpAgentApp(() => serverReady); const conn = app.connect(stream); const client = acpClientFromContext(conn.client); @@ -144,7 +147,13 @@ export async function runAcpServerWithStream( // `IAcpConnection.get()`. acpConnection.bind(client); const acpEnvironmentProvider = new AcpEnvironmentProviderFactory(acpConnection, core.accessor.get(IHostEnvironment), core.accessor.get(IHostProcessService)); - const acpProviderRegistration = await core.accessor.get(IEnvironmentService).addProvider(acpEnvironmentProvider); + const acpProviderRegistration = await core.accessor + .get(IEnvironmentService) + .addProvider(acpEnvironmentProvider) + .catch((error: unknown) => { + rejectServer(error); + throw error; + }); const server = new AcpServer(client, klient, acpConnection, { agentInfo: opts.agentInfo, disableAuth: opts.disableAuth, diff --git a/packages/acp-server/test/initialize.test.ts b/packages/acp-server/test/initialize.test.ts index 5d0cb3d4dde..b7b53fbe874 100644 --- a/packages/acp-server/test/initialize.test.ts +++ b/packages/acp-server/test/initialize.test.ts @@ -4,6 +4,7 @@ import { join } from 'node:path'; import { PassThrough, Readable, Writable } from 'node:stream'; import { ndJsonStream } from '@agentclientprotocol/sdk'; +import { IEnvironmentService } from '@moonshot-ai/agent-core-v2'; import { describe, expect, it } from 'vitest'; import { runAcpServerWithStream } from '../src/start'; @@ -132,6 +133,47 @@ describe('acp-server initialize handshake', () => { 30_000, ); + it( + 'rejects an early initialize request when startup fails', + async () => { + const homeDir = await mkdtemp(join(tmpdir(), 'acp-server-failed-init-')); + const toAgent = new PassThrough(); + const toClient = new PassThrough(); + const startupError = new Error('environment attach failed'); + try { + const stream = ndJsonStream(Writable.toWeb(toClient), Readable.toWeb(toAgent)); + const environmentService = { + addProvider: () => Promise.reject(startupError), + } as unknown as IEnvironmentService; + const serverPromise = runAcpServerWithStream(stream, { + homeDir, + extraSeeds: [[IEnvironmentService, environmentService]], + }); + toAgent.write( + `${JSON.stringify({ + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { protocolVersion: 1, clientCapabilities: {} }, + })}\n`, + ); + + const responsePromise = readOneMessage(toClient); + await expect(serverPromise).rejects.toBe(startupError); + const response = await responsePromise; + expect(response.error).toMatchObject({ + code: -32603, + data: { details: startupError.message }, + }); + } finally { + toAgent.end(); + toClient.end(); + await rm(homeDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 }); + } + }, + 5_000, + ); + it( 'negotiates down to the highest supported version when the client advertises a newer one', async () => {