From 9013a63a76fa339b5e2898947e9cff040505d72c Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Sat, 29 Aug 2026 22:57:52 -0600 Subject: [PATCH] fix: bound public daemon client ingress --- .pylon/features.yaml | 17 ++ .pylon/upstream-review.md | 11 + .../.changes/bounded-daemon-ingress.md | 1 + packages/coding-agent/docs/daemon.md | 18 +- packages/coding-agent/docs/sdk.md | 25 ++ packages/coding-agent/src/index.ts | 4 + .../src/modes/daemon/daemon-client.ts | 155 +++++++++--- packages/coding-agent/src/modes/index.ts | 4 + packages/coding-agent/src/modes/rpc/jsonl.ts | 139 +++++++++++ packages/coding-agent/src/sdk-features.ts | 9 + .../test/daemon-bounded-ingress-bench.ts | 186 +++++++++++++++ .../coding-agent/test/daemon-client.test.ts | 220 +++++++++++++++++- packages/coding-agent/test/rpc-jsonl.test.ts | 142 ++++++++++- 13 files changed, 900 insertions(+), 31 deletions(-) create mode 100644 packages/coding-agent/.changes/bounded-daemon-ingress.md create mode 100644 packages/coding-agent/src/sdk-features.ts create mode 100644 packages/coding-agent/test/daemon-bounded-ingress-bench.ts diff --git a/.pylon/features.yaml b/.pylon/features.yaml index b89401bcd8..967ce79d28 100644 --- a/.pylon/features.yaml +++ b/.pylon/features.yaml @@ -226,3 +226,20 @@ decisions: revisit_when: - Prime upstream exposes an equivalent privacy-safe, capability-negotiated cleanup proof with durable descriptor-last retries. - Comet and Pylon can remove the fork capability without weakening crash cleanup or stock-Prime fallback behavior. + + bounded-public-daemon-ingress: + area: runtime-reliability + state: candidate + owner: shared + decision: hybridize + pylon_refs: + - https://github.com/pylon-code/prime-agent/issues/13 + - https://github.com/pylon-code/pylon/issues/114 + - https://github.com/rynfar/comet/issues/5 + upstream_refs: + - https://github.com/PrimeIntellect-ai/prime-agent/pull/480 + fork_change: bounded-daemon-ingress-v1 + upstream_support: Prime through a903d4b6768f has an optional decoded-character JSONL line limit with prefix callbacks and discard/resume behavior, but its public DaemonClient does not bound raw inbound frames or terminally reject recovery state before decode and parse. + revisit_when: + - Prime upstream exposes an equivalent public-root SDK proof, finite raw-byte default and override, privacy-safe typed terminal error, and no-replay reconnect semantics. + - Comet and Pylon can remove the fork token without weakening stock-package fail-closed detection or host memory bounds. diff --git a/.pylon/upstream-review.md b/.pylon/upstream-review.md index 576a81ac58..a385eac153 100644 --- a/.pylon/upstream-review.md +++ b/.pylon/upstream-review.md @@ -81,3 +81,14 @@ This ledger records Prime upstream evidence and the decision taken for each over - Fork boundaries remain intact: no inherited Linear gate or release workflow is restored; optional protocol behavior remains capability-negotiated; the `main` mirror and `pylon` product branch roles are unchanged. - Validation: exact merge `npm run check`, installer render, browser smoke, root build, and `git diff --check` passed; the live generated model catalog was restored. A clean-env focused boundary set passed 269 executions (18 continuation, 12 correlated lifecycle, 18 lazy-subagent, 109 supervisor-monitor, and 112 recursion). Native-install full shard 1 passed 1,466 tests with 24 skipped; shard 3 passed 1,241 with 22 skipped. Shard 2 recorded 1,865 passes with 9 skips while three load-sensitive cases tripped wall-clock/process thresholds; all 28 affected cases passed immediately in isolation. The real-process supervisor suite passed 13 tests with 8 fixture-gated skips. Both stock/current `v0.8.1` adoption directions and the 36 MiB compiled transfer bound passed. Trusted exact-head CI remains mandatory before merge. - Independent exact-commit security/regression review found no P0/P1, verified both-parent ancestry and all 1,245 tree entries, and approved the retained conflict semantics. Revisit each hybrid only when Prime supplies an explicitly negotiated equivalent without weakening Pylon privacy, lifecycle, recovery, cleanup, or stock-Prime fallback guarantees. + +## 2026-08-30 — bounded public daemon client ingress candidate + +- Pylon base: `pylon@e7871eb699d0f65047a21d179216ebfec7755d0c`, the reviewed merge of upstream gate PR #15. Upstream evidence remains fully audited through `PrimeIntellect-ai/prime-agent@a903d4b6768f484bd6d459b7b0aa7dee38e461e2`; this client-local candidate does not advance `reviewed_upstream_commit`. +- Overlap decision: **hybridize**. Prime upstream PR #480 improves reconnect/request replay and current upstream provides generic optional `maxLineLength`, but the public `DaemonClient` still reads unbounded frames. The generic option counts decoded JavaScript units, exposes a prefix callback, discards through LF, and resumes. It cannot prove or implement a raw-byte, privacy-safe, terminal public-client boundary. +- Public contract: the immutable root-exported `PRIME_AGENT_SDK_FEATURES` registry contains `bounded_daemon_ingress_v1`; `DaemonClientOptions.maxInboundFrameBytes` has a finite 128 MiB default; and `DaemonInboundFrameTooLargeError` exposes only code `daemon_inbound_frame_too_large` and the configured limit. This is local artifact metadata, not a daemon capability, hello offer, protocol/schema/version signal, constructor-arity probe, or hostile-code attestation. +- Framing and recovery: the public socket remains in byte mode and copies pending bytes into bounded 64 KiB owned pages. LF is excluded and CR before LF is counted then stripped. Overflow is terminal for the exact socket identity, releases reader state, rejects connect/hello/all requests with preservation disabled, clears stale hello and replay state, notifies close listeners once, suppresses the current automatic recovery loop, and permits only a later explicit empty-buffer reconnect. Existing decoded character-limit/discard-resume behavior remains unchanged for its current stderr and RPC callers. +- Consumer budget: Prime keeps a compatibility-first 128 MiB per-frame default. Comet #5 will require the exact registry token, explicitly select 64 MiB, run the isolated long-lived Node host with a 512 MiB heap, keep downstream control frames/queues separately small, and map the typed SDK error to a fixed privacy-safe host code. Stock 0.8.1 remains bootstrap-compatible but is rejected locally for session hosting before create/attach because it lacks the token. +- Design review: independent Prime transport and Comet consumer reviews agreed on the default, byte/CRLF semantics, fixed-page allocation, terminal no-replay state, explicit reconnect, error privacy, and stock migration. They identified raw decoding, request preservation, discard/resume, auto-reconnect loops, retained buffer backing stores, connecting-phase races, native-error projection, and token-without-semantics as P0 gates. +- Validation: the two focused files pass 53 tests, including a real Unix-socket immediate-on-accept overflow, exact byte/CRLF/multibyte boundaries, many tiny chunks, connect/hello/all-request cleanup, preserved-request rejection, stale-hello clearing, no replay, running-auto-loop fencing, survival of an immediate explicit reconnect and its later fresh auto policy, one privacy-safe close, and cumulative bounded snapshot chunks. The real public-client transport receipt reconstructed 100 MiB across 267 snapshot chunks and 269 total frames (maximum 393,433 raw bytes, SHA-256 `5b05b298e974f3b9e40f0a1a8188f50984a4f18fb329e050324296632d3d9dfc`) and 500 MiB across 1,334 snapshot chunks and 1,336 total frames (maximum 393,435 raw bytes, SHA-256 `07171a278ab228de25c3d7ece952e0562d9d42d16d91580a9db5b80023d2493b`) through an actual `DaemonClient` configured to 64 MiB. These are deterministic cumulative transport/reconstruction receipts, not claims that the legacy monolithic benchmark is bounded attach integration. Both stock/current 0.8.1 adoption directions and the compiled 36 MiB indivisible-message supervisor-to-public-client regression passed (3/3). The clean real-process supervisor suite passed 13 tests with 8 fixture-gated skips. A 16-file wider affected run recorded 640 passes before one update-restart subprocess inherited the host worker's `PRIME_AGENT_INTERNAL_*` state and timed out before hello; after scrubbing every internal prefix, that exact case passed alone in 4.4 seconds. Exact valid 64 MiB frames passed under a 512 MiB V8 heap with 458 MB peak RSS/252 MB peak heap for ASCII; the high-expansion invalid-UTF-8 replacement case also passed with 670 MB peak RSS/452 MB peak heap, confirming the need for small Comet queues and OS headroom beyond the V8 ceiling. An exact valid default 128 MiB frame passed under a 1 GiB heap with 730 MB peak RSS/455 MB peak heap. `npm run check`, root build, installer/browser checks, YAML parse, and `git diff --check` pass; live model generation was restored. The rebuilt public root exports the frozen registry, client, options d.ts, default, and error, and `npm pack --dry-run` contains their JS/d.ts plus both docs. Installed stock 0.8.1 has no registry and silently ignores the second constructor argument, so namespace token gating fails closed as designed. Fresh API/resource re-review found no other current P0/P1. Exact committed-head security renewal and trusted hosted CI remain mandatory before merge. +- Revisit when Prime upstream provides an equivalent public-root proof and raw-byte terminal client contract, and both Pylon and Comet can remove the fork token without weakening stock-package detection, privacy, recovery integrity, or host memory bounds. diff --git a/packages/coding-agent/.changes/bounded-daemon-ingress.md b/packages/coding-agent/.changes/bounded-daemon-ingress.md new file mode 100644 index 0000000000..c40fae08de --- /dev/null +++ b/packages/coding-agent/.changes/bounded-daemon-ingress.md @@ -0,0 +1 @@ +- Bounded public daemon client JSONL ingress by raw bytes and added explicit SDK safety metadata ([#13](https://github.com/pylon-code/prime-agent/issues/13)). diff --git a/packages/coding-agent/docs/daemon.md b/packages/coding-agent/docs/daemon.md index 35e4ce9dd9..7f1d5b47e9 100644 --- a/packages/coding-agent/docs/daemon.md +++ b/packages/coding-agent/docs/daemon.md @@ -73,7 +73,7 @@ Due ticks are claimed and advanced before prompt delivery. A crash therefore doe Resident workers keep scheduling across supervisor replacement. Worker recovery marks uncertain claims interrupted, keeps the advanced schedule, and resumes future ticks only. The supervisor routes schedule commands and merges worker summaries for global listing. -## Public Daemon Protocol v4 +## Public Daemon Protocol v7 The public local socket is JSONL-framed. The current protocol provides: @@ -94,6 +94,18 @@ Protocol v1 is retained only for the one-release update handoff that prepares an JSON and RPC client modes do not expose daemon greetings, envelopes, snapshot records, lifecycle events, or connection metadata. +### Public client ingress bound + +The public SDK `DaemonClient` bounds each inbound JSONL frame before UTF-8 decoding or JSON parsing. The default is 128 MiB and embedders can set a lower finite limit with `maxInboundFrameBytes`. Limits are positive safe integers and stay fixed for the lifetime of the client, including reconnects. + +The bound counts raw bytes before LF. LF is excluded. A CR immediately before LF counts toward the limit and is then removed. The limit applies to each frame, not to the cumulative snapshot: normal snapshots remain valid as a sequence of bounded chunks. A single indivisible message or legacy monolithic frame above the limit requires a deliberate higher finite override. + +Overflow, including bytes without LF, raises `DaemonInboundFrameTooLargeError`. It is terminal for that socket epoch: the client does not truncate, discard-and-resume, process later bytes from that socket, preserve requests for replay, or automatically reconnect to the same peer. Every handshake and request waiter is rejected, close listeners are notified once, and an explicit later reconnect starts with an empty byte buffer and the same configured bound. The error exposes only its stable code and configured limit, never frame content. + +`PRIME_AGENT_SDK_FEATURES` contains `bounded_daemon_ingress_v1` when this complete client-local contract is present. This token describes the installed SDK artifact. It is not a daemon capability, hello offer, protocol version, or schema revision. + +The raw frame limit bounds pending reader-owned bytes. A valid near-limit frame also needs memory for decoded strings, parsed objects, and consumer state, so embedders must set their process heap and downstream queues separately. + ## Reconnect, Replay, and Snapshots Every sequenced event belongs to a worker generation. Clients retain the last `{ generation, sequence }` cursor and present it on attach. The server reports whether the requested interval is complete, partial, or unavailable. @@ -160,7 +172,9 @@ npx tsx test/daemon-multiclient-bench.ts npx tsx test/daemon-multiclient-bench.ts --generated-session-mib 100 npx tsx test/daemon-multiclient-bench.ts --generated-session-mib 500 npx tsx test/daemon-multiclient-bench.ts --session-file /path/to/session.jsonl +npx tsx test/daemon-bounded-ingress-bench.ts --generated-session-mib 100 +npx tsx test/daemon-bounded-ingress-bench.ts --generated-session-mib 500 PRIME_AGENT_STRESS_WORKERS=50 npx tsx ../../node_modules/vitest/dist/cli.js --run test/daemon-supervisor-process.test.ts -t "hosts resident roots" ``` -The benchmark compares fanout and attach paths, including serialization count, throughput, elapsed time, and sampled RSS. The stress case starts many resident roots and verifies that their schedules advance independently while sessions are busy. +The multi-client benchmark compares legacy and chunked fanout/attach paths, including serialization count, throughput, elapsed time, and sampled RSS. Its legacy monolithic path is not an ingress-safety receipt. The bounded-ingress benchmark sends deterministic snapshot-like chunks through a real public `DaemonClient`, asserts each raw frame stays within both its 512 KiB target and a 64 MiB client limit, then verifies the full 100/500 MiB cumulative byte count and SHA-256. The stress case starts many resident roots and verifies that their schedules advance independently while sessions are busy. diff --git a/packages/coding-agent/docs/sdk.md b/packages/coding-agent/docs/sdk.md index a8ee9aa372..fcc2a98997 100644 --- a/packages/coding-agent/docs/sdk.md +++ b/packages/coding-agent/docs/sdk.md @@ -47,6 +47,31 @@ The SDK is included in the main package. No separate installation needed. ## Core Concepts +### SDK feature metadata and bounded daemon ingress + +Long-lived daemon embedders should verify client-local transport features from the package's public root before constructing a client. Use a namespace import when the same code must also load an older package: a named import of a new export fails while linking old ESM builds. + +```typescript +const sdk = await import("@earendil-works/pi-coding-agent"); +const features: unknown = sdk.PRIME_AGENT_SDK_FEATURES; +if (!Array.isArray(features) || !features.includes("bounded_daemon_ingress_v1")) { + throw new Error("This SDK cannot safely host a long-lived daemon session"); +} + +const client = new sdk.DaemonClient(socketPath, { + maxInboundFrameBytes: 64 * 1024 * 1024, +}); +await client.connect(); +``` + +`PRIME_AGENT_SDK_FEATURES` is immutable metadata for behavior implemented by the local SDK artifact. Do not infer it from package versions, constructor arity, method presence, daemon hello capabilities, protocol versions, or schema revisions. Older JavaScript constructors can silently ignore an extra options argument. + +`DaemonClientOptions.maxInboundFrameBytes` is the maximum raw bytes before LF in one inbound JSONL frame. It defaults to `DEFAULT_DAEMON_CLIENT_MAX_INBOUND_FRAME_BYTES` (128 MiB) and must be a positive safe integer. LF is excluded. A CR immediately before LF is counted and then stripped. + +`DaemonInboundFrameTooLargeError` has code `daemon_inbound_frame_too_large` and exposes the configured limit. It never includes frame content. Overflow terminally closes that socket, rejects handshake and request waiters even when request recovery was enabled, suppresses automatic replay/reconnect, and discards the partial buffer. A later explicit reconnect uses a fresh reader with the same bound. Applications that surface errors across a trust boundary should map the class or code to their own fixed message rather than forwarding an SDK error, stack, socket path, or daemon log path. + +The frame limit is not a total heap limit. Valid frames also allocate decoded strings, parsed values, and application state. Set a lower explicit frame limit only with enough heap for valid boundary frames, and bound all downstream queues independently. + ### createAgentSession() The main factory function for a single `AgentSession`. diff --git a/packages/coding-agent/src/index.ts b/packages/coding-agent/src/index.ts index d878e30ff2..127ae91752 100644 --- a/packages/coding-agent/src/index.ts +++ b/packages/coding-agent/src/index.ts @@ -317,6 +317,7 @@ export { type DaemonClientCapability, type DaemonClientId, type DaemonClientMessageListener, + type DaemonClientOptions, type DaemonCommand, type DaemonCommandEnvelope, type DaemonCommandId, @@ -324,6 +325,7 @@ export { type DaemonEventId, type DaemonEventMeta, type DaemonEventSequence, + DaemonInboundFrameTooLargeError, type DaemonModeOptions, type DaemonOutbound, type DaemonOwnedSessionCleanupResult, @@ -336,6 +338,7 @@ export { type DaemonResponse, type DaemonResumeCursor, type DaemonSessionSnapshot, + DEFAULT_DAEMON_CLIENT_MAX_INBOUND_FRAME_BYTES, defaultDaemonSocketPath, type InteractiveInitialPrompt, InteractiveMode, @@ -410,6 +413,7 @@ export { Theme, type ThemeColor, } from "./modes/interactive/theme/theme.js"; +export { PRIME_AGENT_SDK_FEATURES, type PrimeAgentSdkFeature } from "./sdk-features.js"; // Clipboard utilities export { copyToClipboard } from "./utils/clipboard.js"; export { parseFrontmatter, stripFrontmatter } from "./utils/frontmatter.js"; diff --git a/packages/coding-agent/src/modes/daemon/daemon-client.ts b/packages/coding-agent/src/modes/daemon/daemon-client.ts index cffdb69782..e92f44efc1 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-client.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-client.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import { createConnection, type Socket } from "node:net"; import { getDaemonLogPath } from "../../config.js"; -import { attachJsonlLineReader, serializeJsonLine } from "../rpc/jsonl.js"; +import { attachBoundedJsonlByteReader, serializeJsonLine } from "../rpc/jsonl.js"; import { createDaemonCommandEnvelope, DAEMON_COMMAND_ENVELOPE_MIN_PROTOCOL_VERSION, @@ -32,6 +32,22 @@ export type DaemonClientMessageListener = (message: DaemonOutbound) => void; export type DaemonClientCloseListener = (error: Error) => void; export type DaemonClientProgressListener = (message: DaemonRequestProgress) => void; +export const DEFAULT_DAEMON_CLIENT_MAX_INBOUND_FRAME_BYTES = 128 * 1024 * 1024; + +export interface DaemonClientOptions { + /** Maximum raw bytes before LF in one inbound daemon JSONL frame. */ + readonly maxInboundFrameBytes?: number; +} + +export class DaemonInboundFrameTooLargeError extends Error { + readonly code = "daemon_inbound_frame_too_large" as const; + + constructor(readonly maxInboundFrameBytes: number) { + super(`Prime Agent daemon inbound frame exceeded the configured ${maxInboundFrameBytes}-byte limit.`); + this.name = "DaemonInboundFrameTooLargeError"; + } +} + export interface DaemonClientRequestOptions { onProgress?: DaemonClientProgressListener; } @@ -118,14 +134,29 @@ export class DaemonClient { private closed = false; private helloMessage?: DaemonHello; private daemonClosingReason?: DaemonClosingReason; + private terminalTransportError?: DaemonInboundFrameTooLargeError; private reconnectPromise?: Promise; + private connectingSocket?: { socket: Socket; reject: (error: Error) => void }; + private readonly maxInboundFrameBytes: number; private readonly helloWaiters = new Set<{ resolve: (hello: DaemonHello) => void; reject: (error: Error) => void; timeout: ReturnType; }>(); - constructor(private readonly socketPath: string) {} + constructor( + private readonly socketPath: string, + options: DaemonClientOptions = {}, + ) { + const maxInboundFrameBytes = + options.maxInboundFrameBytes === undefined + ? DEFAULT_DAEMON_CLIENT_MAX_INBOUND_FRAME_BYTES + : options.maxInboundFrameBytes; + if (!Number.isSafeInteger(maxInboundFrameBytes) || maxInboundFrameBytes <= 0) { + throw new RangeError("maxInboundFrameBytes must be a positive safe integer"); + } + this.maxInboundFrameBytes = maxInboundFrameBytes; + } get hello(): DaemonHello | undefined { return this.helloMessage; @@ -145,6 +176,9 @@ export class DaemonClient { return this.helloMessage; } if (!this.socket || this.socket.destroyed) { + if (this.terminalTransportError) { + throw this.terminalTransportError; + } throw new Error( `Cannot wait for the Prime Agent daemon handshake because the daemon is not connected. ${daemonEndpointDetails(this.socketPath)}`, ); @@ -172,41 +206,56 @@ export class DaemonClient { } this.helloMessage = undefined; this.daemonClosingReason = undefined; + this.terminalTransportError = undefined; const socket = createConnection(this.socketPath); this.socket = socket; - this.detachReader = attachJsonlLineReader(socket, (line) => this.handleLine(line)); - await new Promise((resolve, reject) => { - const timeout = setTimeout(() => { - cleanup(); - this.clearSocketReference(socket); - socket.destroy(); - reject( - new Error( - `Timed out after ${timeoutMs}ms connecting to the Prime Agent daemon. ${daemonEndpointDetails(this.socketPath)}`, - ), - ); - }, timeoutMs); + const connection = new Promise((resolve, reject) => { const cleanup = () => { clearTimeout(timeout); socket.off("connect", onConnect); socket.off("error", onError); + socket.off("close", onCloseBeforeConnect); + }; + const rejectConnection = (error: Error) => { + cleanup(); + if (this.connectingSocket?.socket === socket) { + this.connectingSocket = undefined; + } + this.clearSocketReference(socket); + socket.destroy(); + reject(error); }; const onConnect = () => { cleanup(); + if (this.connectingSocket?.socket === socket) { + this.connectingSocket = undefined; + } resolve(); }; const onError = (error: Error) => { - cleanup(); - this.clearSocketReference(socket); - reject( + rejectConnection( + error instanceof DaemonInboundFrameTooLargeError + ? error + : new Error( + `Failed to connect to the Prime Agent daemon: ${error.message}. ${daemonEndpointDetails(this.socketPath)}`, + ), + ); + }; + const onCloseBeforeConnect = () => { + rejectConnection(new DaemonSocketClosedError(this.socketPath, this.daemonClosingReason)); + }; + const timeout = setTimeout(() => { + rejectConnection( new Error( - `Failed to connect to the Prime Agent daemon: ${error.message}. ${daemonEndpointDetails(this.socketPath)}`, + `Timed out after ${timeoutMs}ms connecting to the Prime Agent daemon. ${daemonEndpointDetails(this.socketPath)}`, ), ); - }; + }, timeoutMs); + this.connectingSocket = { socket, reject: rejectConnection }; socket.once("connect", onConnect); socket.once("error", onError); + socket.once("close", onCloseBeforeConnect); }); socket.on("error", (error) => @@ -220,6 +269,12 @@ export class DaemonClient { socket.on("close", () => this.notifyClosed(socket, new DaemonSocketClosedError(this.socketPath, this.daemonClosingReason)), ); + this.detachReader = attachBoundedJsonlByteReader(socket, (line) => this.handleLine(line), { + maxFrameBytes: this.maxInboundFrameBytes, + onFrameTooLarge: () => this.failInboundFrame(socket), + }); + + await connection; } async reconnect(timeoutMs = 3000): Promise { @@ -253,8 +308,11 @@ export class DaemonClient { /** Discard a partially recovered transport so the next retry can reconnect cleanly. */ resetTransportForReconnect(): void { - const socket = this.socket; - if (!socket) { + this.resetTransportForReconnectAttempt(this.socket); + } + + private resetTransportForReconnectAttempt(socket: Socket | undefined): void { + if (!socket || this.socket !== socket) { return; } this.clearSocketReference(socket); @@ -296,6 +354,9 @@ export class DaemonClient { options: DaemonClientRequestOptions = {}, ): Promise { if (!this.socket || this.socket.destroyed) { + if (this.terminalTransportError) { + throw this.terminalTransportError; + } throw new Error( `Cannot send daemon command "${command.type}" because the Prime Agent daemon is not connected. ${daemonEndpointDetails(this.socketPath)}`, ); @@ -348,6 +409,9 @@ export class DaemonClient { compatibilities: readonly DaemonCommandCompatibility[] = [], ): Promise { if (!this.socket || this.socket.destroyed) { + if (this.terminalTransportError) { + throw this.terminalTransportError; + } throw new Error( `Cannot send daemon command "${command.type}" because the Prime Agent daemon is not connected. ${daemonEndpointDetails(this.socketPath)}`, ); @@ -399,6 +463,7 @@ export class DaemonClient { close(): void { this.closed = true; this.reconnectOptions = undefined; + this.terminalTransportError = undefined; this.detachReader?.(); this.detachReader = undefined; this.rejectAll( @@ -411,6 +476,26 @@ export class DaemonClient { this.socket = undefined; } + private failInboundFrame(socket: Socket): void { + if (this.socket !== socket) { + return; + } + const error = new DaemonInboundFrameTooLargeError(this.maxInboundFrameBytes); + this.terminalTransportError = error; + const rejectConnecting = this.connectingSocket?.socket === socket ? this.connectingSocket.reject : undefined; + this.reconnectOptions = undefined; + this.helloMessage = undefined; + this.daemonClosingReason = undefined; + this.clearSocketReference(socket); + this.rejectAll(error, false); + this.emitCloseListeners(error); + if (rejectConnecting) { + rejectConnecting(error); + } else { + socket.destroy(); + } + } + private clearSocketReference(socket: Socket): void { if (this.socket !== socket) { return; @@ -536,15 +621,23 @@ export class DaemonClient { } } + private emitCloseListeners(error: Error): void { + for (const listener of [...this.closeListeners]) { + try { + listener(error); + } catch { + // One consumer must not hide a terminal transport failure from others. + } + } + } + private notifyClosed(socket: Socket, error: Error): void { if (this.socket !== socket) { return; } this.clearSocketReference(socket); this.rejectAll(error, this.requestRecoveryEnabled); - for (const listener of [...this.closeListeners]) { - listener(error); - } + this.emitCloseListeners(error); if (this.reconnectOptions && !this.closed) { void this.autoReconnect(error); } @@ -564,18 +657,28 @@ export class DaemonClient { let attempt = 0; let lastError: Error = cause; while (!this.closed && this.reconnectOptions === options && Date.now() < deadline) { + let reconnectSocket: Socket | undefined; try { await options.recoverDaemon(); if (this.closed || this.reconnectOptions !== options) { return; } - await this.connect(RECONNECT_CONNECT_TIMEOUT_MS); + const socketBeforeConnect = this.socket; + const connectAttempt = this.connect(RECONNECT_CONNECT_TIMEOUT_MS); + reconnectSocket = socketBeforeConnect === undefined ? this.socket : undefined; + await connectAttempt; await this.waitForHello(RECONNECT_HELLO_TIMEOUT_MS); + if (this.closed || this.reconnectOptions !== options || !this.isConnected) { + return; + } this.emitReconnectStatus({ status: "connected" }); return; } catch (error) { lastError = error instanceof Error ? error : new Error(String(error)); - this.resetTransportForReconnect(); + if (this.closed || this.reconnectOptions !== options) { + return; + } + this.resetTransportForReconnectAttempt(reconnectSocket); const remainingMs = deadline - Date.now(); if (remainingMs <= 0) { break; diff --git a/packages/coding-agent/src/modes/index.ts b/packages/coding-agent/src/modes/index.ts index 349e06e340..025476e1ed 100644 --- a/packages/coding-agent/src/modes/index.ts +++ b/packages/coding-agent/src/modes/index.ts @@ -2,6 +2,7 @@ * Run modes for the coding agent. */ +export { PRIME_AGENT_SDK_FEATURES, type PrimeAgentSdkFeature } from "../sdk-features.js"; export { type AcpModeOptions, acpStopReason, @@ -66,6 +67,9 @@ export { DaemonCapabilityUnavailableError, DaemonClient, type DaemonClientMessageListener, + type DaemonClientOptions, + DaemonInboundFrameTooLargeError, + DEFAULT_DAEMON_CLIENT_MAX_INBOUND_FRAME_BYTES, } from "./daemon/daemon-client.js"; export { type DaemonModeOptions, runDaemonMode } from "./daemon/daemon-mode.js"; export type { diff --git a/packages/coding-agent/src/modes/rpc/jsonl.ts b/packages/coding-agent/src/modes/rpc/jsonl.ts index 6ecb62f0a9..38717444cc 100644 --- a/packages/coding-agent/src/modes/rpc/jsonl.ts +++ b/packages/coding-agent/src/modes/rpc/jsonl.ts @@ -115,3 +115,142 @@ export function attachJsonlLineReader( stream.off("end", onEnd); }; } + +export interface BoundedJsonlByteReaderOptions { + /** Maximum bytes before LF in one frame. LF is excluded; a preceding CR counts. */ + maxFrameBytes: number; + /** Called once after the reader becomes permanently terminal. */ + onFrameTooLarge: () => void; +} + +const BOUNDED_JSONL_PAGE_BYTES = 64 * 1024; + +/** + * Attach a terminal, raw-byte-bounded JSONL reader. + * + * The stream must remain in byte mode. Bytes are copied into fixed-size owned + * pages before UTF-8 decoding so pending storage is bounded without retaining + * arbitrary caller buffers. Unlike `maxLineLength`, overflow never resumes at + * a later record: the caller must permanently close this transport epoch. + */ +export function attachBoundedJsonlByteReader( + stream: Readable, + onLine: (line: string) => void, + options: BoundedJsonlByteReaderOptions, +): () => void { + if (!Number.isSafeInteger(options.maxFrameBytes) || options.maxFrameBytes <= 0) { + throw new RangeError("maxFrameBytes must be a positive safe integer"); + } + + let pages: Buffer[] = []; + let pageLengths: number[] = []; + let allocatedBytes = 0; + let pendingBytes = 0; + let terminal = false; + let attached = true; + + const resetPending = () => { + pages = []; + pageLengths = []; + allocatedBytes = 0; + pendingBytes = 0; + }; + + const detach = () => { + if (!attached) return; + attached = false; + stream.off("data", onData); + stream.off("end", onEnd); + }; + + const failTooLarge = () => { + if (terminal) return; + terminal = true; + detach(); + resetPending(); + options.onFrameTooLarge(); + }; + + const appendPending = (segment: Buffer): boolean => { + if (segment.length === 0) return true; + if (segment.length > options.maxFrameBytes - pendingBytes) { + failTooLarge(); + return false; + } + + let offset = 0; + while (offset < segment.length) { + let page = pages.at(-1); + let pageLength = pageLengths.at(-1) ?? 0; + if (!page || pageLength === page.length) { + const capacity = Math.min(BOUNDED_JSONL_PAGE_BYTES, options.maxFrameBytes - allocatedBytes); + page = Buffer.allocUnsafe(capacity); + pages.push(page); + pageLengths.push(0); + allocatedBytes += capacity; + pageLength = 0; + } + const copied = Math.min(page.length - pageLength, segment.length - offset); + segment.copy(page, pageLength, offset, offset + copied); + pageLengths[pageLengths.length - 1] = pageLength + copied; + offset += copied; + pendingBytes += copied; + } + return true; + }; + + const emitPending = () => { + const decoder = new StringDecoder("utf8"); + const decoded: string[] = []; + for (let i = 0; i < pages.length; i++) { + const length = pageLengths[i] ?? 0; + if (length > 0) { + decoded.push(decoder.write(pages[i]!.subarray(0, length))); + } + } + decoded.push(decoder.end()); + let line = decoded.join(""); + if (line.endsWith("\r")) { + line = line.slice(0, -1); + } + resetPending(); + onLine(line); + }; + + function onData(chunk: string | Buffer): void { + if (terminal) return; + const bytes = typeof chunk === "string" ? Buffer.from(chunk, "utf8") : chunk; + let start = 0; + while (start <= bytes.length && !terminal) { + const newline = bytes.indexOf(0x0a, start); + if (newline === -1) { + appendPending(bytes.subarray(start)); + return; + } + if (!appendPending(bytes.subarray(start, newline))) { + return; + } + emitPending(); + start = newline + 1; + } + } + + function onEnd(): void { + if (terminal) return; + detach(); + if (pendingBytes > 0) { + emitPending(); + } else { + resetPending(); + } + } + + stream.on("data", onData); + stream.on("end", onEnd); + + return () => { + terminal = true; + detach(); + resetPending(); + }; +} diff --git a/packages/coding-agent/src/sdk-features.ts b/packages/coding-agent/src/sdk-features.ts new file mode 100644 index 0000000000..1319b46d98 --- /dev/null +++ b/packages/coding-agent/src/sdk-features.ts @@ -0,0 +1,9 @@ +/** + * Client-artifact features exposed by this SDK build. + * + * These tokens describe local SDK behavior. They are not daemon capabilities, + * protocol versions, schema revisions, or proof about a remote peer. + */ +export const PRIME_AGENT_SDK_FEATURES = Object.freeze(["bounded_daemon_ingress_v1"] as const); + +export type PrimeAgentSdkFeature = (typeof PRIME_AGENT_SDK_FEATURES)[number]; diff --git a/packages/coding-agent/test/daemon-bounded-ingress-bench.ts b/packages/coding-agent/test/daemon-bounded-ingress-bench.ts new file mode 100644 index 0000000000..120e0c5081 --- /dev/null +++ b/packages/coding-agent/test/daemon-bounded-ingress-bench.ts @@ -0,0 +1,186 @@ +/** + * Real-socket bounded DaemonClient ingress receipt. + * + * Run from packages/coding-agent: + * + * npx tsx test/daemon-bounded-ingress-bench.ts --generated-session-mib 100 + * npx tsx test/daemon-bounded-ingress-bench.ts --generated-session-mib 500 + */ +import { createHash, randomUUID } from "node:crypto"; +import { once } from "node:events"; +import { unlink } from "node:fs/promises"; +import { createServer, type Socket } from "node:net"; +import { DaemonClient } from "../src/modes/daemon/daemon-client.js"; + +const MEBIBYTE = 1024 * 1024; +const CLIENT_MAX_FRAME_BYTES = 64 * MEBIBYTE; +const TARGET_PAYLOAD_BYTES = 384 * 1024; +const TARGET_MAX_WIRE_FRAME_BYTES = 512 * 1024; + +function generatedSessionMib(): number { + const flag = process.argv.indexOf("--generated-session-mib"); + const value = flag === -1 ? Number.NaN : Number(process.argv[flag + 1]); + if (!Number.isSafeInteger(value) || value <= 0) { + throw new Error("--generated-session-mib must be a positive integer"); + } + return value; +} + +async function writeFrame(socket: Socket, value: unknown): Promise { + const frame = Buffer.from(`${JSON.stringify(value)}\n`); + const rawFrameBytes = frame.length - 1; + if (rawFrameBytes > TARGET_MAX_WIRE_FRAME_BYTES || rawFrameBytes > CLIENT_MAX_FRAME_BYTES) { + throw new Error(`Generated frame exceeded its declared bound: ${rawFrameBytes}`); + } + if (!socket.write(frame)) { + await once(socket, "drain"); + } + return rawFrameBytes; +} + +const sessionMib = generatedSessionMib(); +const totalPayloadBytes = sessionMib * MEBIBYTE; +const activeSessionId = "active-bounded-ingress-benchmark"; +const snapshotId = `snapshot-${randomUUID()}`; +const socketPath = + process.platform === "win32" + ? `\\\\.\\pipe\\prime-bounded-ingress-${process.pid}-${randomUUID()}` + : `/tmp/prime-bounded-ingress-${process.pid}-${randomUUID().slice(0, 8)}.sock`; + +let resolveSender!: (result: { chunkCount: number; digest: string; maxFrameBytes: number }) => void; +let rejectSender!: (error: Error) => void; +const sender = new Promise<{ chunkCount: number; digest: string; maxFrameBytes: number }>((resolve, reject) => { + resolveSender = resolve; + rejectSender = reject; +}); + +const server = createServer((socket) => { + socket.on("error", (error) => rejectSender(error)); + void (async () => { + let maxFrameBytes = await writeFrame(socket, { + type: "session_snapshot_begin", + activeSessionId, + snapshotId, + snapshot: { activeSessionId }, + messageCount: Math.ceil(totalPayloadBytes / TARGET_PAYLOAD_BYTES), + targetChunkBytes: TARGET_MAX_WIRE_FRAME_BYTES, + purpose: "attach", + }); + const hash = createHash("sha256"); + let offset = 0; + let index = 0; + while (offset < totalPayloadBytes) { + const length = Math.min(TARGET_PAYLOAD_BYTES, totalPayloadBytes - offset); + const content = "x".repeat(length); + hash.update(content); + maxFrameBytes = Math.max( + maxFrameBytes, + await writeFrame(socket, { + type: "session_snapshot_chunk", + activeSessionId, + snapshotId, + index, + messages: [{ role: "user", content, timestamp: index }], + }), + ); + offset += length; + index++; + } + const digest = hash.digest("hex"); + maxFrameBytes = Math.max( + maxFrameBytes, + await writeFrame(socket, { + type: "session_snapshot_end", + activeSessionId, + snapshotId, + chunkCount: index, + lastEventSequence: 0, + digest, + totalPayloadBytes, + }), + ); + resolveSender({ chunkCount: index, digest, maxFrameBytes }); + })().catch((error) => rejectSender(error instanceof Error ? error : new Error(String(error)))); +}); +await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(socketPath, resolve); +}); + +const client = new DaemonClient(socketPath, { maxInboundFrameBytes: CLIENT_MAX_FRAME_BYTES }); +const receivedHash = createHash("sha256"); +let receivedBytes = 0; +let receivedChunks = 0; +let resolveComplete!: (end: { chunkCount: number; digest: string; totalPayloadBytes: number }) => void; +let rejectComplete!: (error: Error) => void; +const complete = new Promise<{ chunkCount: number; digest: string; totalPayloadBytes: number }>((resolve, reject) => { + resolveComplete = resolve; + rejectComplete = reject; +}); +client.onMessage((message) => { + const candidate = message as unknown as Record; + if (candidate.type === "session_snapshot_chunk") { + if (candidate.snapshotId !== snapshotId || candidate.index !== receivedChunks) { + rejectComplete(new Error("Received an out-of-order snapshot chunk")); + return; + } + const messages = candidate.messages; + const content = + Array.isArray(messages) && messages.length === 1 && typeof messages[0]?.content === "string" + ? messages[0].content + : undefined; + if (content === undefined) { + rejectComplete(new Error("Received an invalid snapshot chunk payload")); + return; + } + receivedHash.update(content); + receivedBytes += Buffer.byteLength(content); + receivedChunks++; + return; + } + if (candidate.type === "session_snapshot_end") { + resolveComplete(candidate as unknown as { chunkCount: number; digest: string; totalPayloadBytes: number }); + } +}); +client.onClose((error) => rejectComplete(error)); +let timeout!: ReturnType; +const timedComplete = Promise.race([ + complete, + new Promise((_, reject) => { + timeout = setTimeout(() => reject(new Error("bounded ingress timed out")), 120_000); + timeout.unref(); + }), +]); + +try { + await client.connect(); + const [sent, end] = await Promise.all([sender, timedComplete]); + const receivedDigest = receivedHash.digest("hex"); + if ( + receivedBytes !== totalPayloadBytes || + receivedChunks !== sent.chunkCount || + end.chunkCount !== sent.chunkCount || + end.totalPayloadBytes !== totalPayloadBytes || + end.digest !== sent.digest || + receivedDigest !== sent.digest + ) { + throw new Error("Bounded ingress reconstruction did not match the generated transcript"); + } + console.log( + JSON.stringify({ + sessionMib, + totalPayloadBytes, + chunkCount: sent.chunkCount, + maxFrameBytes: sent.maxFrameBytes, + clientMaxFrameBytes: CLIENT_MAX_FRAME_BYTES, + digest: sent.digest, + }), + ); +} finally { + clearTimeout(timeout); + client.close(); + await new Promise((resolve) => server.close(() => resolve())); + if (process.platform !== "win32") { + await unlink(socketPath).catch(() => {}); + } +} diff --git a/packages/coding-agent/test/daemon-client.test.ts b/packages/coding-agent/test/daemon-client.test.ts index 8e291ac05c..b21b200bcd 100644 --- a/packages/coding-agent/test/daemon-client.test.ts +++ b/packages/coding-agent/test/daemon-client.test.ts @@ -1,5 +1,15 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { DaemonClient, getDaemonSocketCloseReason } from "../src/modes/daemon/daemon-client.js"; +import type { + DaemonClientOptions as RootDaemonClientOptions, + PrimeAgentSdkFeature as RootPrimeAgentSdkFeature, +} from "../src/index.js"; +import * as publicSdk from "../src/index.js"; +import { + DaemonClient, + DaemonInboundFrameTooLargeError, + DEFAULT_DAEMON_CLIENT_MAX_INBOUND_FRAME_BYTES, + getDaemonSocketCloseReason, +} from "../src/modes/daemon/daemon-client.js"; import { DAEMON_COMMAND_COMPATIBILITY, DAEMON_PROTOCOL_VERSION, @@ -121,6 +131,212 @@ describe("DaemonClient", () => { vi.useRealTimers(); }); + it("exports the bounded-ingress SDK proof and validates finite byte limits", () => { + const rootOptions: RootDaemonClientOptions = { maxInboundFrameBytes: 1 }; + const rootFeature: RootPrimeAgentSdkFeature = "bounded_daemon_ingress_v1"; + expect(Array.isArray(publicSdk.PRIME_AGENT_SDK_FEATURES)).toBe(true); + expect(publicSdk.PRIME_AGENT_SDK_FEATURES).toEqual([rootFeature]); + expect(Object.isFrozen(publicSdk.PRIME_AGENT_SDK_FEATURES)).toBe(true); + expect(publicSdk.DaemonClient).toBe(DaemonClient); + expect(publicSdk.DaemonInboundFrameTooLargeError).toBe(DaemonInboundFrameTooLargeError); + expect(publicSdk.DEFAULT_DAEMON_CLIENT_MAX_INBOUND_FRAME_BYTES).toBe( + DEFAULT_DAEMON_CLIENT_MAX_INBOUND_FRAME_BYTES, + ); + expect(DEFAULT_DAEMON_CLIENT_MAX_INBOUND_FRAME_BYTES).toBe(128 * 1024 * 1024); + expect(() => new publicSdk.DaemonClient("/tmp/prime-agent.sock", rootOptions)).not.toThrow(); + + for (const invalid of [ + 0, + -1, + 1.5, + Number.NaN, + Number.POSITIVE_INFINITY, + Number.MAX_SAFE_INTEGER + 1, + null, + "128", + true, + ]) { + expect( + () => + new DaemonClient("/tmp/prime-agent.sock", { + maxInboundFrameBytes: invalid as unknown as number, + }), + ).toThrow(RangeError); + } + expect(() => new DaemonClient("/tmp/prime-agent.sock")).not.toThrow(); + expect(() => new DaemonClient("/tmp/prime-agent.sock", { maxInboundFrameBytes: 1 })).not.toThrow(); + }); + + it("fails an oversized transport during connect with one privacy-safe typed close", async () => { + const marker = "raw-secret-marker"; + const client = new DaemonClient("/tmp/private-prime-agent.sock", { maxInboundFrameBytes: 256 }); + const closed: Error[] = []; + client.onClose(() => { + throw new Error("broken close consumer"); + }); + client.onClose((error) => closed.push(error)); + + const connectionError = captureRejection(client.connect()); + const socket = netMock.sockets[0]!; + socket.emit("data", Buffer.from(marker.repeat(20))); + const error = await connectionError; + + expect(error).toBeInstanceOf(DaemonInboundFrameTooLargeError); + expect(error).toMatchObject({ + code: "daemon_inbound_frame_too_large", + maxInboundFrameBytes: 256, + }); + expect(error.message).not.toContain(marker); + expect(error.message).not.toContain("/tmp/private-prime-agent.sock"); + expect(error.message).not.toContain("Daemon log"); + expect(closed).toEqual([error]); + expect(client.hello).toBeUndefined(); + await expect(client.waitForHello()).rejects.toBe(error); + await expect(client.request({ type: "list" })).rejects.toBe(error); + expect(client.isConnected).toBe(false); + expect(socket.destroyed).toBe(true); + + socket.emit("error", new Error("late socket error")); + socket.emit("close"); + socket.emit("data", Buffer.from('{"type":"late"}\n')); + expect(closed).toEqual([error]); + }); + + it("rejects preserved requests and hello waiters without replay after overflow", async () => { + const client = new DaemonClient("/tmp/prime-agent.sock", { maxInboundFrameBytes: 512 }); + client.enableRequestRecovery(); + const firstConnect = client.connect(); + const firstSocket = netMock.sockets[0]!; + firstSocket.emit("connect"); + await firstConnect; + emitHello(firstSocket); + + const firstResponse = client.request({ type: "list" }); + const secondResponse = client.request({ type: "list", all: true }); + expect(firstSocket.writes).toHaveLength(2); + firstSocket.emit("close"); + + const secondConnect = client.connect(); + const secondSocket = netMock.sockets[1]!; + secondSocket.emit("connect"); + await secondConnect; + const hello = client.waitForHello(); + const closed: Error[] = []; + client.onClose((error) => closed.push(error)); + secondSocket.emit("data", Buffer.alloc(513, 0x78)); + + const [helloError, firstError, secondError] = await Promise.all([ + captureRejection(hello), + captureRejection(firstResponse), + captureRejection(secondResponse), + ]); + expect(helloError).toBeInstanceOf(DaemonInboundFrameTooLargeError); + expect(firstError).toBe(helloError); + expect(secondError).toBe(helloError); + expect(closed).toEqual([helloError]); + expect(client.hello).toBeUndefined(); + expect(secondSocket.writes).toEqual([]); + + const thirdConnect = client.connect(); + const thirdSocket = netMock.sockets[2]!; + thirdSocket.emit("connect"); + await thirdConnect; + emitHello(thirdSocket); + expect(thirdSocket.writes).toEqual([]); + await expect(client.waitForHello()).resolves.toMatchObject({ type: "daemon_hello" }); + client.close(); + }); + + it("suppresses automatic recovery after an inbound framing violation", async () => { + const client = new DaemonClient("/tmp/prime-agent.sock", { maxInboundFrameBytes: 256 }); + const connect = client.connect(); + const socket = netMock.sockets[0]!; + socket.emit("connect"); + await connect; + emitHello(socket); + const recoverDaemon = vi.fn(async () => {}); + client.enableAutoReconnect({ recoverDaemon }); + + socket.emit("data", Buffer.alloc(257, 0x78)); + await Promise.resolve(); + expect(recoverDaemon).not.toHaveBeenCalled(); + expect(netMock.sockets).toHaveLength(1); + expect(client.hello).toBeUndefined(); + expect(client.isConnected).toBe(false); + client.close(); + }); + + it("does not let an old automatic attempt destroy an explicit reconnect after overflow", async () => { + const client = new DaemonClient("/tmp/prime-agent.sock", { maxInboundFrameBytes: 256 }); + const firstConnect = client.connect(); + const firstSocket = netMock.sockets[0]!; + firstSocket.emit("connect"); + await firstConnect; + emitHello(firstSocket); + + const recoverDaemon = vi.fn(async () => {}); + let explicitReconnect: Promise | undefined; + client.onClose((error) => { + if (error instanceof DaemonInboundFrameTooLargeError) { + explicitReconnect = client.reconnect(); + } + }); + client.enableAutoReconnect({ recoverDaemon }); + firstSocket.emit("close"); + await vi.waitFor(() => expect(netMock.sockets).toHaveLength(2)); + + const automaticSocket = netMock.sockets[1]!; + automaticSocket.emit("data", Buffer.alloc(257, 0x78)); + expect(explicitReconnect).toBeDefined(); + expect(netMock.sockets).toHaveLength(3); + const explicitSocket = netMock.sockets[2]!; + explicitSocket.emit("connect"); + await explicitReconnect; + emitHello(explicitSocket); + await expect(client.waitForHello()).resolves.toMatchObject({ type: "daemon_hello" }); + await Promise.resolve(); + + expect(recoverDaemon).toHaveBeenCalledOnce(); + expect(explicitSocket.destroyed).toBe(false); + expect(client.isConnected).toBe(true); + + const freshRecoverDaemon = vi.fn(async () => {}); + client.enableAutoReconnect({ recoverDaemon: freshRecoverDaemon }); + explicitSocket.emit("close"); + await vi.waitFor(() => expect(netMock.sockets).toHaveLength(4)); + const freshAutomaticSocket = netMock.sockets[3]!; + freshAutomaticSocket.emit("connect"); + await Promise.resolve(); + emitHello(freshAutomaticSocket); + await vi.waitFor(() => expect(client.isConnected).toBe(true)); + expect(freshRecoverDaemon).toHaveBeenCalledOnce(); + client.close(); + }); + + it("accepts cumulative snapshot chunks when every raw frame is within the bound", async () => { + const records = Array.from({ length: 12 }, (_, index) => + Buffer.from(`${JSON.stringify({ type: "session_snapshot_chunk", index, data: "x".repeat(128) })}\n`), + ); + const maxFrameBytes = Math.max(...records.map((record) => record.length - 1)); + const client = new DaemonClient("/tmp/prime-agent.sock", { maxInboundFrameBytes: maxFrameBytes }); + const connect = client.connect(); + const socket = netMock.sockets[0]!; + socket.emit("connect"); + await connect; + const received: number[] = []; + client.onMessage((message) => received.push((message as unknown as { index: number }).index)); + + for (const record of records) { + for (let offset = 0; offset < record.length; offset += 7) { + socket.emit("data", record.subarray(offset, offset + 7)); + } + } + + expect(received).toEqual(Array.from({ length: 12 }, (_, index) => index)); + expect(records.reduce((total, record) => total + record.length, 0)).toBeGreaterThan(maxFrameBytes); + client.close(); + }); + it("allows connect retry after the socket emits an error before connecting", async () => { const client = new DaemonClient("/tmp/prime-agent-missing.sock"); @@ -976,7 +1192,7 @@ describe("DaemonClient", () => { }); }); -async function captureRejection(promise: Promise): Promise { +async function captureRejection(promise: Promise): Promise { try { await promise; } catch (error) { diff --git a/packages/coding-agent/test/rpc-jsonl.test.ts b/packages/coding-agent/test/rpc-jsonl.test.ts index a16d392eb2..d81bd8eb4a 100644 --- a/packages/coding-agent/test/rpc-jsonl.test.ts +++ b/packages/coding-agent/test/rpc-jsonl.test.ts @@ -1,8 +1,12 @@ +import { randomUUID } from "node:crypto"; import { EventEmitter } from "node:events"; +import { unlink } from "node:fs/promises"; +import { createServer } from "node:net"; import { performance } from "node:perf_hooks"; import { Readable } from "node:stream"; import { describe, expect, test } from "vitest"; -import { attachJsonlLineReader, serializeJsonLine } from "../src/modes/rpc/jsonl.js"; +import { DaemonClient, DaemonInboundFrameTooLargeError } from "../src/modes/daemon/daemon-client.js"; +import { attachBoundedJsonlByteReader, attachJsonlLineReader, serializeJsonLine } from "../src/modes/rpc/jsonl.js"; /** * Drive the reader with precise control over chunk boundaries (Readable.from @@ -152,4 +156,140 @@ describe("RPC JSONL framing", () => { expect(overflows).toEqual(["abcde"]); expect(lines).toEqual(["ok"]); }); + + test("fails immediately when a no-LF frame exceeds its raw-byte limit", () => { + const lines: string[] = []; + let failures = 0; + const emitter = new EventEmitter(); + attachBoundedJsonlByteReader(emitter as unknown as Readable, (line) => lines.push(line), { + maxFrameBytes: 5, + onFrameTooLarge: () => failures++, + }); + + emitter.emit("data", Buffer.from("abc")); + emitter.emit("data", Buffer.from("de")); + expect(failures).toBe(0); + emitter.emit("data", Buffer.from("f")); + expect(failures).toBe(1); + + emitter.emit("data", Buffer.from("\nok\n")); + emitter.emit("end"); + expect(lines).toEqual([]); + expect(failures).toBe(1); + }); + + test("never resumes after a complete oversized frame", () => { + const lines: string[] = []; + let failures = 0; + const emitter = new EventEmitter(); + attachBoundedJsonlByteReader(emitter as unknown as Readable, (line) => lines.push(line), { + maxFrameBytes: 5, + onFrameTooLarge: () => failures++, + }); + + emitter.emit("data", Buffer.from("abcdef\nok\n")); + expect(lines).toEqual([]); + expect(failures).toBe(1); + }); + + test("counts UTF-8 bytes and counts CR while excluding LF", () => { + const allowed: string[] = []; + let allowedFailures = 0; + const allowedEmitter = new EventEmitter(); + attachBoundedJsonlByteReader(allowedEmitter as unknown as Readable, (line) => allowed.push(line), { + maxFrameBytes: 4, + onFrameTooLarge: () => allowedFailures++, + }); + const euro = Buffer.from("€", "utf8"); + allowedEmitter.emit("data", euro.subarray(0, 1)); + allowedEmitter.emit("data", euro.subarray(1)); + allowedEmitter.emit("data", Buffer.from("\nabc\r")); + allowedEmitter.emit("data", Buffer.from("\n")); + allowedEmitter.emit("end"); + expect(allowed).toEqual(["€", "abc"]); + expect(allowedFailures).toBe(0); + + let byteFailures = 0; + const byteEmitter = new EventEmitter(); + attachBoundedJsonlByteReader( + byteEmitter as unknown as Readable, + () => { + throw new Error("oversized UTF-8 was decoded"); + }, + { + maxFrameBytes: 2, + onFrameTooLarge: () => byteFailures++, + }, + ); + byteEmitter.emit("data", euro); + expect(byteFailures).toBe(1); + + let crFailures = 0; + const crEmitter = new EventEmitter(); + attachBoundedJsonlByteReader( + crEmitter as unknown as Readable, + () => { + throw new Error("oversized CRLF was decoded"); + }, + { + maxFrameBytes: 3, + onFrameTooLarge: () => crFailures++, + }, + ); + crEmitter.emit("data", Buffer.from("abc\r\n")); + expect(crFailures).toBe(1); + }); + + test("preserves bounded EOF lines and handles many tiny chunks", () => { + const lines: string[] = []; + let failures = 0; + const emitter = new EventEmitter(); + attachBoundedJsonlByteReader(emitter as unknown as Readable, (line) => lines.push(line), { + maxFrameBytes: 70_000, + onFrameTooLarge: () => failures++, + }); + for (let i = 0; i < 65_537; i++) { + emitter.emit("data", Buffer.from("x")); + } + emitter.emit("data", Buffer.from("\nfinal")); + emitter.emit("end"); + + expect(lines).toEqual(["x".repeat(65_537), "final"]); + expect(failures).toBe(0); + }); + + test("fails a real socket that overflows immediately on accept", async () => { + const socketPath = `/tmp/prime-ingress-${process.pid}-${randomUUID()}.sock`; + const server = createServer((socket) => { + socket.on("error", () => {}); + socket.write(Buffer.alloc(65, 0x78)); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(socketPath, resolve); + }); + + const client = new DaemonClient(socketPath, { maxInboundFrameBytes: 64 }); + const closed: Error[] = []; + client.onClose((error) => closed.push(error)); + try { + const outcome = client.connect().then(() => client.waitForHello(1000)); + const error = await outcome.then( + () => { + throw new Error("Expected immediate socket overflow"); + }, + (reason: unknown) => reason, + ); + expect(error).toBeInstanceOf(DaemonInboundFrameTooLargeError); + await new Promise((resolve) => setImmediate(resolve)); + expect(closed).toEqual([error]); + expect(client.isConnected).toBe(false); + } finally { + client.close(); + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + await unlink(socketPath).catch(() => {}); + } + }); });