From 20d53e517bf8d726ad55181a7ee4b1294056a0de Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 29 Sep 2026 16:10:01 -0500 Subject: [PATCH] Preserve identified empty ACP assistant chunks Co-Authored-By: Paperclip --- ...ssage-boundaries-package-contract.test.mjs | 58 +++++++++++++++++++ patches/acpx@0.13.1.patch | 18 ++++++ 2 files changed, 76 insertions(+) create mode 100644 packages/paperclip-runner/test/acpx-message-boundaries-package-contract.test.mjs diff --git a/packages/paperclip-runner/test/acpx-message-boundaries-package-contract.test.mjs b/packages/paperclip-runner/test/acpx-message-boundaries-package-contract.test.mjs new file mode 100644 index 00000000000..af58aff37df --- /dev/null +++ b/packages/paperclip-runner/test/acpx-message-boundaries-package-contract.test.mjs @@ -0,0 +1,58 @@ +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { mkdtemp, rm } from "node:fs/promises"; +import { createRequire } from "node:module"; +import { tmpdir } from "node:os"; +import { dirname, join } from "node:path"; +import { pathToFileURL } from "node:url"; +import test from "node:test"; + +const require = createRequire(import.meta.url); +const root = process.env.PAPERCLIP_TEST_ACPX_ROOT ?? dirname(require.resolve("acpx/package.json")); +const { createAcpRuntime, createAgentRegistry, createFileSessionStore } = await import(pathToFileURL(join(root, "dist/runtime.js"))); +const fixture = String.raw` +const readline = require('node:readline'); +const send = value => process.stdout.write(JSON.stringify({jsonrpc:'2.0',...value})+'\n'); +readline.createInterface({input:process.stdin}).on('line',line=>{ + const request=JSON.parse(line); + if(request.method==='initialize') send({id:request.id,result:{protocolVersion:1,agentCapabilities:{sessionCapabilities:{close:{}}},authMethods:[]}}); + else if(request.method==='session/new') send({id:request.id,result:{sessionId:'fixture-session'}}); + else if(request.method==='session/prompt') { + for(const update of JSON.parse(request.params.prompt[0].text)) send({method:'session/update',params:{sessionId:'fixture-session',update}}); + send({id:request.id,result:{stopReason:'end_turn'}}); + } else if(request.id!==undefined) send({id:request.id,result:{}}); +});`; + +test("actual ACP SDK and runtime retain identified empty assistant boundaries in order", { timeout: 15_000 }, async () => { + const cwd = await mkdtemp(join(tmpdir(), "paperclip-acpx-message-boundary-")); + const children = []; + const runtime = createAcpRuntime({ cwd, sessionStore: createFileSessionStore({ stateDir: join(cwd, "sessions") }), + agentRegistry: createAgentRegistry({ overrides: { pi: ["verified-fixture"] } }), permissionMode: "approve-all", + spawnAgent: () => { const child = spawn(process.execPath, ["-e", fixture], { cwd, stdio: ["pipe", "pipe", "pipe"] }); children.push(child); return child; }, + }); + let handle; + try { + handle = await runtime.ensureSession({ sessionKey: "fixture-boundaries", agent: "pi", mode: "persistent", cwd }); + const chunk = (messageId, text, kind = "delta", thought = false) => ({ sessionUpdate: thought ? "agent_thought_chunk" : "agent_message_chunk", + ...(messageId ? { messageId } : {}), content: { type: "text", text }, + _meta: { origin: "pi-native-assistant", source: "pi-rpc-message-v1", kind, unrelated: "DROP_ME" }, + }); + const updates = [chunk("first", "", "start"), chunk("first", "pre-tool narration"), chunk("first", "", "end:toolUse"), + chunk("last", "", "start"), chunk("last", "thought", "delta", true), chunk("last", "EXACT_MARKER"), chunk("last", "", "end:stop"), + chunk(undefined, ""), { sessionUpdate: "agent_message_chunk", messageId: "no-text", content: { type: "image", data: "", mimeType: "image/png" } }]; + for (let index = 0; index < 2; index++) { + const turn = runtime.startTurn({ handle, text: JSON.stringify(updates), mode: "prompt", sessionMode: "persistent", requestId: `request-${index}`, timeoutMs: 5000 }); + const events = []; + for await (const event of turn.events) if (event.type === "text_delta") events.push(event); + assert.equal((await turn.result).status, "completed"); + assert.deepEqual(events.map(event => [event.messageId, event.text, event.meta?.kind, event.stream]), [ + ["first", "", "start", "output"], ["first", "pre-tool narration", "delta", "output"], ["first", "", "end:toolUse", "output"], + ["last", "", "start", "output"], ["last", "thought", "delta", "thought"], ["last", "EXACT_MARKER", "delta", "output"], ["last", "", "end:stop", "output"], + ]); + assert.equal(JSON.stringify(events).includes("DROP_ME"), false); + } + } finally { + try { if (handle) await runtime.close({ handle, discardPersistentState: true }); } + finally { for (const child of children) child.kill("SIGTERM"); await rm(cwd, { recursive: true, force: true }); } + } +}); diff --git a/patches/acpx@0.13.1.patch b/patches/acpx@0.13.1.patch index 1df67b522d5..42a0701b4a9 100644 --- a/patches/acpx@0.13.1.patch +++ b/patches/acpx@0.13.1.patch @@ -739,6 +739,24 @@ diff --git a/dist/runtime.js b/dist/runtime.js index a1f4a70a003792c6eacf68b6b038f37bfec1db53..896617544f6467c88da0dedf5c4867fc63e1fffe 100644 --- a/dist/runtime.js +++ b/dist/runtime.js +@@ -154,7 +154,7 @@ function resolveTextChunk(params) { + const contentType = asTrimmedString(contentRaw.type); + if (contentType && contentType !== "text") return null; + const text = asString(contentRaw.text); +- if (text && text.length > 0) return { ++ if (typeof text === "string" && (text.length > 0 || origin.messageId)) return { + type: "text_delta", + text, + stream: params.stream, +@@ -165,7 +165,7 @@ function resolveTextChunk(params) { + }; + } + const text = asString(params.payload.text); +- if (!text || text.length === 0) return null; ++ if (typeof text !== "string" || (text.length === 0 && !origin.messageId)) return null; + return { + type: "text_delta", + text, @@ -371,7 +371,7 @@ const PROMPT_EVENT_PARSERS = { current_mode_update: (payload) => statusUpdateEvent("current_mode_update", payload), config_option_update: (payload) => statusUpdateEvent("config_option_update", payload),