From f94b488c4fb663d764ffceeb97d3fb4c791fce8e Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Tue, 23 Jun 2026 22:25:16 +0700 Subject: [PATCH] Fix ACP user message echo dedupe --- docs/providers.md | 2 +- .../server/agent/providers/acp-agent.test.ts | 124 ++++++++++++++++++ .../src/server/agent/providers/acp-agent.ts | 43 +++++- 3 files changed, 166 insertions(+), 3 deletions(-) diff --git a/docs/providers.md b/docs/providers.md index e516a687c..b2f1ec0b6 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -32,7 +32,7 @@ OpenCode MCP injection is dynamic and session-scoped. Call OpenCode's `mcp.add` OpenCode owns user message IDs. Do not pass Paseo-generated IDs to OpenCode prompt APIs; let OpenCode create `msg*` IDs and record the user timeline item from the `message.updated` event. -Every provider adapter owns its canonical user-message timeline rows. When a foreground prompt is accepted, the adapter must emit exactly one `user_message` timeline item for that submitted prompt, using the same message ID it gives to or receives from the provider runtime. Optimistic client messages are UI-only and provider transcript echoes are optional; neither is allowed to be the only source of truth. If the provider later echoes the same submitted user message, dedupe by provider-visible message ID, not by text. +Every provider adapter owns its canonical user-message timeline rows. When a foreground prompt is accepted, the adapter must emit exactly one `user_message` timeline item for that submitted prompt, using the same message ID it gives to or receives from the provider runtime. Optimistic client messages are UI-only and provider transcript echoes are optional; neither is allowed to be the only source of truth. If the provider later echoes the same submitted user message, dedupe it only within the active turn. Prefer provider-visible message IDs, but ACP runtimes may omit that ID or replace it with a provider-owned one; in that case suppress only echo chunks whose accumulated text is a prefix of the active submitted prompt. Do not perform global transcript text dedupe. Draft metadata lookups should avoid creating provider sessions when the upstream provider has top-level APIs for that metadata. Prefer `AgentClient.fetchCatalog`, `listCommands`, or `listFeatures` over creating a scratch `AgentSession`; scratch sessions can show up as empty native sessions in provider import/history UIs. `fetchCatalog` is the single discovery API for models and modes — provider implementations may use one process, separate upstream calls, or static data internally, but callers outside the provider do not get separate runtime model/mode probes. 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 75f90f1ff..5423e9af2 100644 --- a/packages/server/src/server/agent/providers/acp-agent.test.ts +++ b/packages/server/src/server/agent/providers/acp-agent.test.ts @@ -1916,6 +1916,130 @@ describe("ACPAgentSession", () => { ).toHaveLength(1); }); + test("startTurn dedupes ACP user echo chunks without message ids for the submitted message", async () => { + const session = createSession(); + const events: AgentStreamEvent[] = []; + const prompt = vi.fn(() => new Promise(() => {})); + + asInternals(session).sessionId = "session-1"; + asInternals(session).connection = { prompt }; + + session.subscribe((event) => { + events.push(event); + }); + + await session.startTurn("hello", { messageId: "msg-client-1" }); + await session.sessionUpdate({ + sessionId: "session-1", + update: { + sessionUpdate: "user_message_chunk", + content: { type: "text", text: "hello" }, + } as SessionUpdate, + }); + + expect( + events.filter((event) => event.type === "timeline" && event.item.type === "user_message"), + ).toEqual([ + { + type: "timeline", + provider: "claude-acp", + item: { type: "user_message", text: "hello", messageId: "msg-client-1" }, + turnId: expect.any(String), + }, + ]); + }); + + test("startTurn dedupes ACP user echo chunks without message ids across turns", async () => { + const session = createSession(); + const events: AgentStreamEvent[] = []; + let resolvePrompt!: (value: PromptResponse) => void; + const prompt = vi.fn( + () => + new Promise((resolve) => { + resolvePrompt = resolve; + }), + ); + + asInternals(session).sessionId = "session-1"; + asInternals(session).connection = { prompt }; + + session.subscribe((event) => { + events.push(event); + }); + + await session.startTurn("first", { messageId: "msg-client-1" }); + await session.sessionUpdate({ + sessionId: "session-1", + update: { + sessionUpdate: "user_message_chunk", + content: { type: "text", text: "first" }, + } as SessionUpdate, + }); + resolvePrompt({ stopReason: "end_turn" }); + await Promise.resolve(); + await Promise.resolve(); + + await session.startTurn("second", { messageId: "msg-client-2" }); + await session.sessionUpdate({ + sessionId: "session-1", + update: { + sessionUpdate: "user_message_chunk", + content: { type: "text", text: "second" }, + } as SessionUpdate, + }); + + expect( + events.filter((event) => event.type === "timeline" && event.item.type === "user_message"), + ).toEqual([ + { + type: "timeline", + provider: "claude-acp", + item: { type: "user_message", text: "first", messageId: "msg-client-1" }, + turnId: expect.any(String), + }, + { + type: "timeline", + provider: "claude-acp", + item: { type: "user_message", text: "second", messageId: "msg-client-2" }, + turnId: expect.any(String), + }, + ]); + }); + + test("startTurn dedupes ACP user echo chunks with provider-owned ids for the submitted message", async () => { + const session = createSession(); + const events: AgentStreamEvent[] = []; + const prompt = vi.fn(() => new Promise(() => {})); + + asInternals(session).sessionId = "session-1"; + asInternals(session).connection = { prompt }; + + session.subscribe((event) => { + events.push(event); + }); + + await session.startTurn("hello", { messageId: "msg-client-1" }); + await session.sessionUpdate({ + sessionId: "session-1", + update: { + sessionUpdate: "user_message_chunk", + messageId: "msg-provider-1", + content: { type: "text", text: "hello" }, + } as SessionUpdate, + }); + + expect( + events.filter((event) => event.type === "timeline" && event.item.type === "user_message"), + ).toEqual([ + { + type: "timeline", + provider: "claude-acp", + item: { type: "user_message", text: "hello", messageId: "msg-client-1" }, + turnId: expect.any(String), + }, + ]); + }); + test("startTurn converts background prompt rejections into turn_failed events", async () => { const session = createSession(); const events: Array<{ type: string; turnId?: string; error?: string }> = []; diff --git a/packages/server/src/server/agent/providers/acp-agent.ts b/packages/server/src/server/agent/providers/acp-agent.ts index aafe5f3a2..ed09b3918 100644 --- a/packages/server/src/server/agent/providers/acp-agent.ts +++ b/packages/server/src/server/agent/providers/acp-agent.ts @@ -393,6 +393,12 @@ interface MessageAssemblyState { text: string; } +interface SubmittedUserMessageEcho { + messageId: string; + text: string; + turnId: string; +} + export type SessionStateResponse = NewSessionResponse | LoadSessionResponse | ResumeSessionResponse; interface TerminalExit { @@ -978,6 +984,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { private readonly pendingPermissions = new Map(); private readonly messageAssemblies = new Map(); private readonly submittedUserMessageIds = new Set(); + private activeSubmittedUserMessage: SubmittedUserMessageEcho | null = null; private readonly toolCalls = new Map(); private readonly terminalEntries = new Map(); private readonly persistedHistory: AgentTimelineItem[] = []; @@ -1136,6 +1143,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { const turnId = randomUUID(); const messageId = options?.messageId ?? randomUUID(); this.activeForegroundTurnId = turnId; + this.activeSubmittedUserMessage = null; this.emitBootstrapThreadEvent(); this.pushEvent({ type: "turn_started", provider: this.provider, turnId }); this.emitSubmittedUserMessage(prompt, messageId, turnId); @@ -2016,7 +2024,10 @@ export class ACPAgentSession implements AgentSession, ACPClient { if (!item) { return []; } - if (update.messageId && this.submittedUserMessageIds.has(update.messageId)) { + if (item.type !== "user_message") { + return [this.wrapTimeline(item)]; + } + if (this.isSubmittedUserMessageEcho(item)) { return []; } return [this.wrapTimeline(item)]; @@ -2099,7 +2110,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { if (!chunkText) { return null; } - const key = `${type}:${update.messageId ?? "default"}`; + const key = this.messageAssemblyKey(type, update.messageId); const state = this.messageAssemblies.get(key) ?? { text: "" }; state.text += chunkText; this.messageAssemblies.set(key, state); @@ -2113,6 +2124,15 @@ export class ACPAgentSession implements AgentSession, ACPClient { return { type: "reasoning", text: chunkText }; } + private messageAssemblyKey( + type: "user_message" | "assistant_message" | "reasoning", + messageId: string | null | undefined, + ): string { + const fallbackId = + type === "user_message" ? (this.activeForegroundTurnId ?? "default") : "default"; + return `${type}:${messageId ?? fallbackId}`; + } + private handleCurrentModeUpdate(update: CurrentModeUpdate): void { this.currentMode = this.transformModeId(update.currentModeId); } @@ -2231,6 +2251,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { return; } this.submittedUserMessageIds.add(messageId); + this.activeSubmittedUserMessage = { messageId, text, turnId }; this.pushEvent({ type: "timeline", provider: this.provider, @@ -2257,9 +2278,27 @@ export class ACPAgentSession implements AgentSession, ACPClient { event: Extract, ): void { this.activeForegroundTurnId = null; + if (this.activeSubmittedUserMessage?.turnId === event.turnId) { + this.activeSubmittedUserMessage = null; + } this.pushEvent(event); } + private isSubmittedUserMessageEcho( + item: Extract, + ): boolean { + const active = this.activeSubmittedUserMessage; + if (!active || active.turnId !== this.activeForegroundTurnId) { + return false; + } + if (item.messageId) { + if (this.submittedUserMessageIds.has(item.messageId)) { + return true; + } + } + return active.text.startsWith(item.text); + } + private emitBootstrapThreadEvent(): void { if (!this.bootstrapThreadEventPending || !this.sessionId) { return;