fix: queue ACP requests until startup completes

.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.

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<AcpServer>): 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),
     );
 }
 

packages/acp-server/src/start.ts

@@ -130,21 +130,31 @@ 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;
+  let rejectServer!: (error: unknown) => void;
+  const serverReady = new Promise<AcpServer>((resolve, reject) => {
+    resolveServer = resolve;
+    rejectServer = reject;
+  });
+  void serverReady.catch(() => {});
+  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
   // file IO. The `acp` `IHostFileSystem` reads it lazily via
   // `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);
-  server = new AcpServer(client, klient, acpConnection, {
+  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,
     terminalAuthEnv: opts.terminalAuthEnv,
@@ -176,6 +186,7 @@ export async function runAcpServerWithStream(
         : sessionMediaOriginalsDir(handle.accessor.get(ISessionContext).sessionDir);
     },
   });
+  resolveServer(server);
 
   let closePromise: Promise<void> | undefined;
   const close = async (): Promise<void> => {

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';
@@ -98,6 +99,81 @@ 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(
+    '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 () => {