diff --git a/packages/app/e2e/browser/rewind-menu.ui-contract.spec.ts b/packages/app/e2e/browser/rewind-menu.ui-contract.spec.ts index 6ae16b62a7..e2c04aa315 100644 --- a/packages/app/e2e/browser/rewind-menu.ui-contract.spec.ts +++ b/packages/app/e2e/browser/rewind-menu.ui-contract.spec.ts @@ -26,6 +26,13 @@ async function expectUserMessageVisible(page: Page, text: string): Promise await expect(userMessage(page, text)).toBeVisible(); } +async function selectRewindOption(page: Page, name: string): Promise { + const option = page.getByRole("menuitem", { name, exact: true }); + // Menus mount offscreen while their anchor and content are measured. + await expect(option).toBeInViewport({ ratio: 1 }); + await option.click(); +} + async function rewriteCachedMessageAsLegacyRow( page: Page, input: { prompt: string; agentId: string; workspaceId: string }, @@ -190,6 +197,7 @@ test.describe("Rewind sheet", () => { await expectUserMessageVisible(page, secondPrompt); await expect(page.getByText("Cycle 1", { exact: true })).toBeVisible(); await expectUserMessageCount(page, 2); + await session.client.waitForFinish(session.agentId); await scrollChatAwayFromBottom(page, { deltaY: -900, @@ -202,7 +210,7 @@ test.describe("Rewind sheet", () => { await expect( rewindSheet.getByText("This action cannot be undone", { exact: true }), ).toBeVisible(); - await page.getByTestId("rewind-menu-conversation").click(); + await selectRewindOption(page, "Rewind conversation"); await expect(page.getByTestId("rewind-menu-content")).toHaveCount(0); await expect(userMessage(page, secondPrompt)).toHaveCount(0); @@ -221,7 +229,7 @@ test.describe("Rewind sheet", () => { await userMessage(page, replacementPrompt).hover(); await page.getByTestId("rewind-menu-trigger").last().click(); await expect(page.getByTestId("rewind-menu-content")).toBeVisible(); - await page.getByTestId("rewind-menu-files").click(); + await selectRewindOption(page, "Rewind files"); await expect(page.getByTestId("rewind-menu-content")).toHaveCount(0); await expectComposerDraft(page, ""); await expectUserMessageCount(page, 2); @@ -232,7 +240,7 @@ test.describe("Rewind sheet", () => { await userMessage(page, replacementPrompt).hover(); await page.getByTestId("rewind-menu-trigger").last().click(); await expect(page.getByTestId("rewind-menu-content")).toBeVisible(); - await page.getByTestId("rewind-menu-files").click(); + await selectRewindOption(page, "Rewind files"); await expect(page.getByTestId("rewind-menu-content")).toHaveCount(0); await expectComposerDraft(page, preservedDraft); await expectUserMessageCount(page, 2); @@ -242,7 +250,7 @@ test.describe("Rewind sheet", () => { await userMessage(page, replacementPrompt).hover(); await page.getByTestId("rewind-menu-trigger").last().click(); await expect(page.getByTestId("rewind-menu-content")).toBeVisible(); - await page.getByTestId("rewind-menu-both").click(); + await selectRewindOption(page, "Rewind conversation and files"); await expect(page.getByTestId("rewind-menu-content")).toHaveCount(0); await expectComposerDraft(page, replacementPrompt); await expectUserMessageCount(page, 1); diff --git a/packages/app/e2e/browser/streaming-markdown.spec.ts b/packages/app/e2e/browser/streaming-markdown.spec.ts index b1c76f3d89..3a48e8b10a 100644 --- a/packages/app/e2e/browser/streaming-markdown.spec.ts +++ b/packages/app/e2e/browser/streaming-markdown.spec.ts @@ -1,4 +1,10 @@ import { test } from "../support/fixtures"; +import { + withAcpStreamingMarkdown, + requestAcpMarkdown, + expectAcpMarkdown, + reloadAcpMarkdown, +} from "../support/helpers/acp-streaming-markdown"; import { expectFinishedMarkdown, expectReloadedMarkdown, @@ -19,3 +25,18 @@ test("formats unfinished Markdown while streaming and preserves the completed re await expectReloadedMarkdown(page); }); }); + +for (const width of [1100, 390]) { + test(`ACP plugin chunks preserve Markdown and separate turns at ${width}px`, async ({ + page, + }, testInfo) => { + await page.setViewportSize({ width, height: 844 }); + await withAcpStreamingMarkdown(page, async (agent) => { + await requestAcpMarkdown(agent); + await expectAcpMarkdown(page, 1); + await requestAcpMarkdown(agent); + await expectAcpMarkdown(page, 2); + await reloadAcpMarkdown(page, testInfo); + }); + }); +} diff --git a/packages/app/e2e/support/fixtures/acp-chunks-plugin/index.server.ts b/packages/app/e2e/support/fixtures/acp-chunks-plugin/index.server.ts new file mode 100644 index 0000000000..c98a04f2c1 --- /dev/null +++ b/packages/app/e2e/support/fixtures/acp-chunks-plugin/index.server.ts @@ -0,0 +1,46 @@ +import type { PluginServerContext } from "@getpaseo/plugin/server"; +import { runAcpProvider, type AcpStreamMessage } from "@getpaseo/plugin/server/acp"; + +function connect() { + let controller: ReadableStreamDefaultController; + return { + readable: new ReadableStream({ + start(value) { + controller = value; + }, + }), + writable: new WritableStream({ + write(message) { + if (!("method" in message) || !("id" in message)) return; + let result: unknown = {}; + if (message.method === "initialize") result = { protocolVersion: 1, agentCapabilities: {} }; + if (message.method === "session/new") result = { sessionId: "chunks" }; + if (message.method === "session/prompt") { + for (const text of [ + "- **Current tem", + "perature**: 25°C. Read [Paseo ", + "docs](https://example.com/docs).", + ]) { + controller.enqueue({ + jsonrpc: "2.0", + method: "session/update", + params: { + sessionId: "chunks", + update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text } }, + }, + }); + } + result = { stopReason: "end_turn" }; + } + controller.enqueue({ jsonrpc: "2.0", id: message.id, result }); + }, + }), + }; +} + +export default function contribute(server: PluginServerContext) { + server.registerProvider( + runAcpProvider({ id: "acp-chunks", label: "ACP chunks", connector: connect }), + ); + return () => {}; +} diff --git a/packages/app/e2e/support/fixtures/acp-chunks-plugin/paseo-plugin.json b/packages/app/e2e/support/fixtures/acp-chunks-plugin/paseo-plugin.json new file mode 100644 index 0000000000..44e12e11c5 --- /dev/null +++ b/packages/app/e2e/support/fixtures/acp-chunks-plugin/paseo-plugin.json @@ -0,0 +1 @@ +{ "id": "acp-chunks-test", "requirements": { "paseo": ">=0.8.0" } } diff --git a/packages/app/e2e/support/helpers/acp-streaming-markdown.ts b/packages/app/e2e/support/helpers/acp-streaming-markdown.ts new file mode 100644 index 0000000000..0ebe9c2b3d --- /dev/null +++ b/packages/app/e2e/support/helpers/acp-streaming-markdown.ts @@ -0,0 +1,72 @@ +import path from "node:path"; +import { expect, type Page, type TestInfo } from "@playwright/test"; +import { seedWorkspace, type SeedDaemonClient } from "./seed-client"; +import { openAgentRoute } from "./mock-agent"; +import { connectNewWorkspaceDaemonClient } from "./new-workspace"; + +interface AcpMarkdownAgent { + client: SeedDaemonClient; + agentId: string; +} + +export async function withAcpStreamingMarkdown( + page: Page, + run: (agent: AcpMarkdownAgent) => Promise, +): Promise { + const workspace = await seedWorkspace({ repoPrefix: "acp-markdown-" }); + const { client } = workspace; + const pluginClient = await connectNewWorkspaceDaemonClient(); + const previous = await pluginClient.getDaemonConfig(); + try { + await pluginClient.patchDaemonConfig({ pluginsEnabled: true }); + await pluginClient.installDirectoryPlugin( + path.resolve(__dirname, "../fixtures/acp-chunks-plugin"), + ); + const agent = await client.createAgent({ + provider: "acp-chunks", + cwd: workspace.repoPath, + workspaceId: workspace.workspaceId, + title: "ACP Markdown", + }); + await openAgentRoute(page, { workspaceId: workspace.workspaceId, agentId: agent.id }); + await run({ client, agentId: agent.id }); + } finally { + try { + await pluginClient.removePlugin("acp-chunks-test"); + await pluginClient.patchDaemonConfig({ pluginsEnabled: previous.config.pluginsEnabled }); + } finally { + await pluginClient.close(); + await workspace.cleanup(); + } + } +} + +export async function requestAcpMarkdown(agent: AcpMarkdownAgent): Promise { + await agent.client.sendAgentMessage(agent.agentId, "Show the temperature and docs."); + await agent.client.waitForFinish(agent.agentId, 30_000); +} + +export async function expectAcpMarkdown(page: Page, count: number): Promise { + // The message container has no semantic role; text and links inside it are user-facing assertions. + const messages = page.getByTestId("assistant-message"); + await expect(messages).toHaveCount(count); + for (let index = 0; index < count; index++) { + const message = messages.nth(index); + await expect(message).toHaveText("•Current temperature: 25°C. Read Paseo docs."); + await expect( + message.getByRole("link", { name: "Paseo docs" }).and(message.locator("a")), + ).toHaveAttribute("href", "https://example.com/docs"); + await expect(message.locator('[data-paseo-markdown-tag="strong"]')).toHaveText( + "Current temperature", + ); + } +} + +export async function reloadAcpMarkdown(page: Page, testInfo: TestInfo): Promise { + await page.reload({ waitUntil: "domcontentloaded" }); + await expectAcpMarkdown(page, 2); + await testInfo.attach("acp-markdown-after-reload", { + body: await page.screenshot({ path: testInfo.outputPath("acp-markdown-after-reload.png") }), + contentType: "image/png", + }); +} diff --git a/packages/plugin/src/server/acp-internal/connection.ts b/packages/plugin/src/server/acp-internal/connection.ts index be8fae25b9..5589cab825 100644 --- a/packages/plugin/src/server/acp-internal/connection.ts +++ b/packages/plugin/src/server/acp-internal/connection.ts @@ -360,6 +360,16 @@ class AcpRuntime { private modes: NewSessionResponse["modes"] = null; private commandWaiter: (() => void) | null = null; private messageSequence = 0; + /** + * 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). + */ + private readonly fallbackChunkIds = new Map< + "agent_message_chunk" | "agent_thought_chunk", + string + >(); private closing = false; private processFailed = false; private configTransaction = false; @@ -544,6 +554,7 @@ class AcpRuntime { } const prompt = toAcpPrompt(input.prompt); const turnId = `acp:${input.prompt.clientMessageId}`; + this.fallbackChunkIds.clear(); this.emit({ type: "timeline.item", sessionId: this.options.boundarySessionId, @@ -574,6 +585,7 @@ class AcpRuntime { ).then( (response): void => { const state = response.stopReason === "cancelled" ? "canceled" : "completed"; + this.fallbackChunkIds.clear(); this.terminalizeTransientItems(state); this.emit({ type: "session.turn", @@ -584,6 +596,7 @@ class AcpRuntime { return undefined; }, (error): void => { + this.fallbackChunkIds.clear(); this.terminalizeTransientItems("failed"); this.emit({ type: "session.turn", @@ -868,10 +881,19 @@ class AcpRuntime { update.sessionUpdate === "agent_thought_chunk" || update.sessionUpdate === "user_message_chunk" ) { - if (update.sessionUpdate === "user_message_chunk") return; + if (update.sessionUpdate === "user_message_chunk") { + 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. + this.fallbackChunkIds.delete( + update.sessionUpdate === "agent_message_chunk" + ? "agent_thought_chunk" + : "agent_message_chunk", + ); if (update.content.type !== "text") return; - const fallbackId = `${update.sessionUpdate}:${++this.messageSequence}`; - const id = update.messageId ?? fallbackId; + const id = this.resolveChunkId(update.sessionUpdate, update.messageId); const text = `${this.messages.get(id) ?? ""}${update.content.text}`; this.messages.set(id, text); this.emit({ @@ -889,12 +911,30 @@ class AcpRuntime { return; } if (update.sessionUpdate === "tool_call" || update.sessionUpdate === "tool_call_update") { + if (update.sessionUpdate === "tool_call") this.fallbackChunkIds.clear(); this.reduceToolCall(update); return; } + if (update.sessionUpdate === "plan") this.fallbackChunkIds.clear(); this.reduceStateUpdate(update); } + private resolveChunkId( + kind: "agent_message_chunk" | "agent_thought_chunk", + messageId: string | null | undefined, + ): string { + if (messageId) { + this.fallbackChunkIds.delete(kind); + return messageId; + } + let id = this.fallbackChunkIds.get(kind); + if (!id) { + id = `${kind}:${++this.messageSequence}`; + this.fallbackChunkIds.set(kind, id); + } + return id; + } + private reduceStateUpdate(update: SessionUpdate): void { if (update.sessionUpdate === "plan") { this.emit({ diff --git a/packages/plugin/src/server/acp.test.ts b/packages/plugin/src/server/acp.test.ts index 75ea3ae2ea..be52710dba 100644 --- a/packages/plugin/src/server/acp.test.ts +++ b/packages/plugin/src/server/acp.test.ts @@ -1,7 +1,12 @@ import { mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; -import { AgentSideConnection, PROTOCOL_VERSION, type Agent } from "@agentclientprotocol/sdk"; +import { + AgentSideConnection, + PROTOCOL_VERSION, + type Agent, + type SessionUpdate, +} from "@agentclientprotocol/sdk"; import type { ProviderConnection, ProviderEvent, ProviderInput } from "./provider.js"; import { afterEach, describe, expect, it } from "vitest"; import { runAcpProvider, type AcpStreamMessage } from "./acp.js"; @@ -68,6 +73,7 @@ interface ConnectorInstance { closeFromAgent(): void; request(method: string, params: unknown): Promise; respond(request: AcpRequestMessage, result: unknown): void; + reject(request: AcpRequestMessage, message: string): void; notify(method: string, params: unknown): void; } @@ -79,6 +85,8 @@ interface ConnectorResponse { interface ConnectorHarnessOptions { capabilities?: Record; configOptions?: unknown[]; + /** Session updates the agent streams for a prompt (keyed by its text) before ending the turn. */ + promptUpdates?(promptText: string): unknown[]; handleMessage?(instance: ConnectorInstance, message: AcpStreamMessage): boolean; handleRequest?(instance: ConnectorInstance, request: AcpRequestMessage): boolean; } @@ -120,6 +128,11 @@ function respondToConnectorRequest( configOptions: options?.configOptions ?? [], }); } else if (request.method === "session/prompt") { + const { prompt } = request.params as { prompt: Array<{ type: string; text?: string }> }; + const promptText = prompt.find((part) => part.type === "text")?.text ?? ""; + for (const update of options?.promptUpdates?.(promptText) ?? []) { + instance.notify("session/update", { sessionId: "connector-session", update }); + } instance.respond(request, { stopReason: "end_turn" }); } else if (request.method === "session/close") { instance.respond(request, {}); @@ -156,6 +169,8 @@ function connectorHarness(options?: ConnectorHarnessOptions) { return response; }, respond: (request, result) => controller.enqueue({ jsonrpc: "2.0", id: request.id, result }), + reject: (request, message) => + controller.enqueue({ jsonrpc: "2.0", id: request.id, error: { code: -32603, message } }), notify: (method, params) => controller.enqueue({ jsonrpc: "2.0", method, params }), }; instances.push(instance); @@ -1003,3 +1018,265 @@ lines.on("line", (line) => { expect(Date.now() - startedAt).toBeLessThan(2_500); }); }); + +describe("runAcpProvider streamed chunks without messageId (#4699)", () => { + interface Chunk { + kind: "agent_message_chunk" | "agent_thought_chunk"; + text: string; + messageId?: string; + } + + it("continues one timeline item per turn instead of one per chunk", async () => { + const chunksByPrompt: Record = { + first: [ + { kind: "agent_message_chunk", text: "- **Current tem" }, + { kind: "agent_message_chunk", text: "perature**: 25°C" }, + ], + second: [{ kind: "agent_message_chunk", text: "next turn" }], + // An explicit-id text chunk between two id-less thoughts separates them. + third: [ + { kind: "agent_thought_chunk", text: "think A" }, + { kind: "agent_message_chunk", text: "say", messageId: "m-explicit" }, + { kind: "agent_thought_chunk", text: "think B" }, + ], + }; + const harness = connectorHarness({ + promptUpdates: (promptText) => + (chunksByPrompt[promptText] ?? []).map((chunk) => { + const update: Record = { + sessionUpdate: chunk.kind, + content: { type: "text", text: chunk.text }, + }; + if (chunk.messageId) update.messageId = chunk.messageId; + return update; + }), + }); + const registration = runAcpProvider({ + id: "chunk-acp", + label: "Chunk ACP", + connector: harness.connector, + }); + const connection = await registration.connect({ + versions: [1], + capabilities: ["prompt.message"], + }); + const events: ProviderEvent[] = []; + connection.onEvent((event) => events.push(event)); + await connection.send(openInput()); + await waitForEvent(events, (event) => event.type === "session.ready"); + + for (const clientMessageId of ["first", "second", "third"]) { + await connection.send({ + type: "session.prompt", + sessionId: "session-1", + prompt: { + clientMessageId, + delivery: "auto", + input: { type: "message", content: [{ type: "text", text: clientMessageId }] }, + }, + }); + await waitForEvent( + events, + (event) => + event.type === "session.turn" && + event.turnId === `acp:${clientMessageId}` && + event.state === "completed", + ); + } + + const items = events.flatMap((event) => + event.type === "timeline.item" && + (event.item.type === "assistant_message" || event.item.type === "reasoning") + ? [{ id: event.item.id, text: event.item.text }] + : [], + ); + expect(items).toEqual([ + { id: "agent_message_chunk:1", text: "- **Current tem" }, + { id: "agent_message_chunk:1", text: "- **Current temperature**: 25°C" }, + { id: "agent_message_chunk:2", text: "next turn" }, + { id: "agent_thought_chunk:3", text: "think A" }, + { id: "m-explicit", text: "say" }, + { id: "agent_thought_chunk:4", text: "think B" }, + ]); + await connection.close(); + }); +}); + +function textChunk( + text: string, + kind: "agent_message_chunk" | "agent_thought_chunk" = "agent_message_chunk", + messageId?: string, +): SessionUpdate { + return { + sessionUpdate: kind, + content: { type: "text", text }, + ...(messageId === undefined ? {} : { messageId }), + }; +} + +async function streamChunkTurns( + updates: SessionUpdate[], + ending: "completed" | "canceled" | "failed" = "completed", +) { + let promptCount = 0; + const harness = connectorHarness({ + handleRequest(instance, request) { + if (request.method !== "session/prompt") return false; + promptCount++; + for (const update of promptCount === 1 ? updates : [textChunk("Next turn")]) { + instance.notify("session/update", { sessionId: "connector-session", update }); + } + if (promptCount === 1 && ending === "failed") instance.reject(request, "Prompt failed"); + else + instance.respond(request, { + stopReason: promptCount === 1 && ending === "canceled" ? "cancelled" : "end_turn", + }); + return true; + }, + }); + const registration = runAcpProvider({ + id: "chunk-boundaries", + label: "Chunk boundaries", + connector: harness.connector, + }); + const connection = await registration.connect({ + versions: [1], + capabilities: ["prompt.message"], + }); + const events: ProviderEvent[] = []; + connection.onEvent((event) => events.push(event)); + try { + await connection.send(openInput()); + await waitForEvent(events, (event) => event.type === "session.ready"); + for (const clientMessageId of ["first", "second"]) { + await connection.send({ + type: "session.prompt", + sessionId: "session-1", + prompt: { + clientMessageId, + delivery: "auto", + input: { type: "message", content: [{ type: "text", text: clientMessageId }] }, + }, + }); + await waitForEvent( + events, + (event) => + event.type === "session.turn" && + event.turnId === `acp:${clientMessageId}` && + event.state === (clientMessageId === "first" ? ending : "completed"), + ); + } + const messages = new Map(); + for (const event of events) { + if ( + event.type === "timeline.item" && + (event.item.type === "assistant_message" || event.item.type === "reasoning") + ) { + messages.set(event.item.id, { type: event.item.type, text: event.item.text }); + } + } + return [...messages.values()]; + } finally { + await connection.close(); + } +} + +describe("ACP streamed message boundaries", () => { + it.each([ + { + name: "tool call", + update: { + sessionUpdate: "tool_call", + toolCallId: "tool", + title: "Read", + kind: "read", + status: "in_progress", + }, + }, + { name: "plan", update: { sessionUpdate: "plan", entries: [] } }, + { + name: "user message", + update: { sessionUpdate: "user_message_chunk", content: { type: "text", text: "User" } }, + }, + ] satisfies Array<{ name: string; update: SessionUpdate }>)( + "starts a new message after a $name", + async ({ update }) => { + expect( + await streamChunkTurns([ + textChunk("Before"), + update, + textChunk("After "), + textChunk("boundary"), + ]), + ).toEqual([ + { type: "assistant_message", text: "Before" }, + { type: "assistant_message", text: "After boundary" }, + { type: "assistant_message", text: "Next turn" }, + ]); + }, + ); + + it("continues text while an existing tool receives a status update", async () => { + expect( + await streamChunkTurns([ + { + sessionUpdate: "tool_call", + toolCallId: "tool", + title: "Read", + kind: "read", + status: "in_progress", + }, + textChunk("Still "), + { sessionUpdate: "tool_call_update", toolCallId: "tool", status: "completed" }, + textChunk("speaking"), + ]), + ).toEqual([ + { type: "assistant_message", text: "Still speaking" }, + { type: "assistant_message", text: "Next turn" }, + ]); + }); + + it("keeps explicit messages separate from neighboring anonymous chunks", async () => { + expect( + await streamChunkTurns([ + textChunk("Before"), + textChunk("Explicit ", "agent_message_chunk", "message"), + textChunk("message", "agent_message_chunk", "message"), + textChunk("After"), + ]), + ).toEqual([ + { type: "assistant_message", text: "Before" }, + { type: "assistant_message", text: "Explicit message" }, + { type: "assistant_message", text: "After" }, + { type: "assistant_message", text: "Next turn" }, + ]); + }); + + it("continues reasoning but starts a new segment after nontext assistant content", async () => { + expect( + await streamChunkTurns([ + textChunk("Think ", "agent_thought_chunk"), + textChunk("first", "agent_thought_chunk"), + { + sessionUpdate: "agent_message_chunk", + content: { type: "image", data: "", mimeType: "image/png" }, + }, + textChunk("Then think", "agent_thought_chunk"), + ]), + ).toEqual([ + { type: "reasoning", text: "Think first" }, + { type: "reasoning", text: "Then think" }, + { type: "assistant_message", text: "Next turn" }, + ]); + }); + + it.each(["completed", "canceled", "failed"] as const)( + "keeps the next turn separate after a %s prompt", + async (ending) => { + expect(await streamChunkTurns([textChunk("First "), textChunk("turn")], ending)).toEqual([ + { type: "assistant_message", text: "First turn" }, + { type: "assistant_message", text: "Next turn" }, + ]); + }, + ); +});