diff --git a/src/supervisor/agents/acp/session.test.ts b/src/supervisor/agents/acp/session.test.ts index c488be2cc..cdc5c24d6 100644 --- a/src/supervisor/agents/acp/session.test.ts +++ b/src/supervisor/agents/acp/session.test.ts @@ -71,11 +71,15 @@ function makeConfigSyncSession( currentConfig?: ThreadConfig; agentMcpCapabilities?: { http?: boolean; sse?: boolean } | undefined; assumedMcpCapabilities?: { http?: boolean; sse?: boolean }; + optimisticMcpTransports?: readonly ("stdio" | "http" | "sse")[]; mcpServers?: Array<{ id: string; name: string; timeoutMs: number; - transport: { type: "http"; url: string; headers: Record }; + transport: + | { type: "http"; url: string; headers: Record } + | { type: "sse"; url: string; headers: Record } + | { type: "stdio"; command: string; args: string[]; env: Record }; }>; fsTextCapability?: boolean; initializeMeta?: Record; @@ -163,6 +167,7 @@ function makeConfigSyncSession( session["agentMcpCapabilities"] = "agentMcpCapabilities" in overrides ? overrides.agentMcpCapabilities : { http: true }; session["assumedMcpCapabilities"] = overrides.assumedMcpCapabilities; + session["optimisticMcpTransports"] = overrides.optimisticMcpTransports; session["currentConfig"] = overrides.currentConfig ?? { model: "model-a", effort: "low", @@ -1616,6 +1621,123 @@ describe("ACP client protocol helpers", () => { expect(connection.newSession).toHaveBeenCalledTimes(1); }); + it("relays optimistic stdio transports on the first attempt and keeps them when accepted", async () => { + // Kimi-shaped agent: advertises http/sse but has no way to advertise that + // it lacks stdio (the ACP schema has no such flag). Once it grows support + // the servers must flow with no code change, so the first attempt carries + // them and success keeps them. + const { connection, session } = makeConfigSyncSession({ + agentMcpCapabilities: { http: true, sse: true }, + optimisticMcpTransports: ["stdio"], + mcpServers: [ + { + id: "fs", + name: "fs", + timeoutMs: 30_000, + transport: { type: "stdio", command: "npx", args: ["-y", "fs-mcp"], env: {} }, + }, + { + id: "browser", + name: "browser", + timeoutMs: 30_000, + transport: { + type: "http", + url: "http://127.0.0.1:9123/mcp", + headers: {}, + }, + }, + ], + }); + + await expect(session.openThread({ model: "model-a", browserMcp: true })).resolves.toBe( + "session-1", + ); + + expect(connection.newSession).toHaveBeenCalledTimes(1); + expect(connection.newSession).toHaveBeenCalledWith({ + cwd: "C:\\repo", + mcpServers: [ + { name: "fs", command: "npx", args: ["-y", "fs-mcp"], env: [] }, + { type: "http", name: "browser", url: "http://127.0.0.1:9123/mcp", headers: [] }, + ], + }); + }); + + it("retries without optimistic stdio transports on Kimi's runtime-identity failure", async () => { + const log = vi.spyOn(console, "log").mockImplementation(() => undefined); + const { connection, session } = makeConfigSyncSession({ + agentMcpCapabilities: { http: true, sse: true }, + optimisticMcpTransports: ["stdio"], + mcpServers: [ + { + id: "fs", + name: "fs", + timeoutMs: 30_000, + transport: { type: "stdio", command: "npx", args: ["server"], env: {} }, + }, + { + id: "browser", + name: "browser", + timeoutMs: 30_000, + transport: { + type: "http", + url: "http://127.0.0.1:9123/mcp", + headers: {}, + }, + }, + ], + }); + // Kimi 0.38.0 surfaces the converter throw as a bare -32603 whose data + // carries the details string (MoonshotAI/kimi-code#3069). + connection.newSession + .mockRejectedValueOnce( + RequestError.internalError({ + details: "ACP stdio MCP server fs does not declare a runtime identity", + }), + ) + .mockResolvedValueOnce({ + sessionId: "session-1", + modes: { availableModes: [] }, + configOptions: [], + }); + + try { + await expect(session.openThread({ model: "model-a", browserMcp: true })).resolves.toBe( + "session-1", + ); + + expect(connection.newSession).toHaveBeenCalledTimes(2); + // The retry keeps remote servers (advertised) and drops only stdio. + expect(connection.newSession).toHaveBeenLastCalledWith({ + cwd: "C:\\repo", + mcpServers: [ + { type: "http", name: "browser", url: "http://127.0.0.1:9123/mcp", headers: [] }, + ], + }); + } finally { + log.mockRestore(); + } + }); + + it("does not retry optimistic stdio transports after an unrelated session-open failure", async () => { + const { connection, session } = makeConfigSyncSession({ + agentMcpCapabilities: { http: true, sse: true }, + optimisticMcpTransports: ["stdio"], + mcpServers: [ + { + id: "fs", + name: "fs", + timeoutMs: 30_000, + transport: { type: "stdio", command: "npx", args: ["server"], env: {} }, + }, + ], + }); + connection.newSession.mockRejectedValueOnce(new Error("transport closed")); + + await expect(session.openThread({ model: "model-a" })).rejects.toThrow("transport closed"); + expect(connection.newSession).toHaveBeenCalledTimes(1); + }); + it("preserves stale-session invalidParams instead of retrying load without MCP servers", async () => { const { connection, session } = makeConfigSyncSession({ agentMcpCapabilities: undefined, diff --git a/src/supervisor/agents/acp/session.ts b/src/supervisor/agents/acp/session.ts index 53a629456..dbe785833 100644 --- a/src/supervisor/agents/acp/session.ts +++ b/src/supervisor/agents/acp/session.ts @@ -54,6 +54,7 @@ import type { ThreadServerRequestId, ThreadStatus, ResolvedMcpServer, + McpTransportKind, } from "@/shared/contracts"; import { areAgentSlashCommandsEqual } from "@/shared/contracts"; import { buildPromptContentBlocks } from "@/shared/promptContent"; @@ -234,6 +235,14 @@ export interface AcpStructuredSessionOptions { * still falls back to the strictly gated set if opening fails. */ assumedMcpCapabilities?: AcpMcpCapabilities; + /** + * MCP transports relayed optimistically: sent on the first open attempt and + * excluded from the compatibility-failure retry set. For agents that fail + * session-open on a transport the ACP schema gives them no way to decline + * (stdio has no capability flag) — see `acpOptimisticMcpTransports` in the + * adapter contract. + */ + optimisticMcpTransports?: readonly McpTransportKind[]; /** * Home-relative directories (posix-style, e.g. ".kimi-code") the agent may * read and write through the ACP fs bridge even though they sit outside the @@ -281,6 +290,7 @@ export class AcpStructuredSession implements StructuredSessionHandle { private readonly projectLocation: ProjectLocation; private readonly mcpServers: readonly ResolvedMcpServer[]; private readonly assumedMcpCapabilities: AcpMcpCapabilities | undefined; + private readonly optimisticMcpTransports: readonly McpTransportKind[] | undefined; private readonly fsAgentHomeDirs: readonly string[]; private readonly fsTextCapability: boolean; private planModeToolTrackerInstance: AcpPlanModeToolTracker | undefined; @@ -450,6 +460,7 @@ export class AcpStructuredSession implements StructuredSessionHandle { } this.mcpServers = options?.mcpServers ?? []; this.assumedMcpCapabilities = options?.assumedMcpCapabilities; + this.optimisticMcpTransports = options?.optimisticMcpTransports; this.fsAgentHomeDirs = options?.fsAgentHomeDirs ?? []; this.fsTextCapability = options?.fsTextCapability !== false; } @@ -764,27 +775,44 @@ export class AcpStructuredSession implements StructuredSessionHandle { private async openWithMcpServers( open: (mcpServers: ProtocolMcpServer[]) => Promise, ): Promise { - const built = buildAcpMcpServers(this.mcpServers); - const assumed = this.gateMcpServers( - built, - resolveAcpMcpCapabilities(this.agentMcpCapabilities, this.assumedMcpCapabilities), + const capabilities = resolveAcpMcpCapabilities( + this.agentMcpCapabilities, + this.assumedMcpCapabilities, ); - const advertised = - this.agentMcpCapabilities === undefined && this.assumedMcpCapabilities !== undefined - ? gateAcpMcpServers(built, this.agentMcpCapabilities) - : assumed; - if (advertised.length === assumed.length) return open(assumed); + const built = buildAcpMcpServers(this.mcpServers); + const attempted = this.gateMcpServers(built, capabilities); + const optimisticTransports = this.optimisticMcpTransports; + let fallback: ProtocolMcpServer[]; + if (optimisticTransports !== undefined && optimisticTransports.length > 0) { + // Optimistic transports ride along on the first attempt only; the retry + // set excludes them so a compatibility failure can still open the + // session without them. + fallback = gateAcpMcpServers( + buildAcpMcpServers( + this.mcpServers.filter((server) => !optimisticTransports.includes(server.transport.type)), + ), + capabilities, + ); + } else if ( + this.agentMcpCapabilities === undefined && + this.assumedMcpCapabilities !== undefined + ) { + fallback = gateAcpMcpServers(built, this.agentMcpCapabilities); + } else { + fallback = attempted; + } + if (fallback.length === attempted.length) return open(attempted); try { - return await open(assumed); + return await open(attempted); } catch (error) { if (!isAssumedMcpCompatibilityError(error)) throw error; console.log( "[acp] session open failed with %d assumed-transport MCP server(s) (ACP error %d); retrying without them", - assumed.length - advertised.length, + attempted.length - fallback.length, error.code, ); - return open(advertised); + return open(fallback); } } diff --git a/src/supervisor/agents/acp/sessionFactory.ts b/src/supervisor/agents/acp/sessionFactory.ts index 91cafb245..1cfe27c8f 100644 --- a/src/supervisor/agents/acp/sessionFactory.ts +++ b/src/supervisor/agents/acp/sessionFactory.ts @@ -74,6 +74,9 @@ export function createAcpStructuredSession( ? { extensionNotificationHandler: input.acpExtensionNotificationHandler } : {}), ...(input.mcpServers !== undefined ? { mcpServers: input.mcpServers } : {}), + ...(input.acpOptimisticMcpTransports + ? { optimisticMcpTransports: input.acpOptimisticMcpTransports } + : {}), ...(input.acpFsAgentHomeDirs ? { fsAgentHomeDirs: input.acpFsAgentHomeDirs } : {}), ...(input.acpFsTextCapability !== undefined ? { fsTextCapability: input.acpFsTextCapability } diff --git a/src/supervisor/agents/base/types.ts b/src/supervisor/agents/base/types.ts index b7d439427..0d82c274b 100644 --- a/src/supervisor/agents/base/types.ts +++ b/src/supervisor/agents/base/types.ts @@ -17,6 +17,7 @@ import type { ThreadServerRequestId, ThreadStatus, ThreadGoalControl, + McpTransportKind, ResolvedMcpServer, } from "@/shared/contracts"; import type { OscNotification, OscShellEvent, OscTitle } from "@/shared/osc"; @@ -230,6 +231,16 @@ export interface CreateStructuredSessionInput { * is the same content the bridge would have served. */ acpFsTextCapability?: boolean; + /** + * MCP transports relayed optimistically: included in the first + * `session/new` / `session/load` attempt and dropped from the retry set if + * opening fails with a protocol compatibility error. Use for agents that + * fail session-open on a transport the ACP schema gives them no way to + * decline (stdio has no capability flag): the worst case is one failed + * roundtrip per launch, and an agent that grows support for the transport + * picks its servers up with no code change here. + */ + acpOptimisticMcpTransports?: readonly McpTransportKind[]; } export type AcpEmptyResponseErrorResolver = (input: { diff --git a/src/supervisor/agents/kimi/acpFsCapability.test.ts b/src/supervisor/agents/kimi/acpFsCapability.test.ts index 81e15b01f..288100d47 100644 --- a/src/supervisor/agents/kimi/acpFsCapability.test.ts +++ b/src/supervisor/agents/kimi/acpFsCapability.test.ts @@ -10,16 +10,18 @@ vi.mock("./kimiTrust", () => ({ ensureKimiWorkspaceTrust: vi.fn<() => Promise>(async () => {}), })); -import type { ProjectLocation, ThreadConfig } from "@/shared/contracts"; +import type { ProjectLocation, ResolvedMcpServer, ThreadConfig } from "@/shared/contracts"; import { createAcpStructuredSession } from "../acp"; import { createKimiAdapter } from "./index"; -async function createSessionOptions() { +async function createSessionOptions(mcpServers?: ResolvedMcpServer[]) { + vi.mocked(createAcpStructuredSession).mockClear(); const adapter = createKimiAdapter(); await adapter.createStructuredSession?.({ threadId: "thread-1", projectLocation: { kind: "windows", path: "C:\\repo" } as ProjectLocation, config: { mode: "agent" } as ThreadConfig, + ...(mcpServers ? { mcpServers } : {}), }); return vi.mocked(createAcpStructuredSession).mock.calls[0]?.[1]; } @@ -41,3 +43,43 @@ describe("Kimi ACP fs capability", () => { expect(await createSessionOptions()).not.toHaveProperty("acpFsAgentHomeDirs"); }); }); + +describe("Kimi ACP MCP compatibility", () => { + it("relays every server and marks stdio transports optimistic", async () => { + // Kimi's ACP server fails session/new on stdio MCP servers ("does not + // declare a runtime identity") but has no mcpCapabilities flag to say so. + // The adapter must not pre-filter: it marks stdio optimistic so the shared + // session retries once without them on that failure — and relays them for + // free once Kimi ships support (MoonshotAI/kimi-code#3069). + const servers: ResolvedMcpServer[] = [ + { + id: "stdio", + name: "stdio", + timeoutMs: 30_000, + transport: { type: "stdio", command: "npx", args: ["server"], env: {} }, + }, + { + id: "http", + name: "http", + timeoutMs: 30_000, + transport: { type: "http", url: "http://127.0.0.1:9000/mcp", headers: {} }, + }, + { + id: "sse", + name: "sse", + timeoutMs: 30_000, + transport: { type: "sse", url: "http://127.0.0.1:9001/sse", headers: {} }, + }, + { + id: "stdio-2", + name: "stdio-2", + timeoutMs: 30_000, + transport: { type: "stdio", command: "node", args: ["server.js"], env: {} }, + }, + ]; + + const options = await createSessionOptions(servers); + expect(options?.mcpServers).toEqual(servers); + expect(options?.acpOptimisticMcpTransports).toEqual(["stdio"]); + }); +}); diff --git a/src/supervisor/agents/kimi/index.ts b/src/supervisor/agents/kimi/index.ts index 79c35c027..22eeb848a 100644 --- a/src/supervisor/agents/kimi/index.ts +++ b/src/supervisor/agents/kimi/index.ts @@ -144,6 +144,15 @@ export function createKimiAdapter(): AgentAdapter { ); session = createAcpStructuredSession(command, { ...input, + // Kimi's ACP server rejects the protocol's standard stdio MCP shape + // during session/new ("ACP stdio MCP server does not declare a + // runtime identity", surfaced only as -32603 Internal error), and the + // ACP schema gives it no mcpCapabilities flag to declare that gap. + // Relay stdio servers optimistically instead of pre-dropping them: + // the shared session retries once without them on that exact failure + // (MoonshotAI/kimi-code#3069, fix pending in #3070) and, once a Kimi + // release ships support, relays them with no change here. + acpOptimisticMcpTransports: ["stdio"], // Kimi's ACP host filesystem routes *every* text read/write through // the client once fs capability is advertised — including its own // per-session state under ~/.kimi-code — and rethrows the client's