diff --git a/packages/plugin/src/server/acp-internal/connection.ts b/packages/plugin/src/server/acp-internal/connection.ts index 05403594cf..b2fbee8e2c 100644 --- a/packages/plugin/src/server/acp-internal/connection.ts +++ b/packages/plugin/src/server/acp-internal/connection.ts @@ -374,13 +374,19 @@ class AcpRuntime { /** * Fallback ids for streamed chunks that carry no `messageId`, one per * chunk kind. Consecutive chunks of a kind continue the same timeline - * item; a chunk of the other kind, a tool call, a plan, a user message, - * or a turn boundary ends it (#4699). + * item; a chunk of another kind, a tool call, a plan, or a turn boundary + * ends it (#4699). */ private readonly fallbackChunkIds = new Map< - "agent_message_chunk" | "agent_thought_chunk", + "agent_message_chunk" | "agent_thought_chunk" | "user_message_chunk", string >(); + /** + * Active while a persisted session loads. ACP agents replay the loaded + * session's updates before session/load resolves; the value follows the + * open request's history contract. + */ + private replayWindow: "replay" | "skip" | null = null; private closing = false; private processFailed = false; private configTransaction = false; @@ -499,15 +505,23 @@ class AcpRuntime { }; let response: Pick; if (nativeSessionId) { - response = await this.call( - this.connection.loadSession({ - sessionId: nativeSessionId, - cwd: input.config.cwd, - mcpServers, - _meta: metadata, - }), - ); + // ACP agents replay the loaded session's updates before session/load resolves, + // so the native id must be registered first for them to pass the session filter. this.nativeSessionId = nativeSessionId; + this.replayWindow = input.history === "replay" ? "replay" : "skip"; + try { + response = await this.call( + this.connection.loadSession({ + sessionId: nativeSessionId, + cwd: input.config.cwd, + mcpServers, + _meta: metadata, + }), + ); + await this.drainNotifications(); + } finally { + this.replayWindow = null; + } } else { const newSession = await this.call( this.connection.newSession({ @@ -898,6 +912,14 @@ class AcpRuntime { private sessionUpdate(notification: SessionNotification): void { if (notification.sessionId !== this.nativeSessionId) return; + if (this.replayWindow === "skip") return; + if ( + this.replayWindow === "replay" && + notification.update.sessionUpdate === "user_message_chunk" + ) { + this.reduceUserMessageChunk(notification.update); + return; + } this.reduceUpdate(notification.update); } @@ -911,13 +933,14 @@ class AcpRuntime { this.fallbackChunkIds.clear(); return; } - // A chunk of one kind ends the other kind's item, whatever its content - // or id: reasoning, then text, then reasoning again are three items. + // A chunk of one kind ends every other kind's open item, whatever its + // content or id: reasoning, then text, then reasoning again are three items. this.fallbackChunkIds.delete( update.sessionUpdate === "agent_message_chunk" ? "agent_thought_chunk" : "agent_message_chunk", ); + this.fallbackChunkIds.delete("user_message_chunk"); if (update.content.type !== "text") return; const id = this.resolveChunkId(update.sessionUpdate, update.messageId); const text = `${this.messages.get(id) ?? ""}${update.content.text}`; @@ -945,8 +968,31 @@ class AcpRuntime { this.reduceStateUpdate(update); } + // Historical user messages arrive only as replayed chunks; a live prompt + // already echoes its own user message when it is admitted. + private reduceUserMessageChunk( + update: Extract, + ): void { + this.fallbackChunkIds.delete("agent_message_chunk"); + this.fallbackChunkIds.delete("agent_thought_chunk"); + if (update.content.type !== "text") return; + const id = this.resolveChunkId("user_message_chunk", update.messageId); + const text = `${this.messages.get(id) ?? ""}${update.content.text}`; + this.messages.set(id, text); + this.emit({ + type: "timeline.item", + sessionId: this.options.boundarySessionId, + item: { + type: "user_message", + id, + text, + ...(update.messageId ? { messageId: update.messageId } : {}), + }, + }); + } + private resolveChunkId( - kind: "agent_message_chunk" | "agent_thought_chunk", + kind: "agent_message_chunk" | "agent_thought_chunk" | "user_message_chunk", messageId: string | null | undefined, ): string { if (messageId) { diff --git a/packages/plugin/src/server/acp.test.ts b/packages/plugin/src/server/acp.test.ts index 40c8019439..f4b429578f 100644 --- a/packages/plugin/src/server/acp.test.ts +++ b/packages/plugin/src/server/acp.test.ts @@ -1500,3 +1500,187 @@ describe("ACP streamed message boundaries", () => { }, ); }); + +function userChunk(text: string, messageId?: string): SessionUpdate { + return { + sessionUpdate: "user_message_chunk", + content: { type: "text", text }, + ...(messageId === undefined ? {} : { messageId }), + }; +} + +async function openReplaySession( + replay: SessionUpdate[], + history: "replay" | "skip", +): Promise<{ connection: ProviderConnection; events: ProviderEvent[] }> { + const harness = connectorHarness({ + capabilities: { loadSession: true }, + handleRequest(instance, request) { + if (request.method !== "session/load") return false; + for (const update of replay) { + instance.notify("session/update", { sessionId: "connector-session", update }); + } + instance.respond(request, { + sessionId: "connector-session", + modes: null, + configOptions: [], + }); + return true; + }, + }); + const registration = runAcpProvider({ + id: "replay-acp", + label: "Replay ACP", + connector: harness.connector, + }); + const connection = await registration.connect({ + versions: [1], + capabilities: ["prompt.message", "session.persistence"], + }); + const events: ProviderEvent[] = []; + connection.onEvent((event) => events.push(event)); + const input = openInput(); + input.persistence = { version: 1, data: { sessionId: "connector-session" } }; + input.history = history; + await connection.send(input); + await waitForEvent(events, (event) => event.type === "session.ready"); + return { connection, events }; +} + +function timelineProjection(events: ProviderEvent[]): Array> { + return events.flatMap((event) => { + if (event.type !== "timeline.item") return []; + const item = event.item; + if ( + item.type === "user_message" || + item.type === "assistant_message" || + item.type === "reasoning" + ) { + return [{ type: item.type, id: item.id, text: item.text }]; + } + if (item.type === "tool_call") return [{ type: item.type, id: item.id, status: item.status }]; + return [{ type: item.type, id: item.id }]; + }); +} + +describe("ACP session history replay", () => { + it("restores the timeline replayed before session/load resolves", async () => { + const { connection, events } = await openReplaySession( + [ + userChunk("What is the ", "user-1"), + userChunk("temperature?", "user-1"), + textChunk("Checking the sensor", "agent_thought_chunk"), + { + sessionUpdate: "tool_call", + toolCallId: "tool-1", + title: "Read sensor", + kind: "read", + status: "in_progress", + }, + { sessionUpdate: "tool_call_update", toolCallId: "tool-1", status: "completed" }, + textChunk("It is 25°C"), + ], + "replay", + ); + try { + expect(timelineProjection(events)).toEqual([ + { type: "user_message", id: "user-1", text: "What is the " }, + { type: "user_message", id: "user-1", text: "What is the temperature?" }, + { type: "reasoning", id: "agent_thought_chunk:1", text: "Checking the sensor" }, + { type: "tool_call", id: "tool-1", status: "running" }, + { type: "tool_call", id: "tool-1", status: "completed" }, + { type: "assistant_message", id: "agent_message_chunk:2", text: "It is 25°C" }, + ]); + const lastReplayed = events.findLastIndex((event) => event.type === "timeline.item"); + const opened = events.findIndex((event) => event.type === "session.opened"); + expect(lastReplayed).toBeLessThan(opened); + } finally { + await connection.close(); + } + }); + + it("continues anonymous user chunks into one message and starts a new one after an agent reply", async () => { + const { connection, events } = await openReplaySession( + [ + userChunk("first "), + userChunk("question"), + textChunk("answer"), + userChunk("second question"), + ], + "replay", + ); + try { + expect(timelineProjection(events)).toEqual([ + { type: "user_message", id: "user_message_chunk:1", text: "first " }, + { type: "user_message", id: "user_message_chunk:1", text: "first question" }, + { type: "assistant_message", id: "agent_message_chunk:2", text: "answer" }, + { type: "user_message", id: "user_message_chunk:3", text: "second question" }, + ]); + } finally { + await connection.close(); + } + }); + + it("does not duplicate the live prompt echo when the agent echoes it as a user chunk", async () => { + const harness = connectorHarness({ + capabilities: { loadSession: true }, + promptUpdates: (promptText) => [ + { sessionUpdate: "user_message_chunk", content: { type: "text", text: promptText } }, + { sessionUpdate: "agent_message_chunk", content: { type: "text", text: "Acknowledged" } }, + ], + }); + const registration = runAcpProvider({ + id: "echo-acp", + label: "Echo ACP", + connector: harness.connector, + }); + const connection = await registration.connect({ + versions: [1], + capabilities: ["prompt.message", "session.persistence"], + }); + const events: ProviderEvent[] = []; + connection.onEvent((event) => events.push(event)); + const input = openInput(); + input.persistence = { version: 1, data: { sessionId: "connector-session" } }; + input.history = "replay"; + await connection.send(input); + await waitForEvent(events, (event) => event.type === "session.ready"); + + await connection.send({ + type: "session.prompt", + sessionId: "session-1", + prompt: { + clientMessageId: "echo-1", + delivery: "auto", + input: { type: "message", content: [{ type: "text", text: "hello" }] }, + }, + }); + try { + await waitForEvent( + events, + (event) => + event.type === "session.turn" && + event.turnId === "acp:echo-1" && + event.state === "completed", + ); + expect(timelineProjection(events)).toEqual([ + { type: "user_message", id: "echo-1", text: "hello" }, + { type: "assistant_message", id: "agent_message_chunk:1", text: "Acknowledged" }, + ]); + } finally { + await connection.close(); + } + }); + + it("drops replayed updates when the open request asks to skip history", async () => { + const { connection, events } = await openReplaySession( + [userChunk("What is in the README?", "user-1"), textChunk("It describes the project.")], + "skip", + ); + try { + expect(timelineProjection(events)).toEqual([]); + } finally { + await connection.close(); + } + }); +});