fix(adapter-utils): enforce the duplex frame size bound on encode in both codec copies (#11983)
## Thinking Path > - Paperclip uses adapter utilities to move bounded messages between agent processes. > - The duplex frame codec encodes and decodes these messages. > - The decoder rejects frames above the documented byte limit. > - The encoder did not apply the same limit before it sent a frame. > - This mismatch let a sender write a frame that the peer rejected after transmission. > - This pull request applies the same byte limit to both codec copies and keeps the broker channel open. > - The benefit is a local error with stable request telemetry instead of a channel loss. ## Linked Issues or Issue Description No public GitHub issue exists for this change. The problem follows the bug report fields below. **What happened?** The duplex encoder could write a frame larger than `DEFAULT_MAX_DUPLEX_FRAME_BYTES`. The peer decoder then rejected the frame after transmission. In the WebSocket 1009 case, this closed the channel and reported a process exit. **Expected behavior** The encoder should reject an oversized frame before it writes bytes. The gateway should return HTTP 413. The broker should return a bounded terminal response and keep other requests active. **Steps to reproduce** 1. Encode a duplex frame above `DEFAULT_MAX_DUPLEX_FRAME_BYTES`. 2. Send the frame through the gateway or broker. 3. Observe that the old path writes the frame or drops the channel after peer rejection. **Paperclip version or commit** Reproduced from the `master` development line before this change. **Deployment mode** Local dev (`pnpm dev`). ## What Changed - Add `encodeDuplexFrameChecked` to the host and embedded gateway codecs. - Measure encoded JSON bytes without the trailing newline. - Return a typed `frame_too_large` result without throwing. - Return HTTP 413 for oversized gateway requests without writing a frame. - Share one frame bound between broker decode and encode checks. - Return a bounded, non-retryable terminal response for oversized broker responses. - Add encode vectors to the shared wire-compatibility fixture. ## Verification - Run `pnpm --filter @paperclipai/adapter-utils typecheck`. - Run `npx vitest run packages/adapter-utils/src/duplex-frame-codec.test.ts packages/adapter-utils/src/duplex-bridge-broker.test.ts packages/adapter-utils/src/execution-target-sandbox.test.ts`. - Confirm the oversized-response broker test keeps the channel open and serves the other in-flight request. - Confirm the gateway test returns HTTP 413 and keeps the channel open. ## Risks The encoder now rejects oversized frames before transmission. This changes an unsafe write into a typed local error. The broker and gateway keep existing frame limits and affect only oversized frames. ## Model Used OpenAI Codex, GPT-5, tool use and code execution. The model assisted with review and repository operations. ## 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 <noreply@paperclip.ing>
This commit is contained in:
Nicky Leach
GitHub
parent
cc42a67e7e
commit
141b815294
|
|
@ -461,6 +461,232 @@ describe("duplex bridge broker request limits", () => {
|
|||
expect(broker.state).toBe("open");
|
||||
});
|
||||
|
||||
it("substitutes a bounded terminal response for an oversized response and keeps the channel open", async () => {
|
||||
const harness = createFakeChannelHarness();
|
||||
const { telemetry, capture } = createTelemetryCapture();
|
||||
const broker = createDuplexBridgeBroker({
|
||||
channel: harness.channel,
|
||||
forwardRequest: controllableForward(harness),
|
||||
telemetry,
|
||||
// A small frame bound makes a legitimately large response body exceed the
|
||||
// bound, so the test exercises the encode guard without a megabyte body.
|
||||
maxFrameBytes: 1000,
|
||||
});
|
||||
brokers.push(broker);
|
||||
broker.start();
|
||||
|
||||
// Two requests are in flight at the same time. One resolves with a normal
|
||||
// body; the other resolves with a body that pushes the response frame over
|
||||
// the bound. The host produced both results, so both requests reached the
|
||||
// host.
|
||||
harness.feed(requestFrame("small-1", "POST"));
|
||||
harness.feed(requestFrame("big-1", "POST"));
|
||||
|
||||
const smallForward = harness.forwards.find((entry) => entry.id === "small-1" && !entry.settled);
|
||||
if (!smallForward) throw new Error("The broker did not forward the small request.");
|
||||
smallForward.settled = true;
|
||||
smallForward.resolve({
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ ok: true }),
|
||||
});
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
|
||||
const bigForward = harness.forwards.find((entry) => entry.id === "big-1" && !entry.settled);
|
||||
if (!bigForward) throw new Error("The broker did not forward the big request.");
|
||||
bigForward.settled = true;
|
||||
bigForward.resolve({
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ data: "y".repeat(2000) }),
|
||||
});
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
|
||||
// The other in-flight request delivered its full response, unaffected by the
|
||||
// oversized one.
|
||||
expect(harness.responses).toHaveLength(2);
|
||||
const delivered = harness.responses.find((entry) => entry.id === "small-1")!;
|
||||
expect(delivered.status).toBe(200);
|
||||
expect(delivered.outcome).toBe("completed");
|
||||
expect(JSON.parse(delivered.body)).toEqual({ ok: true });
|
||||
|
||||
// The oversized response became one bounded terminal response marked
|
||||
// indeterminate and non-retryable, so a caller does not retry a possible
|
||||
// mutation. The broker did not write the oversized frame.
|
||||
const terminal = harness.responses.find((entry) => entry.id === "big-1")!;
|
||||
expect(terminal.status).toBe(502);
|
||||
expect(terminal.outcome).toBe("indeterminate");
|
||||
expect(terminal.headers["x-paperclip-bridge-outcome"]).toBe("indeterminate");
|
||||
expect(JSON.parse(terminal.body)).toEqual({
|
||||
error: "upstream response too large to deliver",
|
||||
outcome: "indeterminate",
|
||||
retryable: false,
|
||||
});
|
||||
|
||||
// The oversized body never crossed the wire. The broker did not truncate it or
|
||||
// pass it through.
|
||||
const writtenText = harness.responses.map((entry) => encodeDuplexFrame(entry)).join("");
|
||||
expect(writtenText).not.toContain("yyyy");
|
||||
|
||||
// The broker recorded the oversized request as an error outcome, never a loss.
|
||||
// The channel stays open and the run does not fail.
|
||||
const errorSpans = capture.spans.filter((span) => span.dimensions.outcome === "error");
|
||||
expect(errorSpans).toHaveLength(1);
|
||||
expect(errorSpans[0]!.name).toBe(DUPLEX_SPAN_REQUEST);
|
||||
expect(broker.lossRecord).toBeNull();
|
||||
expect(broker.state).toBe("open");
|
||||
expect(broker.runDisposition.failed).toBe(false);
|
||||
});
|
||||
|
||||
it("sends a minimal terminal response when the frame bound rejects the full replacement", async () => {
|
||||
const harness = createFakeChannelHarness();
|
||||
const { telemetry, capture } = createTelemetryCapture();
|
||||
// The bound accepts a compact request (130 bytes) and the minimal terminal
|
||||
// response (106 bytes), but rejects the full replacement response (288 bytes).
|
||||
// This proves the broker never drops the channel or strands the request when
|
||||
// the configured bound is smaller than the full replacement response.
|
||||
const broker = createDuplexBridgeBroker({
|
||||
channel: harness.channel,
|
||||
forwardRequest: controllableForward(harness),
|
||||
telemetry,
|
||||
maxFrameBytes: 200,
|
||||
});
|
||||
brokers.push(broker);
|
||||
broker.start();
|
||||
|
||||
// A compact request frame fits the bound, so the broker forwards it.
|
||||
harness.feed({
|
||||
version: DUPLEX_FRAME_VERSION,
|
||||
type: "request",
|
||||
id: "big-1",
|
||||
method: "POST",
|
||||
path: "/api/issues/big-1",
|
||||
query: "",
|
||||
headers: {},
|
||||
body: JSON.stringify({ go: 1 }),
|
||||
});
|
||||
|
||||
const forward = harness.forwards.find((entry) => entry.id === "big-1" && !entry.settled);
|
||||
if (!forward) throw new Error("The broker did not forward the request.");
|
||||
forward.settled = true;
|
||||
// The host produced a large response body that pushes the real response frame
|
||||
// and the full replacement frame over the bound.
|
||||
forward.resolve({
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ data: "y".repeat(2000) }),
|
||||
});
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
|
||||
// The broker sent one minimal terminal response, not the oversized real
|
||||
// response and not the full replacement. The minimal response carries the
|
||||
// indeterminate outcome, so the gateway maps it to a terminal 409.
|
||||
expect(harness.responses).toHaveLength(1);
|
||||
const terminal = harness.responses[0]!;
|
||||
expect(terminal.id).toBe("big-1");
|
||||
expect(terminal.status).toBe(502);
|
||||
expect(terminal.outcome).toBe("indeterminate");
|
||||
expect(terminal.headers).toEqual({});
|
||||
expect(terminal.body).toBe("");
|
||||
|
||||
// The oversized body never crossed the wire.
|
||||
const writtenText = harness.responses.map((entry) => encodeDuplexFrame(entry)).join("");
|
||||
expect(writtenText).not.toContain("yyyy");
|
||||
|
||||
// The broker recorded the request as an error outcome, never a loss. The
|
||||
// channel stays open and the run does not fail.
|
||||
const errorSpans = capture.spans.filter((span) => span.dimensions.outcome === "error");
|
||||
expect(errorSpans).toHaveLength(1);
|
||||
expect(errorSpans[0]!.name).toBe(DUPLEX_SPAN_REQUEST);
|
||||
expect(broker.lossRecord).toBeNull();
|
||||
expect(broker.state).toBe("open");
|
||||
expect(broker.runDisposition.failed).toBe(false);
|
||||
});
|
||||
|
||||
it("keeps the channel open and logs when the bound rejects even the minimal terminal response", async () => {
|
||||
const harness = createFakeChannelHarness();
|
||||
const { telemetry, capture } = createTelemetryCapture();
|
||||
const logs: string[] = [];
|
||||
// The bound accepts a tiny request (99 bytes) but rejects even the minimal
|
||||
// terminal response (102 bytes). The broker cannot send any frame for this
|
||||
// request, so it must keep the channel open and log a clear local error.
|
||||
const broker = createDuplexBridgeBroker({
|
||||
channel: harness.channel,
|
||||
forwardRequest: controllableForward(harness),
|
||||
telemetry,
|
||||
maxFrameBytes: 100,
|
||||
logger: (message) => logs.push(message),
|
||||
});
|
||||
brokers.push(broker);
|
||||
broker.start();
|
||||
|
||||
harness.feed({
|
||||
version: DUPLEX_FRAME_VERSION,
|
||||
type: "request",
|
||||
id: "x",
|
||||
method: "GET",
|
||||
path: "/",
|
||||
query: "",
|
||||
headers: {},
|
||||
body: "",
|
||||
});
|
||||
|
||||
const forward = harness.forwards.find((entry) => entry.id === "x" && !entry.settled);
|
||||
if (!forward) throw new Error("The broker did not forward the request.");
|
||||
forward.settled = true;
|
||||
forward.resolve({
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ data: "y".repeat(200) }),
|
||||
});
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
|
||||
// The broker sent no response frame, because no frame fits the bound.
|
||||
expect(harness.responses).toHaveLength(0);
|
||||
// The broker recorded the request as an error outcome, never a loss.
|
||||
const errorSpans = capture.spans.filter((span) => span.dimensions.outcome === "error");
|
||||
expect(errorSpans).toHaveLength(1);
|
||||
// The broker logged a clear local error and kept the channel open.
|
||||
expect(logs.some((line) => line.includes("could not deliver a terminal response"))).toBe(true);
|
||||
expect(broker.lossRecord).toBeNull();
|
||||
expect(broker.state).toBe("open");
|
||||
expect(broker.runDisposition.failed).toBe(false);
|
||||
});
|
||||
|
||||
it("completes a request within the bound and delivers the real response", async () => {
|
||||
const harness = createFakeChannelHarness();
|
||||
const { telemetry, capture } = createTelemetryCapture();
|
||||
const broker = createDuplexBridgeBroker({
|
||||
channel: harness.channel,
|
||||
forwardRequest: controllableForward(harness),
|
||||
telemetry,
|
||||
maxFrameBytes: 1000,
|
||||
});
|
||||
brokers.push(broker);
|
||||
broker.start();
|
||||
|
||||
harness.feed(requestFrame("ok-1", "POST"));
|
||||
harness.resolveForward("ok-1", JSON.stringify({ ok: true }));
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
|
||||
// The broker delivered the real response promptly with the completed outcome.
|
||||
expect(harness.responses).toHaveLength(1);
|
||||
const delivered = harness.responses[0]!;
|
||||
expect(delivered.id).toBe("ok-1");
|
||||
expect(delivered.status).toBe(200);
|
||||
expect(delivered.outcome).toBe("completed");
|
||||
expect(JSON.parse(delivered.body)).toEqual({ ok: true });
|
||||
|
||||
// The delivered response records an `ok` span, never an error, and never a loss.
|
||||
const okSpans = capture.spans.filter((span) => span.dimensions.outcome === "ok");
|
||||
expect(okSpans).toHaveLength(1);
|
||||
const errorSpans = capture.spans.filter((span) => span.dimensions.outcome === "error");
|
||||
expect(errorSpans).toHaveLength(0);
|
||||
expect(broker.lossRecord).toBeNull();
|
||||
expect(broker.state).toBe("open");
|
||||
expect(broker.runDisposition.failed).toBe(false);
|
||||
});
|
||||
|
||||
it("maps a forward rejection for a safe method to a retryable response", async () => {
|
||||
const harness = createFakeChannelHarness();
|
||||
const broker = createDuplexBridgeBroker({
|
||||
|
|
|
|||
|
|
@ -43,10 +43,12 @@
|
|||
|
||||
import type { CommandManagedDuplexChannel } from "./command-managed-runtime.js";
|
||||
import {
|
||||
DEFAULT_MAX_DUPLEX_FRAME_BYTES,
|
||||
DEFAULT_MAX_DUPLEX_REQUEST_ID_BYTES,
|
||||
DUPLEX_FRAME_VERSION,
|
||||
DuplexFrameDecoder,
|
||||
encodeDuplexFrame,
|
||||
encodeDuplexFrameChecked,
|
||||
type DuplexFrame,
|
||||
type DuplexRequestFrame,
|
||||
type DuplexResponseFrame,
|
||||
|
|
@ -376,9 +378,12 @@ export function createDuplexBridgeBroker(options: DuplexBrokerOptions): DuplexBr
|
|||
const heartbeatIntervalMs = options.heartbeatIntervalMs ?? DEFAULT_DUPLEX_BROKER_HEARTBEAT_INTERVAL_MS;
|
||||
const closeTimeoutMs = options.closeTimeoutMs ?? DEFAULT_DUPLEX_BROKER_CLOSE_TIMEOUT_MS;
|
||||
const now = options.now ?? (() => Date.now());
|
||||
const decoder = new DuplexFrameDecoder(
|
||||
options.maxFrameBytes ? { maxFrameBytes: options.maxFrameBytes } : undefined,
|
||||
);
|
||||
// The one frame size bound the broker enforces on both sides. The decoder
|
||||
// rejects an inbound frame over this bound, and the encode guard refuses to
|
||||
// write an outbound frame over it. Encode and decode share one value, so a
|
||||
// frame the broker writes always decodes on the peer.
|
||||
const maxFrameBytes = options.maxFrameBytes ?? DEFAULT_MAX_DUPLEX_FRAME_BYTES;
|
||||
const decoder = new DuplexFrameDecoder({ maxFrameBytes });
|
||||
|
||||
let state: DuplexBrokerState = "opening";
|
||||
let stopped = false;
|
||||
|
|
@ -491,9 +496,9 @@ export function createDuplexBridgeBroker(options: DuplexBrokerOptions): DuplexBr
|
|||
options.telemetry?.recordLoss(lossClass, BROKER_LOSS_REASON_TO_TYPED[reason] ?? "other");
|
||||
};
|
||||
|
||||
const writeFrame = (frame: DuplexFrame): boolean => {
|
||||
const writeLine = (line: string): boolean => {
|
||||
try {
|
||||
channel.write(encodeDuplexFrame(frame));
|
||||
channel.write(line);
|
||||
return true;
|
||||
} catch (error) {
|
||||
recordLoss("stream_failure", errorMessage(error));
|
||||
|
|
@ -501,6 +506,82 @@ export function createDuplexBridgeBroker(options: DuplexBrokerOptions): DuplexBr
|
|||
}
|
||||
};
|
||||
|
||||
// The result of one bounded send. `sent` means the frame went out; `too_large`
|
||||
// means the encoded frame exceeds the bound and the broker wrote nothing;
|
||||
// `lost` means the write failed and the broker recorded the channel loss.
|
||||
type SendResult = "sent" | "too_large" | "lost";
|
||||
|
||||
const trySendFrame = (frame: DuplexFrame): SendResult => {
|
||||
// Enforce the frame size bound on every broker write. Report a size rejection
|
||||
// as `too_large` without a channel loss; a broker-built frame over the bound
|
||||
// is a defect, not a transport failure. A caller decides how to answer the
|
||||
// request, so no size rejection ever drops the channel.
|
||||
const encoded = encodeDuplexFrameChecked(frame, maxFrameBytes);
|
||||
if (!encoded.ok) return "too_large";
|
||||
return writeLine(encoded.line) ? "sent" : "lost";
|
||||
};
|
||||
|
||||
const writeFrame = (frame: DuplexFrame): boolean => {
|
||||
// A control frame and the bounded terminal responses are always small, so the
|
||||
// size guard is a no-op for them. Drop an oversized frame here without a
|
||||
// channel loss and report the drop to the caller.
|
||||
const result = trySendFrame(frame);
|
||||
if (result === "too_large") {
|
||||
options.logger?.(`Duplex broker dropped an oversized ${frame.type} frame.`);
|
||||
return false;
|
||||
}
|
||||
return result === "sent";
|
||||
};
|
||||
|
||||
const sendTerminalIndeterminate = (id: string): void => {
|
||||
// Answer one request the broker cannot deliver with the real result. The
|
||||
// request reached the host and may have changed state, so the response is
|
||||
// non-retryable and carries the `indeterminate` outcome. The broker tries the
|
||||
// full replacement first. When the frame bound rejects even the full
|
||||
// replacement, the broker sends a minimal replacement that carries only the
|
||||
// `indeterminate` outcome, so the gateway still ends the request and never
|
||||
// waits for its full wait budget. When the bound rejects even the minimal
|
||||
// replacement, the broker logs a clear local error and keeps the channel
|
||||
// open; it records no channel loss.
|
||||
const fullReplacement: DuplexResponseFrame = {
|
||||
version: DUPLEX_FRAME_VERSION,
|
||||
type: "response",
|
||||
id,
|
||||
status: 502,
|
||||
headers: {
|
||||
"content-type": "application/json",
|
||||
"x-paperclip-bridge-outcome": "indeterminate",
|
||||
},
|
||||
body: JSON.stringify({
|
||||
error: "upstream response too large to deliver",
|
||||
outcome: "indeterminate",
|
||||
retryable: false,
|
||||
}),
|
||||
outcome: "indeterminate",
|
||||
};
|
||||
if (trySendFrame(fullReplacement) !== "too_large") return;
|
||||
// The bound rejects the full replacement. Send a minimal terminal response
|
||||
// with empty headers and an empty body. The `indeterminate` outcome still
|
||||
// rides the frame, so the gateway maps the request to a terminal 409.
|
||||
const minimalReplacement: DuplexResponseFrame = {
|
||||
version: DUPLEX_FRAME_VERSION,
|
||||
type: "response",
|
||||
id,
|
||||
status: 502,
|
||||
headers: {},
|
||||
body: "",
|
||||
outcome: "indeterminate",
|
||||
};
|
||||
if (trySendFrame(minimalReplacement) !== "too_large") return;
|
||||
// The bound rejects even the minimal terminal response. The broker cannot
|
||||
// deliver any frame for this request within the bound. Log a clear local
|
||||
// error and keep the channel open for every other request. The gateway ends
|
||||
// its own outstanding request on its wait budget.
|
||||
options.logger?.(
|
||||
`Duplex broker could not deliver a terminal response within the ${maxFrameBytes}-byte frame bound.`,
|
||||
);
|
||||
};
|
||||
|
||||
const respond = (
|
||||
id: string,
|
||||
result: DuplexBrokerForwardResult,
|
||||
|
|
@ -516,12 +597,6 @@ export function createDuplexBridgeBroker(options: DuplexBrokerOptions): DuplexBr
|
|||
// Do not write on a lost or closed channel. The gateway answers its own
|
||||
// outstanding request on loss, so a late write would go to a dead channel.
|
||||
if (state !== "open") return;
|
||||
// Record the request span for the delivered request. The span carries the
|
||||
// latency and the outcome only; no route, query, body, or token rides it.
|
||||
options.telemetry?.recordRequest({
|
||||
latencyMs: now() - entry.dispatchStartMs,
|
||||
outcome: telemetryOutcome,
|
||||
});
|
||||
const frame: DuplexResponseFrame = {
|
||||
version: DUPLEX_FRAME_VERSION,
|
||||
type: "response",
|
||||
|
|
@ -531,7 +606,28 @@ export function createDuplexBridgeBroker(options: DuplexBrokerOptions): DuplexBr
|
|||
body: result.body ?? "",
|
||||
outcome,
|
||||
};
|
||||
writeFrame(frame);
|
||||
const encoded = encodeDuplexFrameChecked(frame, maxFrameBytes);
|
||||
if (!encoded.ok) {
|
||||
// The host produced a result, but the response frame exceeds the size
|
||||
// bound. Do not write the oversized frame and do not record a channel loss.
|
||||
// The request reached the host and may have changed state, so the caller
|
||||
// must not retry a possible mutation. Send a bounded, non-retryable terminal
|
||||
// response marked `indeterminate` in its place. Record the request outcome
|
||||
// as an error, never a loss. The channel stays open for every other request.
|
||||
options.telemetry?.recordRequest({
|
||||
latencyMs: now() - entry.dispatchStartMs,
|
||||
outcome: "error",
|
||||
});
|
||||
sendTerminalIndeterminate(id);
|
||||
return;
|
||||
}
|
||||
// Record the request span for the delivered request. The span carries the
|
||||
// latency and the outcome only; no route, query, body, or token rides it.
|
||||
options.telemetry?.recordRequest({
|
||||
latencyMs: now() - entry.dispatchStartMs,
|
||||
outcome: telemetryOutcome,
|
||||
});
|
||||
writeLine(encoded.line);
|
||||
};
|
||||
|
||||
const respondSaturated = (id: string, retryable: boolean): void => {
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ import {
|
|||
DuplexFrameDecoder,
|
||||
decodeDuplexLine,
|
||||
encodeDuplexFrame,
|
||||
encodeDuplexFrameChecked,
|
||||
type DuplexDecodeResult,
|
||||
type DuplexFrame,
|
||||
} from "./duplex-frame-codec.js";
|
||||
|
|
@ -33,10 +34,20 @@ interface Vector {
|
|||
expected: ExpectedResult[];
|
||||
}
|
||||
|
||||
// One encode-bound vector: a frame, the size limit, and the expected checked
|
||||
// encode result. An ok result must also decode back to the same frame.
|
||||
interface EncodeVector {
|
||||
name: string;
|
||||
maxFrameBytes: number;
|
||||
frame: DuplexFrame;
|
||||
expected: { ok: true } | { ok: false; error: string };
|
||||
}
|
||||
|
||||
interface Fixture {
|
||||
frameVersion: number;
|
||||
defaultMaxFrameBytes: number;
|
||||
vectors: Vector[];
|
||||
encodeVectors: EncodeVector[];
|
||||
}
|
||||
|
||||
const fixturePath = fileURLToPath(
|
||||
|
|
@ -310,3 +321,72 @@ describe("request id byte bound", () => {
|
|||
if (!result.ok) expect(result.error.code).toBe("id_too_large");
|
||||
});
|
||||
});
|
||||
|
||||
describe("size-checked encode", () => {
|
||||
it("runs every shared encode vector and matches the expected result", () => {
|
||||
// Every codec copy runs the same encode vectors. This copy proves it enforces
|
||||
// the same bound the embedded gateway copy enforces.
|
||||
for (const vector of fixture.encodeVectors) {
|
||||
const result = encodeDuplexFrameChecked(vector.frame, vector.maxFrameBytes);
|
||||
expect(result.ok).toBe(vector.expected.ok);
|
||||
if (result.ok) {
|
||||
// An ok line ends with one newline and decodes back to the same frame, so
|
||||
// the encode guard never truncates or passes a bad frame through.
|
||||
expect(result.line.endsWith("\n")).toBe(true);
|
||||
expect(result.line.slice(0, -1)).not.toContain("\n");
|
||||
const decoded = decodeDuplexLine(result.line.slice(0, -1));
|
||||
expect(decoded.ok).toBe(true);
|
||||
if (decoded.ok) expect(decoded.frame).toEqual(vector.frame);
|
||||
} else if (!vector.expected.ok) {
|
||||
expect(result.error.code).toBe(vector.expected.error);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
it("rejects an over-bound frame with a typed outcome and never throws", () => {
|
||||
const frame: DuplexFrame = {
|
||||
version: DUPLEX_FRAME_VERSION,
|
||||
type: "response",
|
||||
id: "big",
|
||||
status: 200,
|
||||
headers: {},
|
||||
body: "x".repeat(2_000),
|
||||
outcome: "completed",
|
||||
};
|
||||
expect(() => encodeDuplexFrameChecked(frame, 1_000)).not.toThrow();
|
||||
const result = encodeDuplexFrameChecked(frame, 1_000);
|
||||
expect(result.ok).toBe(false);
|
||||
if (!result.ok) expect(result.error.code).toBe("frame_too_large");
|
||||
});
|
||||
|
||||
it("measures the bound in bytes without the trailing newline, so the boundary is inclusive", () => {
|
||||
// Build a frame whose encoded JSON is exactly N bytes. The guard measures the
|
||||
// JSON without the newline, so N bytes encodes and N+1 bytes fails. This is the
|
||||
// same boundary the decoder applies, so encode and decode agree exactly.
|
||||
const base: DuplexFrame = {
|
||||
version: DUPLEX_FRAME_VERSION,
|
||||
type: "response",
|
||||
id: "boundary",
|
||||
status: 200,
|
||||
headers: {},
|
||||
body: "",
|
||||
outcome: "completed",
|
||||
};
|
||||
const baseBytes = Buffer.byteLength(JSON.stringify(base), "utf8");
|
||||
const limit = baseBytes + 10;
|
||||
const atLimit: DuplexFrame = { ...base, body: "x".repeat(10) };
|
||||
expect(Buffer.byteLength(JSON.stringify(atLimit), "utf8")).toBe(limit);
|
||||
const okResult = encodeDuplexFrameChecked(atLimit, limit);
|
||||
expect(okResult.ok).toBe(true);
|
||||
const overResult = encodeDuplexFrameChecked({ ...base, body: "x".repeat(11) }, limit);
|
||||
expect(overResult.ok).toBe(false);
|
||||
if (!overResult.ok) expect(overResult.error.code).toBe("frame_too_large");
|
||||
});
|
||||
|
||||
it("defaults the bound to the default max frame bytes", () => {
|
||||
const frame: DuplexFrame = { version: DUPLEX_FRAME_VERSION, type: "heartbeat" };
|
||||
const result = encodeDuplexFrameChecked(frame);
|
||||
expect(result.ok).toBe(true);
|
||||
if (result.ok) expect(result.line).toBe(encodeDuplexFrame(frame));
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -147,6 +147,15 @@ export type DuplexDecodeResult =
|
|||
| { ok: true; frame: DuplexFrame }
|
||||
| { ok: false; error: DuplexProtocolError };
|
||||
|
||||
/**
|
||||
* One size-checked encode result: one line, or a `frame_too_large` error. The
|
||||
* shape mirrors {@link DuplexDecodeResult}, so the encode side reports the
|
||||
* over-limit case as a typed outcome, not an exception.
|
||||
*/
|
||||
export type DuplexEncodeResult =
|
||||
| { ok: true; line: string }
|
||||
| { ok: false; error: { code: "frame_too_large"; message: string } };
|
||||
|
||||
const RESPONSE_OUTCOMES: ReadonlySet<string> = new Set<DuplexResponseOutcome>([
|
||||
"completed",
|
||||
"indeterminate",
|
||||
|
|
@ -187,6 +196,30 @@ export function encodeDuplexFrame(frame: DuplexFrame): string {
|
|||
return `${JSON.stringify(frame)}\n`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Encode one frame to a single line and enforce the maximum frame size. The
|
||||
* function measures the encoded JSON in bytes, without the trailing newline, so
|
||||
* it matches the decoder bound exactly: a line the decoder accepts encodes, and a
|
||||
* line the decoder rejects returns a `frame_too_large` result. The function never
|
||||
* throws; it reports the over-limit case as a typed outcome.
|
||||
*
|
||||
* The size bound follows request and response bodies the host does not control,
|
||||
* so an over-limit frame is an expected condition, not a programming error. Every
|
||||
* write path that can carry a large body uses this function, so no path emits a
|
||||
* frame the peer decoder rejects. The bound applies to every frame type; the
|
||||
* guard is a no-op for a small control frame.
|
||||
*/
|
||||
export function encodeDuplexFrameChecked(
|
||||
frame: DuplexFrame,
|
||||
maxFrameBytes: number = DEFAULT_MAX_DUPLEX_FRAME_BYTES,
|
||||
): DuplexEncodeResult {
|
||||
const json = JSON.stringify(frame);
|
||||
if (Buffer.byteLength(json, "utf8") > maxFrameBytes) {
|
||||
return { ok: false, error: { code: "frame_too_large", message: "frame exceeds the maximum size" } };
|
||||
}
|
||||
return { ok: true, line: `${json}\n` };
|
||||
}
|
||||
|
||||
/**
|
||||
* Decode one line (no trailing newline) to a frame or a protocol error. The
|
||||
* streaming decoder calls this for each complete line. It is exported so a
|
||||
|
|
|
|||
|
|
@ -414,5 +414,55 @@
|
|||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"encodeDescription": "Encode-bound vectors. Each codec copy runs the size-checked encode against the frame with the maxFrameBytes limit, then asserts the same result. An ok result must decode back to the same frame. This proves both copies enforce the same bound on encode.",
|
||||
"encodeVectors": [
|
||||
{
|
||||
"name": "encode-within-bound",
|
||||
"maxFrameBytes": 200,
|
||||
"frame": {
|
||||
"version": 1,
|
||||
"type": "request",
|
||||
"id": "req-enc-1",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"query": "",
|
||||
"headers": {},
|
||||
"body": "small"
|
||||
},
|
||||
"expected": {
|
||||
"ok": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "encode-control-frame-never-rejected",
|
||||
"maxFrameBytes": 200,
|
||||
"frame": {
|
||||
"version": 1,
|
||||
"type": "heartbeat"
|
||||
},
|
||||
"expected": {
|
||||
"ok": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "encode-over-bound",
|
||||
"maxFrameBytes": 200,
|
||||
"frame": {
|
||||
"version": 1,
|
||||
"type": "response",
|
||||
"id": "req-enc-2",
|
||||
"status": 200,
|
||||
"headers": {
|
||||
"content-type": "application/json"
|
||||
},
|
||||
"body": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
|
||||
"outcome": "completed"
|
||||
},
|
||||
"expected": {
|
||||
"ok": false,
|
||||
"error": "frame_too_large"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4842,8 +4842,13 @@ interface EmbeddedDecodeResult {
|
|||
|
||||
// The names the embedded codec source declares. A test wraps the source and
|
||||
// reads these names back.
|
||||
type EmbeddedEncodeResult =
|
||||
| { ok: true; line: string }
|
||||
| { ok: false; error: { code: string; message: string } };
|
||||
|
||||
interface EmbeddedCodec {
|
||||
encodeDuplexFrame: (frame: unknown) => string;
|
||||
encodeDuplexFrameChecked: (frame: unknown, maxFrameBytes?: number) => EmbeddedEncodeResult;
|
||||
decodeDuplexLine: (line: string | Buffer) => EmbeddedDecodeResult;
|
||||
DuplexFrameDecoder: new (options?: { maxFrameBytes?: number }) => {
|
||||
push: (chunk: Buffer) => EmbeddedDecodeResult[];
|
||||
|
|
@ -4864,10 +4869,18 @@ interface DuplexFrameVector {
|
|||
expected: ExpectedVectorResult[];
|
||||
}
|
||||
|
||||
interface DuplexEncodeVector {
|
||||
name: string;
|
||||
maxFrameBytes: number;
|
||||
frame: unknown;
|
||||
expected: { ok: true } | { ok: false; error: string };
|
||||
}
|
||||
|
||||
interface DuplexFrameFixture {
|
||||
frameVersion: number;
|
||||
defaultMaxFrameBytes: number;
|
||||
vectors: DuplexFrameVector[];
|
||||
encodeVectors: DuplexEncodeVector[];
|
||||
}
|
||||
|
||||
describe("sandbox duplex gateway", () => {
|
||||
|
|
@ -5086,7 +5099,7 @@ describe("sandbox duplex gateway", () => {
|
|||
|
||||
it("embedded gateway codec passes every vector in the shared fixture", async () => {
|
||||
const codecFactory = new Function(
|
||||
`${getSandboxDuplexGatewayCodecSource()}\nreturn { encodeDuplexFrame, decodeDuplexLine, DuplexFrameDecoder, DUPLEX_FRAME_VERSION, DEFAULT_MAX_DUPLEX_FRAME_BYTES };`,
|
||||
`${getSandboxDuplexGatewayCodecSource()}\nreturn { encodeDuplexFrame, encodeDuplexFrameChecked, decodeDuplexLine, DuplexFrameDecoder, DUPLEX_FRAME_VERSION, DEFAULT_MAX_DUPLEX_FRAME_BYTES };`,
|
||||
) as unknown as () => EmbeddedCodec;
|
||||
const codec = codecFactory();
|
||||
|
||||
|
|
@ -5152,6 +5165,38 @@ describe("sandbox duplex gateway", () => {
|
|||
expect(decoded.ok).toBe(true);
|
||||
expect(decoded.frame).toEqual(want.frame);
|
||||
}
|
||||
|
||||
// The embedded copy enforces the same encode bound as the host copy. It runs
|
||||
// every shared encode vector and matches the expected result, so both copies
|
||||
// reject the same oversized frame with the same `frame_too_large` code.
|
||||
expect(fixture.encodeVectors.length).toBeGreaterThanOrEqual(3);
|
||||
const encodeFailures: string[] = [];
|
||||
for (const vector of fixture.encodeVectors) {
|
||||
const result = codec.encodeDuplexFrameChecked(vector.frame, vector.maxFrameBytes);
|
||||
if (result.ok !== vector.expected.ok) {
|
||||
encodeFailures.push(`${vector.name}: ok=${result.ok}, want ${vector.expected.ok}`);
|
||||
continue;
|
||||
}
|
||||
if (result.ok) {
|
||||
if (!result.line.endsWith("\n") || result.line.slice(0, -1).includes("\n")) {
|
||||
encodeFailures.push(`${vector.name}: line is not exactly one frame line`);
|
||||
continue;
|
||||
}
|
||||
const decoded = codec.decodeDuplexLine(result.line.slice(0, -1));
|
||||
if (!decoded.ok) {
|
||||
encodeFailures.push(`${vector.name}: an ok encode did not decode back`);
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
expect(decoded.frame).toEqual(vector.frame);
|
||||
} catch {
|
||||
encodeFailures.push(`${vector.name}: decoded frame does not match`);
|
||||
}
|
||||
} else if (!vector.expected.ok && result.error.code !== vector.expected.error) {
|
||||
encodeFailures.push(`${vector.name}: error ${result.error.code} != ${vector.expected.error}`);
|
||||
}
|
||||
}
|
||||
expect(encodeFailures).toEqual([]);
|
||||
});
|
||||
|
||||
it("returns the same HTTP response as the file gateway for a forwarded request", async () => {
|
||||
|
|
@ -5216,6 +5261,55 @@ describe("sandbox duplex gateway", () => {
|
|||
await gateway.stop();
|
||||
}, 20000);
|
||||
|
||||
it("fails a request that exceeds the frame size bound with a local 413 and keeps the channel open", async () => {
|
||||
const token = "duplex-token-oversize-request";
|
||||
// Raise the body limit above the frame bound, so `readBody` accepts the body
|
||||
// and the encode guard is the only limit the request meets. The default frame
|
||||
// bound is 1,000,000 bytes.
|
||||
const gateway = await startDuplexGateway({
|
||||
PAPERCLIP_BRIDGE_TOKEN: token,
|
||||
PAPERCLIP_BRIDGE_MAX_BODY_BYTES: "3000000",
|
||||
});
|
||||
|
||||
// A body over the frame bound makes the request frame exceed the bound. The
|
||||
// gateway must fail this one local request with a clean 413 and forward no
|
||||
// frame.
|
||||
const oversizeBody = JSON.stringify({ body: "x".repeat(1_100_000) });
|
||||
const tooLarge = await fetch(`${gateway.baseUrl}/api/issues/issue-1/comments`, {
|
||||
method: "POST",
|
||||
headers: { authorization: `Bearer ${token}`, "content-type": "application/json" },
|
||||
body: oversizeBody,
|
||||
});
|
||||
expect(tooLarge.status).toBe(413);
|
||||
await expect(tooLarge.json()).resolves.toEqual({ error: "request_too_large" });
|
||||
|
||||
// The oversized request forwarded no frame. It never left the gateway.
|
||||
expect(gateway.frames.filter((frame) => frame.type === "request")).toHaveLength(0);
|
||||
// The gateway did not lose the channel: no close frame and the process runs.
|
||||
expect(gateway.frames.some((frame) => frame.type === "close")).toBe(false);
|
||||
|
||||
// The channel stays open, so a normal request after the oversized one still
|
||||
// forwards and completes.
|
||||
const okPromise = fetch(`${gateway.baseUrl}/api/agents/me`, {
|
||||
headers: { authorization: `Bearer ${token}` },
|
||||
});
|
||||
const requestFrame = await gateway.waitForFrame((frame) => frame.type === "request");
|
||||
gateway.sendFrame({
|
||||
version: 1,
|
||||
type: "response",
|
||||
id: requestFrame.id,
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ ok: true }),
|
||||
outcome: "completed",
|
||||
});
|
||||
const okResponse = await okPromise;
|
||||
expect(okResponse.status).toBe(200);
|
||||
await expect(okResponse.json()).resolves.toEqual({ ok: true });
|
||||
|
||||
await gateway.stop();
|
||||
}, 20000);
|
||||
|
||||
it("enforces the bearer check, JSON-only rule, body limit, and depth limit", async () => {
|
||||
const token = "duplex-token-contract";
|
||||
const gateway = await startDuplexGateway({
|
||||
|
|
|
|||
|
|
@ -1774,6 +1774,15 @@ function encodeDuplexFrame(frame) {
|
|||
return JSON.stringify(frame) + "\\n";
|
||||
}
|
||||
|
||||
function encodeDuplexFrameChecked(frame, maxFrameBytes) {
|
||||
const limit = maxFrameBytes != null ? maxFrameBytes : DEFAULT_MAX_DUPLEX_FRAME_BYTES;
|
||||
const json = JSON.stringify(frame);
|
||||
if (Buffer.byteLength(json, "utf8") > limit) {
|
||||
return { ok: false, error: { code: "frame_too_large", message: "frame exceeds the maximum size" } };
|
||||
}
|
||||
return { ok: true, line: json + "\\n" };
|
||||
}
|
||||
|
||||
function decodeDuplexLine(line) {
|
||||
const text = typeof line === "string" ? line : line.toString("utf8");
|
||||
let parsed;
|
||||
|
|
@ -1915,9 +1924,10 @@ class DuplexFrameDecoder {
|
|||
* Return the exact zero-dependency codec source the generated duplex gateway
|
||||
* embeds. A test runs every fixture vector against this source, so it proves the
|
||||
* embedded copy decodes the same bytes as the host codec. The source declares
|
||||
* `encodeDuplexFrame`, `decodeDuplexLine`, `DuplexFrameDecoder`,
|
||||
* `DUPLEX_FRAME_VERSION`, and `DEFAULT_MAX_DUPLEX_FRAME_BYTES`, but exports none
|
||||
* of them; a caller wraps it to read those names.
|
||||
* `encodeDuplexFrame`, `encodeDuplexFrameChecked`, `decodeDuplexLine`,
|
||||
* `DuplexFrameDecoder`, `DUPLEX_FRAME_VERSION`, and
|
||||
* `DEFAULT_MAX_DUPLEX_FRAME_BYTES`, but exports none of them; a caller wraps it to
|
||||
* read those names.
|
||||
*/
|
||||
export function getSandboxDuplexGatewayCodecSource(): string {
|
||||
return DUPLEX_GATEWAY_CODEC_SOURCE;
|
||||
|
|
@ -2273,6 +2283,16 @@ function runDuplexGateway() {
|
|||
headers: normalizeHeaders(req.headers),
|
||||
body: requestBody,
|
||||
};
|
||||
// Enforce the frame size bound on encode. When the request body makes the
|
||||
// frame exceed the bound, fail this one local request with a clean 413. Do
|
||||
// not write the frame and do not trigger loss. The frame never leaves the
|
||||
// gateway, so no other in-flight request is affected and the channel stays
|
||||
// open.
|
||||
const encodedRequest = encodeDuplexFrameChecked(requestFrame);
|
||||
if (!encodedRequest.ok) {
|
||||
writeJsonResponse(res, 413, { error: "request_too_large" });
|
||||
return;
|
||||
}
|
||||
const response = await new Promise((resolve) => {
|
||||
const timer = setTimeout(() => {
|
||||
if (pending.delete(requestId)) {
|
||||
|
|
@ -2284,7 +2304,7 @@ function runDuplexGateway() {
|
|||
}
|
||||
}, responseTimeoutMs);
|
||||
pending.set(requestId, { resolve: resolve, timer: timer });
|
||||
writeFrame(requestFrame);
|
||||
process.stdout.write(encodedRequest.line);
|
||||
});
|
||||
res.statusCode = typeof response.status === "number" ? response.status : 200;
|
||||
for (const [key, value] of Object.entries(response.headers || {})) {
|
||||
|
|
|
|||
Loading…
Reference in New Issue