From 33eb68b3ae4ce7ee27b31c59bd41db600ad47d19 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Fri, 21 Aug 2026 10:09:30 -0700 Subject: [PATCH] fix(server): end the duplex route on the pre-bind bounds when frames arrive before the bind (#11860) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The host and a plugin worker talk over a duplex channel route with bounds on buffered frames and total bytes > - A worker can batch its data and exit frames with the open reply, so those frames arrive before the route binds and before a listener attaches > - Two of the route bounds did not hold on that pre-bind path: a shared limit let the pre-open hold swallow an over-limit frame before the buffered-frame bound could end the route, and the route end discarded chunks a later listener still needed > - This pull request gives the pre-open hold its own ceiling above the buffered bound, and keeps the buffered chunks across a route end > - The benefit is a duplex route that enforces its bounds and preserves valid data, even when a worker batches frames ahead of the bind ## Linked Issues or Issue Description No existing GitHub issue covers this. Filing it directly here, following the bug report template. **What happened?** Two duplex channel route bounds in `server/src/services/plugin-worker-manager.ts` did not hold when the data and exit frames arrived in the open-reply read batch, before the route bound: - The pre-open hold and the pre-bind buffered-frame bound shared one limit. When a caller lowered the buffered bound, the hold dropped the overflow frame as a protocol error before the buffered bound could end the route, so the route never ended. - The route end discarded the buffered chunks. A frame can end the route during the replay, before a listener attaches, and the chunks the host accepted before that frame are valid data. **Expected behavior** The pre-open hold uses its own ceiling, above the buffered bound, so the replay after the bind lets the buffered bound end the route. A route end keeps the buffered chunks so a listener that attaches after the end still drains them. **Steps to reproduce** 1. Open a duplex channel where the worker batches several data frames with the open reply. 2. Lower `maxPreBindBufferedFrames` below the batch size. 3. Observe the route fails to end on the buffered-frame bound, or a listener that attaches after an end-during-replay never receives the chunks buffered before that end. **Paperclip version or commit** `933749e01f74e82ce5d315c071be534d04e01158` **Deployment mode** Local dev (`pnpm dev`) and server unit tests. **Agent adapter(s) involved** None — this is host/plugin-worker transport infrastructure, not adapter-specific. **Database mode** Not database-related. **Access context** Any board or agent path that runs a plugin worker over a duplex channel route. ## What Changed - Give the pre-open frame hold its own ceiling (`MAX_DUPLEX_CHANNEL_PRE_OPEN_HOLD_FRAMES`), separate from the pre-bind buffered-frame bound, so lowering the buffered bound still ends the route instead of being pre-empted by the hold. - Keep the buffered chunks on a route end instead of discarding them, so a listener that attaches after an end-during-replay still drains the data the host already accepted. - Add two regression tests that batch frames with the open reply, so both bounds run through the pre-bind path deterministically. ## Verification - `cd server && npx vitest run src/__tests__/plugin-worker-manager-duplex.test.ts` — 24/24 tests pass, including the two new regression cases. ## Risks Low risk. This only changes bound bookkeeping on an internal transport path (frame hold ceiling and end-time buffer retention); it does not change the wire protocol or any public API. The new ceiling is a constant above the existing buffered bound, so pre-open holds are still capped. ## Model Used Claude, Sonnet 5 (claude-sonnet-5); assisted with repository-grounded diff review and drafted this PR description from the commit and code history. No functional code in this PR was authored by Claude — the fix itself is Priya Raman's, preserved with original authorship intact. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- .../fixtures/plugin-worker-duplex-channel.cjs | 13 ++ .../plugin-worker-manager-duplex.test.ts | 101 +++++++++++++++ server/src/services/plugin-worker-manager.ts | 117 ++++++++++++++---- 3 files changed, 206 insertions(+), 25 deletions(-) diff --git a/server/src/__tests__/fixtures/plugin-worker-duplex-channel.cjs b/server/src/__tests__/fixtures/plugin-worker-duplex-channel.cjs index 7c116b1dbd..bd7b4ac8ad 100644 --- a/server/src/__tests__/fixtures/plugin-worker-duplex-channel.cjs +++ b/server/src/__tests__/fixtures/plugin-worker-duplex-channel.cjs @@ -13,6 +13,11 @@ // id; a test sets a wrong `sid` to prove the host drops a mismatched // notification and counts a protocol error. // - `exitCode`: when set, the fixture emits an exit notification after the data. +// - `dataAfterExit`: an array of `{ chunk, sid? }`, same shape as `data`. The +// fixture emits these as data notifications after the exit notification, so a +// test can script a batch where a data frame arrives after the exit in the +// read order — proving the host still bounds these frames on their own terms +// and never lets the exit crowd a data frame out of the pre-open hold. // - `echoInput`: when true, the fixture echoes each `duplexChannelWrite` back as // one data notification for the bound session. // - `closeMode`: "ack" | "bad-ack" | "no-ack" (default "ack"). It controls the @@ -47,6 +52,14 @@ function scriptedFrameLines(directive, workerSessionId) { params: { workerSessionId, exitCode: directive.exitCode }, })}\n`; } + const dataAfterExit = Array.isArray(directive.dataAfterExit) ? directive.dataAfterExit : []; + for (const entry of dataAfterExit) { + lines += `${JSON.stringify({ + jsonrpc: "2.0", + method: "duplexChannel.data", + params: { workerSessionId: entry.sid ?? workerSessionId, chunk: entry.chunk }, + })}\n`; + } return lines; } diff --git a/server/src/__tests__/plugin-worker-manager-duplex.test.ts b/server/src/__tests__/plugin-worker-manager-duplex.test.ts index b6a5c86a37..947315de5a 100644 --- a/server/src/__tests__/plugin-worker-manager-duplex.test.ts +++ b/server/src/__tests__/plugin-worker-manager-duplex.test.ts @@ -512,6 +512,107 @@ describe("plugin worker manager duplex channel route", () => { } }); + it("ends the route when batched pre-bind frames pass the frame count bound", async () => { + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxPreBindBufferedFrames: 2 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + // The worker batches the three data frames with the open reply, so all + // three frames arrive before the route binds. No listener attaches, so + // the replay buffers them. The third frame passes the frame-count bound + // and the route ends. The pre-open hold must not drop the third frame + // before the buffered bound can end the route. + batchWithOpenReply: true, + data: [{ chunk: "a" }, { chunk: "b" }, { chunk: "c" }], + }), + ); + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("ends the route on the frame count bound even when a caller raises that bound well past the module default", async () => { + // Regression test: the pre-open hold ceiling must track + // maxDuplexChannelPreBindFrames, not a fixed value. A fixed ceiling at or + // below this bound would drop the 13th frame in the hold before the replay + // ever applies the buffered bound to it, and the route would never end. + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxPreBindBufferedFrames: 12 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + batchWithOpenReply: true, + data: Array.from({ length: 13 }, (_, i) => ({ chunk: String(i) })), + }), + ); + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("ends the route on the frame count bound even when an exit notification arrives before the overflow data frame", async () => { + // Regression test: an exit notification must never share the pre-open hold's + // capacity with data frames. The worker here batches exactly + // maxPreBindBufferedFrames valid data frames (no violation), then its exit, + // then one more data frame that should trip the buffered-frame bound. If the + // exit consumed a hold slot, that last data frame would be dropped by the + // hold before the replay ever applies the buffered bound to it, and the + // route would end normally on the exit (exitCode: 0) instead of on the + // bound (exitCode: null). + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxPreBindBufferedFrames: 3 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + batchWithOpenReply: true, + data: [{ chunk: "a" }, { chunk: "b" }, { chunk: "c" }], + exitCode: 0, + dataAfterExit: [{ chunk: "overflow" }], + }), + ); + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("delivers a batched pre-bind chunk that a later listener drains before the byte cap ends the route", async () => { + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxTotalDataBytes: 4 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + // The worker batches the two data frames with the open reply, so both + // frames arrive before the route binds and before a listener attaches. + // "€" is three bytes in UTF-8. The first chunk (3 bytes ≤ 4) buffers. + // The second chunk brings the total to 6 bytes (> 4), so the route ends. + // The route end must not discard the buffered first chunk. The listener + // attaches after the open resolves and drains the first chunk. + batchWithOpenReply: true, + workerSessionId: "ws-A", + data: [{ chunk: "€" }, { chunk: "€" }], + }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + expect(chunks).toEqual(["€"]); + } finally { + await handle.stop().catch(() => undefined); + } + }); + // ------------------------------------------------------------------------- // Authoritative closure and worker retirement. // ------------------------------------------------------------------------- diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index dfbe2a444a..96f8c67a39 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -175,6 +175,23 @@ const MAX_DUPLEX_CHANNEL_PRE_BIND_CHARS = 8 * 1024 * 1024; * channel route before a data listener attaches. */ const MAX_DUPLEX_CHANNEL_PRE_BIND_FRAMES = 10_000; +/** + * The margin the pre-open hold ceiling keeps above the pre-bind buffered-frame + * bound (`maxDuplexChannelPreBindFrames`, see below). A worker can batch frames + * with the open reply, so the host reads them before it knows the bound worker + * session id and before it can apply the per-frame bounds. The host holds these + * frames and replays them after the bind. The hold ceiling bounds that hold, so + * a worker that floods frames before it replies to the open cannot make the + * host hold an unbounded number of frames. + * + * The hold ceiling is derived from the buffered bound, not a fixed constant: a + * fixed ceiling equal to or below a caller-configured (or even the default) + * buffered bound would let the hold drop the frame that should instead trip the + * buffered bound during replay, so the route would never end. One frame of + * margin is enough — it lets the frame that exceeds the buffered bound reach + * the hold, so the replay's buffered-bound check, not the hold, ends the route. + */ +const DUPLEX_CHANNEL_PRE_OPEN_HOLD_MARGIN_FRAMES = 1; /** * The default maximum number of in-flight host→worker requests for one duplex * channel route. A worker that never replies cannot make the host hold an @@ -811,6 +828,11 @@ export function createPluginWorkerHandle( const maxDuplexChannelPreBindFrames = options.duplexChannelLimits?.maxPreBindBufferedFrames ?? MAX_DUPLEX_CHANNEL_PRE_BIND_FRAMES; + // Always strictly above maxDuplexChannelPreBindFrames, including when a + // caller configures that bound at or above the module default. See + // DUPLEX_CHANNEL_PRE_OPEN_HOLD_MARGIN_FRAMES for why the margin must hold. + const maxDuplexChannelPreOpenHoldFrames = + maxDuplexChannelPreBindFrames + DUPLEX_CHANNEL_PRE_OPEN_HOLD_MARGIN_FRAMES; const maxDuplexChannelPendingRequests = options.duplexChannelLimits?.maxPendingRequests ?? MAX_DUPLEX_CHANNEL_PENDING_REQUESTS; @@ -1483,13 +1505,26 @@ export function createPluginWorkerHandle( listener: ((chunk: string) => void) | null; buffered: string[]; bufferedChars: number; - // Raw data and exit notifications that arrive before the route binds. The - // host reads the worker stdout line by line. The open reply and a data or - // exit notification can arrive in one read batch, so the host dispatches the - // notification before the deferred open-reply continuation flips the state to - // `open`. The host holds these frames here and replays them in order right - // after it binds the route, so a batched frame is never lost. + // Raw data notifications that arrive before the route binds. The host reads + // the worker stdout line by line. The open reply and a data notification can + // arrive in one read batch, so the host dispatches the notification before + // the deferred open-reply continuation flips the state to `open`. The host + // holds these frames here and replays them in order right after it binds the + // route, so a batched frame is never lost. Bounded by the pre-open hold + // ceiling — see `bufferPreOpenDuplexChannelNotification`. preOpen: JsonRpcNotification[]; + // The most recent exit notification that arrived before the route binds, or + // null. A single slot, not an array: only the last exit a worker sends before + // the bind is ever meaningful, so the host overwrites this on every pre-open + // exit instead of holding each one. That keeps a worker that batches many + // exit notifications before it replies to the open from growing this past one + // entry — the pre-open hold ceiling bounds only `preOpen`, so an exit that + // shared that array with data frames could otherwise consume a data frame's + // hold slot and let a data frame that should trip the buffered-frame bound + // get dropped by the hold instead. Replayed after `preOpen` drains, so data + // still delivers before the exit resolves the wait, matching a real worker's + // order. + preOpenExit: JsonRpcNotification | null; pendingRequests: number; protocolErrors: number; totalDataBytes: number; @@ -1536,9 +1571,14 @@ export function createPluginWorkerHandle( route.terminalized = true; route.state = "closed"; route.listener = null; - route.buffered = []; - route.bufferedChars = 0; + // Keep the buffered chunks that the host accepted before the route ended, so + // a listener that attaches after the end still drains them. A frame can end + // the route during the pre-open replay, before a listener attaches, and the + // buffered chunks the host accepted before that frame are valid data the + // listener must still receive. The buffered bytes stay bounded by the + // pre-bind buffered bound, and `onData` clears them once it drains them. route.preOpen = []; + route.preOpenExit = null; clearDuplexChannelLifetimeTimer(route); // A terminalized route reports a null exit code, which the caller treats as a // failure. @@ -1567,37 +1607,62 @@ export function createPluginWorkerHandle( } // Hold one data or exit notification that arrives before the route binds. The - // host replays the held frames in order after it binds the route. Bound the - // hold by the pre-bind frame count, so a worker that floods frames before it - // replies to the open cannot make the host hold an unbounded number of frames. - // Count one protocol error for each frame past the bound. + // host replays the held notifications after it binds the route: `preOpen` + // drains first, in order, then the held exit (if any) resolves the wait last — + // see `drainPreOpenDuplexChannelNotifications`. + // + // A data notification goes on `preOpen`, bounded by a pre-open ceiling derived + // from the pre-bind buffered-frame bound, so a worker that floods data frames + // before it replies to the open cannot make the host hold an unbounded number + // of them. Count one protocol error for each data frame past the ceiling. This + // ceiling is separate from the pre-bind buffered-frame bound, but it tracks it: + // the replay after the bind applies the buffered bound to each held frame, so + // the buffered bound ends the route when a caller lowers (or raises) it. The + // hold ceiling stays above the buffered bound by construction + // (maxDuplexChannelPreOpenHoldFrames = maxDuplexChannelPreBindFrames + margin), + // or it would drop a frame before the buffered bound can end the route. + // + // An exit notification never touches `preOpen`. It overwrites the single + // `preOpenExit` slot instead, so a worker that batches an exit among enough + // data frames to fill the hold cannot consume a data frame's hold slot: the + // ceiling above bounds `preOpen` alone, so it stays exactly the margin above + // the buffered bound regardless of how many exit notifications arrive pre-open. function bufferPreOpenDuplexChannelNotification( route: DuplexChannelRoute, notification: JsonRpcNotification, ): void { - if (route.preOpen.length >= maxDuplexChannelPreBindFrames) { + if (notification.method === DUPLEX_CHANNEL_EXIT_NOTIFICATION) { + route.preOpenExit = notification; + return; + } + if (route.preOpen.length >= maxDuplexChannelPreOpenHoldFrames) { recordDuplexChannelProtocolError(route); return; } route.preOpen.push(notification); } - // Replay the held pre-open frames in order right after the route binds. The - // route is `open` now, so each frame passes through the normal per-frame bounds - // and the session-identifier match. A frame that ends the route terminalizes - // it, and every later frame in the replay is a no-op, because the routing - // functions drop a frame when the route is not `open`. + // Replay the held pre-open notifications right after the route binds. The + // route is `open` now, so each notification passes through the normal + // per-frame bounds and the session-identifier match. Drain the held data + // frames first, in order, then replay the held exit (if any) last, so data + // still delivers before the exit resolves the wait, matching a real worker's + // order. A frame that ends the route terminalizes it, and every later + // notification in the replay is a no-op, because the routing functions drop a + // notification when the route is not `open`. function drainPreOpenDuplexChannelNotifications(route: DuplexChannelRoute): void { - if (route.preOpen.length === 0) return; - const pending = route.preOpen; - route.preOpen = []; - for (const notification of pending) { - if (notification.method === DUPLEX_CHANNEL_DATA_NOTIFICATION) { + if (route.preOpen.length > 0) { + const pending = route.preOpen; + route.preOpen = []; + for (const notification of pending) { routeDuplexChannelData(notification); - } else if (notification.method === DUPLEX_CHANNEL_EXIT_NOTIFICATION) { - routeDuplexChannelExit(notification); } } + if (route.preOpenExit) { + const exitNotification = route.preOpenExit; + route.preOpenExit = null; + routeDuplexChannelExit(exitNotification); + } } // Deliver one duplex channel chunk to the bound listener in isolation. A @@ -1713,6 +1778,7 @@ export function createPluginWorkerHandle( route.buffered = []; route.bufferedChars = 0; route.preOpen = []; + route.preOpenExit = null; clearDuplexChannelLifetimeTimer(route); settleRouteWait(route, { exitCode: null }); } @@ -1742,6 +1808,7 @@ export function createPluginWorkerHandle( buffered: [], bufferedChars: 0, preOpen: [], + preOpenExit: null, pendingRequests: 0, protocolErrors: 0, totalDataBytes: 0,