Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion src/lib/bounded-body.ts
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@ export interface BoundedBodyOptions {
totalTimeoutMs?: number;
/** Deadline between non-empty raw chunks. Exposed for focused tests. */
inactivityTimeoutMs?: number;
/** Deadline for the first non-empty raw chunk. Defaults to inactivityTimeoutMs. */
firstByteTimeoutMs?: number;
}

export interface BoundedBodyResult {
Expand Down Expand Up @@ -129,7 +131,7 @@ export async function readBoundedResponseBody(
let cancelReason: unknown;
const total = timeoutPromise(options.totalTimeoutMs ?? BOUNDED_BODY_TIMEOUT_MS, TOTAL_TIMEOUT);
let inactivity = timeoutPromise(
options.inactivityTimeoutMs ?? BOUNDED_BODY_TIMEOUT_MS,
options.firstByteTimeoutMs ?? options.inactivityTimeoutMs ?? BOUNDED_BODY_TIMEOUT_MS,
INACTIVITY_TIMEOUT,
);

Expand Down
16 changes: 10 additions & 6 deletions src/server/responses/core.ts
Original file line number Diff line number Diff line change
Expand Up @@ -747,10 +747,18 @@ const UNREADABLE_ENCRYPTED_AGENT_TASK_MESSAGE =
// branch of the passthrough return path). 32 MiB matches the continuation snapshot read
// bound and is far above any legitimate non-streaming completion, including base64 image
// payloads. The stall deadlines only govern the body transfer — generation time before
// the response headers is untouched.
// the response headers is untouched. Generation after early/chunked headers but before
// the first body byte previously used the 30-second inactivity deadline; this call site
// gives it the full body deadline instead.
const MAX_UPSTREAM_JSON_BODY_BYTES = 32 * 1024 * 1024;
const UPSTREAM_JSON_BODY_TOTAL_TIMEOUT_MS = 180_000;
const UPSTREAM_JSON_BODY_INACTIVITY_TIMEOUT_MS = 30_000;
export const UPSTREAM_JSON_BODY_READ_OPTIONS = {
maxBytes: MAX_UPSTREAM_JSON_BODY_BYTES,
totalTimeoutMs: UPSTREAM_JSON_BODY_TOTAL_TIMEOUT_MS,
inactivityTimeoutMs: UPSTREAM_JSON_BODY_INACTIVITY_TIMEOUT_MS,
firstByteTimeoutMs: UPSTREAM_JSON_BODY_TOTAL_TIMEOUT_MS,
};

function unreadableEncryptedAgentTaskResponse(): Response {
return new Response(
Expand Down Expand Up @@ -2226,11 +2234,7 @@ async function handleResponsesInner(
// without limit. This path is no longer rare — WebSocket turns for models whose
// streaming terminal event is unreliable are deliberately answered with bounded JSON.
// Oversize and stall deadlines both fail closed; a partial body is never parsed.
const bounded = await readBoundedResponseBody(upstreamResponse, {
maxBytes: MAX_UPSTREAM_JSON_BODY_BYTES,
totalTimeoutMs: UPSTREAM_JSON_BODY_TOTAL_TIMEOUT_MS,
inactivityTimeoutMs: UPSTREAM_JSON_BODY_INACTIVITY_TIMEOUT_MS,
});
const bounded = await readBoundedResponseBody(upstreamResponse, UPSTREAM_JSON_BODY_READ_OPTIONS);
if (bounded.oversized) {
return formatErrorResponse(502, "upstream_error", "upstream JSON response exceeded the safe body limit");
}
Expand Down
83 changes: 83 additions & 0 deletions tests/bounded-body.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import {
boundedBodyBufferGrowthsForTests,
readBoundedResponseBody,
} from "../src/lib/bounded-body";
import { UPSTREAM_JSON_BODY_READ_OPTIONS } from "../src/server/responses/core";

const encoder = new TextEncoder();

Expand All @@ -18,6 +19,11 @@ function responseFromChunks(...chunks: Uint8Array[]): Response {
}

describe("readBoundedResponseBody", () => {
test("the bounded JSON caller allows a full total deadline for its first byte", () => {
expect(UPSTREAM_JSON_BODY_READ_OPTIONS.firstByteTimeoutMs)
.toBe(UPSTREAM_JSON_BODY_READ_OPTIONS.totalTimeoutMs);
});

test("reads multiple chunks and flushes split UTF-8", async () => {
const bytes = encoder.encode("alpha 한글 🌍");
const response = responseFromChunks(bytes.subarray(0, 8), bytes.subarray(8, 11), bytes.subarray(11));
Expand Down Expand Up @@ -52,6 +58,83 @@ describe("readBoundedResponseBody", () => {
expect(cancelled).toBe(true);
});

test("allows the first byte to arrive after the inter-chunk inactivity deadline", async () => {
const response = new Response(new ReadableStream<Uint8Array>({
start(controller) {
setTimeout(() => {
controller.enqueue(encoder.encode("first byte"));
controller.close();
}, 30);
},
}));

const result = await readBoundedResponseBody(response, {
totalTimeoutMs: 100,
inactivityTimeoutMs: 15,
firstByteTimeoutMs: 60,
});

expect(result.text).toBe("first byte");
expect(result.truncated).toBe(false);
expect(result.inactivityTimedOut).toBe(false);
});

test("times out when the first byte misses its dedicated deadline", async () => {
const response = new Response(new ReadableStream<Uint8Array>({}));

const result = await readBoundedResponseBody(response, {
totalTimeoutMs: 100,
inactivityTimeoutMs: 60,
firstByteTimeoutMs: 15,
});

expect(result.inactivityTimedOut).toBe(true);
expect(result.totalTimedOut).toBe(false);
});

test("keeps the inter-chunk inactivity deadline after the first byte", async () => {
let timer: ReturnType<typeof setTimeout> | undefined;
const response = new Response(new ReadableStream<Uint8Array>({
start(controller) {
controller.enqueue(encoder.encode("first"));
timer = setTimeout(() => controller.enqueue(encoder.encode("late")), 30);
},
cancel() {
if (timer) clearTimeout(timer);
},
}));

const result = await readBoundedResponseBody(response, {
totalTimeoutMs: 100,
inactivityTimeoutMs: 15,
firstByteTimeoutMs: 60,
});

expect(result.text).toBe("first");
expect(result.truncated).toBe(true);
expect(result.inactivityTimedOut).toBe(true);
});

test("uses the inactivity deadline for the first byte by default", async () => {
let timer: ReturnType<typeof setTimeout> | undefined;
const response = new Response(new ReadableStream<Uint8Array>({
start(controller) {
timer = setTimeout(() => controller.enqueue(encoder.encode("late")), 30);
},
cancel() {
if (timer) clearTimeout(timer);
},
}));

const result = await readBoundedResponseBody(response, {
totalTimeoutMs: 100,
inactivityTimeoutMs: 15,
});

expect(result.inactivityTimedOut).toBe(true);
expect(result.totalTimedOut).toBe(false);
});

test("a partial body followed by silence times out and flushes UTF-8", async () => {
const response = new Response(new ReadableStream<Uint8Array>({
start(controller) {
Expand Down
Loading