From 01690cd965eb95f74b8a9974c7886fddafa56d77 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Thu, 22 Oct 2026 12:16:26 -0500 Subject: [PATCH 1/3] fix(pi): log oversized RPC records instead of dropping them silently The Pi stdout framer drops any record over MAX_PI_RECORD_CHARS with no trace. When the dropped record is a response, its request only fails later as a 30-second PiRpcTimeoutError, and nothing in the logs says why. The framer now reports the length of each dropped record, and the reader logs a warning with that length and the cap. Warning level shows at the server's default Info level. The log carries lengths only, never record content. Dropping behavior is unchanged. Co-Authored-By: Claude Opus 5.5 --- .../provider-pi/src/server/adapter.test.ts | 52 +++++++++++++++++++ packages/provider-pi/src/server/rpc.ts | 50 +++++++++++++----- 2 files changed, 90 insertions(+), 12 deletions(-) diff --git a/packages/provider-pi/src/server/adapter.test.ts b/packages/provider-pi/src/server/adapter.test.ts index 652444f175d6a..087e386d0abd6 100644 --- a/packages/provider-pi/src/server/adapter.test.ts +++ b/packages/provider-pi/src/server/adapter.test.ts @@ -23,6 +23,7 @@ import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import * as Logger from "effect/Logger"; import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Sink from "effect/Sink"; @@ -4001,6 +4002,57 @@ describe("PiRpc framing", () => { assert.equal((yield* Queue.take(connection.events))["type"], "after_oversized"); }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), ); + + it.effect("logs the length of an oversized record it drops", () => { + const warnings: Array> = []; + const logCapture = Logger.make(({ logLevel, message }) => { + if (logLevel === "Warn" && Array.isArray(message)) warnings.push(message); + }); + return Effect.gen(function* () { + const stdout = yield* Queue.unbounded(); + const spawner = ChildProcessSpawner.make(() => + Effect.succeed( + ChildProcessSpawner.makeHandle({ + pid: ChildProcessSpawner.ProcessId(FAKE_PID), + exitCode: Effect.never, + isRunning: Effect.succeed(true), + kill: () => Effect.void, + unref: Effect.succeed(Effect.void), + stdin: Sink.drain, + stdout: Stream.fromQueue(stdout), + stderr: Stream.empty, + all: Stream.empty, + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + }), + ), + ); + const connection = yield* makePiRpcConnection({ + command: "pi", + args: ["--mode", "rpc"], + cwd: undefined, + env: {}, + }).pipe(Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner)); + + const head = '{"type":"response","id":"t3-0","success":true,"data":"'; + const tail = '"}\n{"type":"after_oversized"}\n'; + const chunk = new TextEncoder().encode("x".repeat(8 * 1024 * 1024)); + yield* Queue.offer(stdout, new TextEncoder().encode(head)); + for (let index = 0; index < 32; index++) yield* Queue.offer(stdout, chunk); + yield* Queue.offer(stdout, new TextEncoder().encode(tail)); + + assert.equal((yield* Queue.take(connection.events))["type"], "after_oversized"); + assert.lengthOf(warnings, 1); + const [text, annotations] = warnings[0] ?? []; + assert.equal(text, "Dropping oversized pi stdout record."); + assert.propertyVal(annotations, "recordLength", head.length + 32 * chunk.length + 2); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.merge(NodeServices.layer, Logger.layer([logCapture], { mergeWithExisting: false })), + ), + ); + }); }); describe("PiRpc early process exit", () => { diff --git a/packages/provider-pi/src/server/rpc.ts b/packages/provider-pi/src/server/rpc.ts index c36a231051ed6..46510308d883c 100644 --- a/packages/provider-pi/src/server/rpc.ts +++ b/packages/provider-pi/src/server/rpc.ts @@ -99,35 +99,58 @@ interface PendingPiRequest { const MAX_PI_RECORD_CHARS = 2 * MAX_TOOL_OUTPUT_IMAGES * MAX_TOOL_OUTPUT_IMAGE_BASE64_LENGTH + 8 * 1024 * 1024; +interface PiJsonlFrames { + readonly lines: ReadonlyArray; + readonly droppedLengths: ReadonlyArray; +} + function makeJsonlFramer() { let buffer = ""; let dropping = false; - return (chunk: string): ReadonlyArray => { + let droppedLength = 0; + return (chunk: string): PiJsonlFrames => { const lines: string[] = []; + const droppedLengths: number[] = []; let start = 0; while (start < chunk.length) { const newline = chunk.indexOf("\n", start); const end = newline < 0 ? chunk.length : newline; - if (!dropping) { - if (buffer.length + end - start > MAX_PI_RECORD_CHARS) { - buffer = ""; - dropping = true; - } else { - buffer += chunk.slice(start, end); - } + if (dropping) { + droppedLength += end - start; + } else if (buffer.length + end - start > MAX_PI_RECORD_CHARS) { + droppedLength = buffer.length + end - start; + buffer = ""; + dropping = true; + } else { + buffer += chunk.slice(start, end); } if (newline < 0) break; - if (!dropping && buffer.length > 0) { + if (dropping) { + droppedLengths.push(droppedLength); + } else if (buffer.length > 0) { lines.push(buffer.endsWith("\r") ? buffer.slice(0, -1) : buffer); } buffer = ""; dropping = false; start = newline + 1; } - return lines; + return { lines, droppedLengths }; }; } +// A dropped `response` leaves its request to time out, so the drop must be +// visible at the default log level. Lengths only: records carry prompt text. +const logDroppedPiRecords = (droppedLengths: ReadonlyArray) => + Effect.forEach( + droppedLengths, + (recordLength) => + Effect.logWarning("Dropping oversized pi stdout record.", { + recordLength, + maxRecordLength: MAX_PI_RECORD_CHARS, + }), + { discard: true }, + ); + const UnknownFromJsonString = Schema.fromJsonString(Schema.Unknown); const decodeJsonLine = Schema.decodeSync(UnknownFromJsonString); const encodeJsonLine = Schema.encodeSync(UnknownFromJsonString); @@ -310,7 +333,8 @@ export const makePiRpcConnection = Effect.fnUntraced(function* (options: PiRpcSp Stream.decodeText(), Stream.runForEach((chunk) => Effect.gen(function* () { - const lines = frame(chunk); + const { lines, droppedLengths } = frame(chunk); + yield* logDroppedPiRecords(droppedLengths); for (const line of lines) { const record = parsePiRecord(line); if (record === undefined) { @@ -324,7 +348,9 @@ export const makePiRpcConnection = Effect.fnUntraced(function* (options: PiRpcSp }), ), ); - for (const line of frame("\n")) { + const flushed = frame("\n"); + yield* logDroppedPiRecords(flushed.droppedLengths); + for (const line of flushed.lines) { const trailing = parsePiRecord(line); if (trailing !== undefined) yield* routeRecord(trailing); } From 99c3ecb00f7e98bef91afd5ca8831132aa6be738 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sat, 31 Oct 2026 04:43:12 +0000 Subject: [PATCH 2/3] refactor(pi): reuse oversized record framing coverage --- .../provider-pi/src/server/adapter.test.ts | 53 +++---------------- packages/provider-pi/src/server/rpc.ts | 7 +-- 2 files changed, 9 insertions(+), 51 deletions(-) diff --git a/packages/provider-pi/src/server/adapter.test.ts b/packages/provider-pi/src/server/adapter.test.ts index f34dc5748c8bb..05fd10df6284c 100644 --- a/packages/provider-pi/src/server/adapter.test.ts +++ b/packages/provider-pi/src/server/adapter.test.ts @@ -4264,8 +4264,12 @@ describe("PiAdapterV2", () => { }); describe("PiRpc framing", () => { - it.effect("reassembles records across chunk boundaries and strips CR", () => - Effect.gen(function* () { + it.effect("reassembles records and warns when dropping oversized records", () => { + const warnings: Array> = []; + const logCapture = Logger.make(({ logLevel, message }) => { + if (logLevel === "Warn" && Array.isArray(message)) warnings.push(message); + }); + return Effect.gen(function* () { const stdout = yield* Queue.unbounded(); const spawner = ChildProcessSpawner.make(() => Effect.succeed( @@ -4296,7 +4300,8 @@ describe("PiRpc framing", () => { yield* push('{"type":"agent_'); yield* push('start"}\r\n{"type":"agent_settled"}\nnot json\n{"type":"queue_update"}\n'); - yield* push('{"type":"must_not_emit","text":"'); + const head = '{"type":"must_not_emit","text":"'; + yield* push(head); const chunk = new TextEncoder().encode("x".repeat(8 * 1024 * 1024)); for (let index = 0; index < 32; index++) yield* Queue.offer(stdout, chunk); yield* push('"}\n{"type":"after_oversized"}\n'); @@ -4308,48 +4313,6 @@ describe("PiRpc framing", () => { const third = yield* Queue.take(connection.events); assert.equal(third["type"], "queue_update"); assert.equal((yield* Queue.take(connection.events))["type"], "after_oversized"); - }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), - ); - - it.effect("logs the length of an oversized record it drops", () => { - const warnings: Array> = []; - const logCapture = Logger.make(({ logLevel, message }) => { - if (logLevel === "Warn" && Array.isArray(message)) warnings.push(message); - }); - return Effect.gen(function* () { - const stdout = yield* Queue.unbounded(); - const spawner = ChildProcessSpawner.make(() => - Effect.succeed( - ChildProcessSpawner.makeHandle({ - pid: ChildProcessSpawner.ProcessId(FAKE_PID), - exitCode: Effect.never, - isRunning: Effect.succeed(true), - kill: () => Effect.void, - unref: Effect.succeed(Effect.void), - stdin: Sink.drain, - stdout: Stream.fromQueue(stdout), - stderr: Stream.empty, - all: Stream.empty, - getInputFd: () => Sink.drain, - getOutputFd: () => Stream.empty, - }), - ), - ); - const connection = yield* makePiRpcConnection({ - command: "pi", - args: ["--mode", "rpc"], - cwd: undefined, - env: {}, - }).pipe(Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner)); - - const head = '{"type":"response","id":"t3-0","success":true,"data":"'; - const tail = '"}\n{"type":"after_oversized"}\n'; - const chunk = new TextEncoder().encode("x".repeat(8 * 1024 * 1024)); - yield* Queue.offer(stdout, new TextEncoder().encode(head)); - for (let index = 0; index < 32; index++) yield* Queue.offer(stdout, chunk); - yield* Queue.offer(stdout, new TextEncoder().encode(tail)); - - assert.equal((yield* Queue.take(connection.events))["type"], "after_oversized"); assert.lengthOf(warnings, 1); const [text, annotations] = warnings[0] ?? []; assert.equal(text, "Dropping oversized pi stdout record."); diff --git a/packages/provider-pi/src/server/rpc.ts b/packages/provider-pi/src/server/rpc.ts index 5fb96cbc6415b..c3ee54100a85f 100644 --- a/packages/provider-pi/src/server/rpc.ts +++ b/packages/provider-pi/src/server/rpc.ts @@ -93,16 +93,11 @@ interface PendingPiRequest { const MAX_PI_RECORD_CHARS = 2 * MAX_TOOL_OUTPUT_IMAGES * MAX_TOOL_OUTPUT_IMAGE_BASE64_LENGTH + 8 * 1024 * 1024; -interface PiJsonlFrames { - readonly lines: ReadonlyArray; - readonly droppedLengths: ReadonlyArray; -} - function makeJsonlFramer() { let buffer = ""; let dropping = false; let droppedLength = 0; - return (chunk: string): PiJsonlFrames => { + return (chunk: string) => { const lines: string[] = []; const droppedLengths: number[] = []; let start = 0; From a279c01a9cdbe237f9e892af083aa77575c27ae1 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sat, 31 Oct 2026 05:00:59 +0000 Subject: [PATCH 3/3] fix(pi): warn as soon as an oversized record crosses the cap --- .../provider-pi/src/server/adapter.test.ts | 12 ++++++++-- packages/provider-pi/src/server/rpc.ts | 24 ++++++++----------- 2 files changed, 20 insertions(+), 16 deletions(-) diff --git a/packages/provider-pi/src/server/adapter.test.ts b/packages/provider-pi/src/server/adapter.test.ts index 05fd10df6284c..e14b5db635067 100644 --- a/packages/provider-pi/src/server/adapter.test.ts +++ b/packages/provider-pi/src/server/adapter.test.ts @@ -4266,8 +4266,12 @@ describe("PiAdapterV2", () => { describe("PiRpc framing", () => { it.effect("reassembles records and warns when dropping oversized records", () => { const warnings: Array> = []; + const warned = Deferred.makeUnsafe(); const logCapture = Logger.make(({ logLevel, message }) => { - if (logLevel === "Warn" && Array.isArray(message)) warnings.push(message); + if (logLevel === "Warn" && Array.isArray(message)) { + warnings.push(message); + Deferred.doneUnsafe(warned, Effect.void); + } }); return Effect.gen(function* () { const stdout = yield* Queue.unbounded(); @@ -4304,6 +4308,7 @@ describe("PiRpc framing", () => { yield* push(head); const chunk = new TextEncoder().encode("x".repeat(8 * 1024 * 1024)); for (let index = 0; index < 32; index++) yield* Queue.offer(stdout, chunk); + yield* Deferred.await(warned); yield* push('"}\n{"type":"after_oversized"}\n'); const first = yield* Queue.take(connection.events); @@ -4316,7 +4321,10 @@ describe("PiRpc framing", () => { assert.lengthOf(warnings, 1); const [text, annotations] = warnings[0] ?? []; assert.equal(text, "Dropping oversized pi stdout record."); - assert.propertyVal(annotations, "recordLength", head.length + 32 * chunk.length + 2); + assert.deepEqual(annotations, { + recordLength: head.length + 28 * chunk.length, + maxRecordLength: 232084864, + }); }).pipe( Effect.scoped, Effect.provide( diff --git a/packages/provider-pi/src/server/rpc.ts b/packages/provider-pi/src/server/rpc.ts index c3ee54100a85f..0af3fade61c33 100644 --- a/packages/provider-pi/src/server/rpc.ts +++ b/packages/provider-pi/src/server/rpc.ts @@ -96,7 +96,6 @@ const MAX_PI_RECORD_CHARS = function makeJsonlFramer() { let buffer = ""; let dropping = false; - let droppedLength = 0; return (chunk: string) => { const lines: string[] = []; const droppedLengths: number[] = []; @@ -104,19 +103,17 @@ function makeJsonlFramer() { while (start < chunk.length) { const newline = chunk.indexOf("\n", start); const end = newline < 0 ? chunk.length : newline; - if (dropping) { - droppedLength += end - start; - } else if (buffer.length + end - start > MAX_PI_RECORD_CHARS) { - droppedLength = buffer.length + end - start; - buffer = ""; - dropping = true; - } else { - buffer += chunk.slice(start, end); + if (!dropping) { + if (buffer.length + end - start > MAX_PI_RECORD_CHARS) { + droppedLengths.push(buffer.length + end - start); + buffer = ""; + dropping = true; + } else { + buffer += chunk.slice(start, end); + } } if (newline < 0) break; - if (dropping) { - droppedLengths.push(droppedLength); - } else if (buffer.length > 0) { + if (!dropping && buffer.length > 0) { lines.push(buffer.endsWith("\r") ? buffer.slice(0, -1) : buffer); } buffer = ""; @@ -127,8 +124,7 @@ function makeJsonlFramer() { }; } -// A dropped `response` leaves its request to time out, so the drop must be -// visible at the default log level. Lengths only: records carry prompt text. +// Report the observed length before LF arrives; an unterminated response can time out first. const logDroppedPiRecords = (droppedLengths: ReadonlyArray) => Effect.forEach( droppedLengths,