From f0fac8039b2474964fcc91c198687b189d3884ac Mon Sep 17 00:00:00 2001 From: Desmond Leong Date: Thu, 13 Aug 2026 20:41:48 +0800 Subject: [PATCH] fix(server): buffer ACP session updates received before session registration ACP agents may emit session-scoped notifications immediately after the session/new response - before the client's response continuation has assigned the session id. ACPAgentSession.sessionUpdate dropped those updates silently because params.sessionId did not yet match, so an agent that pushes its available_commands_update right after session/new (for example Hermes, which advertises installed skills as slash commands) never had its commands cached and the slash-command popup stayed empty. Buffer up to 100 pre-registration session updates and replay them in arrival order once sessionId is assigned; updates addressed to a different session id are still discarded. --- .../server/agent/providers/acp-agent.test.ts | 134 ++++++++++++++++++ .../src/server/agent/providers/acp-agent.ts | 31 ++++ 2 files changed, 165 insertions(+) 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..ffc2e8ce6a 100644 --- a/packages/server/src/server/agent/providers/acp-agent.test.ts +++ b/packages/server/src/server/agent/providers/acp-agent.test.ts @@ -2403,6 +2403,140 @@ describe("ACPAgentSession slash commands", () => { }); }); +describe("ACPAgentSession pre-registration session updates", () => { + /** + * ACP agents may emit session-scoped notifications (for example + * `available_commands_update`) immediately after the `session/new` response, + * before the client's response continuation has assigned `sessionId`. These + * tests pin the buffering behavior that keeps those notifications from being + * dropped. + */ + function makeNewSession(newSession: ReturnType) { + class TestSession extends ACPAgentSession { + protected override async spawnProcess(): Promise { + return { + child: createProbeChildStub(), + connection: { + newSession, + prompt: vi.fn(), + } as unknown as ClientSideConnection, + initialize: { agentCapabilities: {} }, + } as SpawnedACPProcess; + } + } + + return new TestSession( + { provider: "hermes", cwd: "/tmp/paseo-acp-test" }, + { + provider: "hermes", + logger: createTestLogger(), + defaultCommand: ["hermes", "acp"], + defaultModes: [], + capabilities: { + supportsStreaming: true, + supportsSessionPersistence: true, + supportsDynamicModes: true, + supportsMcpServers: true, + supportsReasoningStream: true, + supportsToolInvocations: true, + }, + }, + ); + } + + test("applies available_commands_update sent with the session/new response", async () => { + let session!: ACPAgentSession; + const newSession = vi.fn().mockImplementation(async () => { + // Simulates an agent that pushes its slash-command batch immediately + // after the session/new response, before the client continuation runs. + await session.sessionUpdate({ + sessionId: "session-1", + update: { + sessionUpdate: "available_commands_update", + availableCommands: [ + { name: "what-did-you-learn", description: "Run an evidence-based retrospective" }, + ], + } as SessionUpdate, + }); + return { + sessionId: "session-1", + modes: null, + models: null, + configOptions: [], + }; + }); + session = makeNewSession(newSession); + + await session.initializeNewSession(); + + expect(await session.listCommands()).toEqual([ + { + name: "what-did-you-learn", + description: "Run an evidence-based retrospective", + argumentHint: "", + kind: "command", + }, + ]); + }); + + test("replays buffered updates in arrival order after registration", async () => { + let session!: ACPAgentSession; + const newSession = vi.fn().mockImplementation(async () => { + await session.sessionUpdate({ + sessionId: "session-1", + update: { + sessionUpdate: "available_commands_update", + availableCommands: [{ name: "first", description: "first batch" }], + } as SessionUpdate, + }); + await session.sessionUpdate({ + sessionId: "session-1", + update: { + sessionUpdate: "available_commands_update", + availableCommands: [{ name: "second", description: "second batch" }], + } as SessionUpdate, + }); + return { + sessionId: "session-1", + modes: null, + models: null, + configOptions: [], + }; + }); + session = makeNewSession(newSession); + + await session.initializeNewSession(); + + expect(await session.listCommands()).toEqual([ + { name: "second", description: "second batch", argumentHint: "", kind: "command" }, + ]); + }); + + test("ignores buffered updates addressed to a different session id", async () => { + let session!: ACPAgentSession; + const newSession = vi.fn().mockImplementation(async () => { + await session.sessionUpdate({ + sessionId: "other-session", + update: { + sessionUpdate: "available_commands_update", + availableCommands: [{ name: "stray", description: "not for this session" }], + } as SessionUpdate, + }); + return { + sessionId: "session-1", + modes: null, + models: null, + configOptions: [], + }; + }); + session = makeNewSession(newSession); + + await session.initializeNewSession(); + + expect(await session.listCommands()).toEqual([]); + }); +}); + describe("ACPAgentSession", () => { test("drops MCP servers from ACP requests when the provider does not support MCP", () => { const session = new ACPAgentSession( diff --git a/packages/server/src/server/agent/providers/acp-agent.ts b/packages/server/src/server/agent/providers/acp-agent.ts index 1a43ed5538..aa3bcdf8a9 100644 --- a/packages/server/src/server/agent/providers/acp-agent.ts +++ b/packages/server/src/server/agent/providers/acp-agent.ts @@ -272,6 +272,7 @@ 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 MAX_PRE_REGISTRATION_SESSION_UPDATES = 100; const ACP_DIAGNOSTIC_PHASE_TIMEOUT_MS = 20_000; function summarizeMalformedACPStdoutError(error: unknown): { type: string; message: string } { @@ -1440,6 +1441,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { private connection: ClientSideConnection | null = null; private agentCapabilities: ACPAgentCapabilities | null = null; private sessionId: string | null = null; + private pendingPreRegistrationUpdates: SessionNotification[] = []; private currentMode: string | null = null; private availableModes: AgentMode[]; private currentModel: string | null = null; @@ -1514,6 +1516,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { }), ); this.sessionId = response.sessionId; + this.flushPreRegistrationUpdates(); this.bootstrapThreadEventPending = true; this.applySessionState(response); await this.applyConfiguredOverrides(); @@ -2286,6 +2289,17 @@ export class ACPAgentSession implements AgentSession, ACPClient { "provider.acp.raw_event", ); if (params.sessionId !== this.sessionId) { + // Agents may push session-scoped notifications (for example + // `available_commands_update`) immediately after the session/new + // response, before the response continuation has assigned sessionId. + // Buffer them instead of dropping; they are replayed by + // flushPreRegistrationUpdates() once the session id is known. + if ( + this.sessionId === null && + this.pendingPreRegistrationUpdates.length < MAX_PRE_REGISTRATION_SESSION_UPDATES + ) { + this.pendingPreRegistrationUpdates.push(params); + } return; } @@ -2304,6 +2318,23 @@ export class ACPAgentSession implements AgentSession, ACPClient { this.deliverTranslatedEvents(events); } + /** + * Replay session notifications that arrived before the session id was + * assigned (see sessionUpdate). Notifications addressed to a different + * session id are discarded here rather than delivered. + */ + private flushPreRegistrationUpdates(): void { + const pending = this.pendingPreRegistrationUpdates; + this.pendingPreRegistrationUpdates = []; + for (const params of pending) { + if (params.sessionId !== this.sessionId) { + continue; + } + const events = this.translateSessionUpdate(params.update); + this.deliverTranslatedEvents(events); + } + } + private deliverTranslatedEvents(events: AgentStreamEvent[]): void { if (this.replayingHistory) { for (const event of events) {