Skip to content

fix(streaming): bound translated SSE inspection #1114

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Ingwannu wants to merge 1 commit into from
Closed

fix(streaming): bound translated SSE inspection #1114

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
60 changes: 16 additions & 44 deletions src/server/relay.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ import {
addFinalRequestLog,
httpStatusForRequestLogTerminal,
inspectResponseLogJson,
inspectResponseLogSsePayload,
inspectResponseLogSsePayloadParsed,
recordFirstOutput,
type RequestLogContext,
Expand Down Expand Up @@ -359,53 +358,42 @@ export function trackSseForRequestLog(
onFirstOutput?: () => void,
): ReadableStream<Uint8Array> {
const reader = body.getReader();
const decoder = new TextDecoder();
let buffer = "";
let terminalReported = false;
const reportFirstOutput = createFirstOutputReporter(onFirstOutput);

const reportTerminal = (status: ResponsesTerminalStatus) => {
if (terminalReported) return;
terminalReported = true;
onTerminal(status);
};

const inspectPayload = (payload: string | null) => {
if (!payload) return;
if (logCtx) inspectResponseLogSsePayload(logCtx, payload);
reportFirstOutput.payload(payload);
const status = terminalStatusFromSsePayload(payload);
if (status) reportTerminal(status);
};

const inspectChunk = (value: Uint8Array) => {
buffer += decoder.decode(value, { stream: true });
let next: { block: string; rest: string } | null;
while ((next = nextSseBlock(buffer))) {
buffer = next.rest;
inspectPayload(sseDataPayload(next.block));
}
};
// Reuse the byte-bounded inspector so translated responses cannot retain an
// unterminated upstream frame or parse the same event once per observer.
const inspector = createSseInspector({
onTerminal: reportTerminal,
logCtx,
onFirstOutput,
});

return new ReadableStream<Uint8Array>({
async pull(controller) {
try {
const { done, value } = await reader.read();
if (done) {
buffer += decoder.decode();
if (buffer.trim()) inspectPayload(sseDataPayload(buffer));
inspector.finish();
if (!terminalReported) reportTerminal("incomplete");
Comment on lines +381 to 382

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 6, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Preserve terminals for oversized completed frames

When a translated stream's terminal frame itself exceeds the 4 MiB inspection cap (the bridge allows much larger SSE events, and response.completed carries the full output/compaction item), createSseInspector discards that frame and never calls reportTerminal; on clean EOF this branch then records "incomplete" even though the client already received a successful terminal event. This corrupts request-log/account outcome data for large but valid translated responses, so terminal reporting needs to come from the bridge side or from a bounded prefix detector before the frame is discarded.

Useful? React with 👍 / 👎.

inspector.dispose();
controller.close();
return;
}
inspectChunk(value);
inspector.feed(value);
controller.enqueue(value);
} catch (err) {
if (!terminalReported) reportTerminal("incomplete");
inspector.dispose();
try { controller.error(err); } catch { /* already torn down */ }
}
},
cancel(reason) {
inspector.dispose();
onCancel();
reader.cancel(reason).catch(() => {});
},
Expand Down Expand Up @@ -516,38 +504,23 @@ export function relaySseWithHeartbeat(
): ReadableStream<Uint8Array> | null {
if (!body) return null;
const reader = body.getReader();
const decoder = new TextDecoder();
const heartbeat = new TextEncoder().encode(": opencodex keepalive\n\n");
let timer: ReturnType<typeof setInterval> | undefined;
let closed = false;
let clientCancelled = false;
let terminalReported = false;
let buffer = "";

const reportTerminal = (status: ResponsesTerminalStatus) => {
if (terminalReported || clientCancelled || closed) return;
terminalReported = true;
onTerminal?.(status);
};

const inspectPayload = (payload: string | null) => {
if (!payload) return;
const status = terminalStatusFromSsePayload(payload);
if (status) reportTerminal(status);
};

const inspectChunk = (value: Uint8Array) => {
buffer += decoder.decode(value, { stream: true });
let next: { block: string; rest: string } | null;
while ((next = nextSseBlock(buffer))) {
buffer = next.rest;
inspectPayload(sseDataPayload(next.block));
}
};
const inspector = createSseInspector({ onTerminal: reportTerminal });

const cleanup = () => {
if (closed) return;
closed = true;
inspector.dispose();
if (timer) clearInterval(timer);
timer = undefined;
options?.onDone?.();
Expand All @@ -569,14 +542,13 @@ export function relaySseWithHeartbeat(
try {
const { done, value } = await reader.read();
if (done) {
buffer += decoder.decode();
if (buffer.trim()) inspectPayload(sseDataPayload(buffer));
inspector.finish();
if (!terminalReported && !clientCancelled) reportTerminal("incomplete");
cleanup();
controller.close();
return;
}
inspectChunk(value);
inspector.feed(value);
controller.enqueue(value);
} catch (err) {
if (!clientCancelled) reportTerminal("incomplete");
Expand Down
5 changes: 5 additions & 0 deletions structure/04_transports-and-sidecars.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,11 @@ The two-shape contract is mirror-commented in `src/server/index.ts`; the real
and the platform matrix lives in `tests/bun-stream-caps.test.ts`. Keep all three
in lockstep with any passthrough-policy change.

Translated response request-log tracking and the heartbeat relay also reuse
`createSseInspector`. This keeps every client-facing SSE observation path on
the same byte-bounded, discard-and-resynchronize frame policy and ensures the
request-log, first-output, and terminal observers share one payload parse.

## Standalone Search and exact account selectors

`POST /v1/alpha/search` retains the selected model in its request body. When that value is an
Expand Down
65 changes: 65 additions & 0 deletions tests/sse-inspector-bounds.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,9 @@ import {
getInspectionCounters,
MAX_COMPLETED_OUTPUT_ITEMS,
MAX_INSPECTION_SSE_FRAME_BYTES,
relaySseWithHeartbeat,
resetInspectionCountersForTest,
trackSseForRequestLog,
} from "../src/server/relay";
import type { RequestLogContext } from "../src/server/request-log";

Expand Down Expand Up @@ -379,3 +381,66 @@ describe("createSseInspector parse-once", () => {
}
});
});

describe("client-facing SSE wrapper bounds", () => {
test("translated request-log tracking discards an oversized frame, resynchronizes, and preserves bytes", async () => {
const oversized = encoder.encode(`data: ${"x".repeat(MAX_INSPECTION_SSE_FRAME_BYTES)}x`);
const delimiter = encoder.encode("\n\n");
const terminal = frame(completedEvent("translated-after-cap"));
const chunks = [oversized, delimiter, terminal];
const terminals: string[] = [];
const tracked = trackSseForRequestLog(
streamFromChunks(chunks),
status => terminals.push(status),
() => {},
{} as RequestLogContext,
() => {},
);

expect(await readAllBytes(tracked)).toEqual(joinBytes(chunks));
expect(terminals).toEqual(["completed"]);
expect(getInspectionCounters().frameCapOverflows).toBe(1);
expect(getInspectionCounters().frameBufferHighWaterBytes)
.toBeLessThanOrEqual(MAX_INSPECTION_SSE_FRAME_BYTES);
});

test("translated request-log tracking parses a complete payload once for all observers", async () => {
const originalParse = JSON.parse;
let parses = 0;
JSON.parse = ((text: string) => {
parses += 1;
return originalParse(text);
}) as typeof JSON.parse;
try {
const tracked = trackSseForRequestLog(
streamFromChunks([frame(completedEvent("translated-parse-once"))]),
() => {},
() => {},
{} as RequestLogContext,
() => {},
);
await readAllBytes(tracked);
expect(parses).toBe(1);
} finally {
JSON.parse = originalParse;
}
});

test("heartbeat relay applies the same frame bound without changing upstream bytes", async () => {
const oversized = encoder.encode(`data: ${"x".repeat(MAX_INSPECTION_SSE_FRAME_BYTES)}x`);
const delimiter = encoder.encode("\r\n\r\n");
const terminal = frame(completedEvent("heartbeat-after-cap"));
const chunks = [oversized, delimiter, terminal];
const terminals: string[] = [];
const relayed = relaySseWithHeartbeat(
streamFromChunks(chunks),
new AbortController(),
60_000,
status => terminals.push(status),
)!;

expect(await readAllBytes(relayed)).toEqual(joinBytes(chunks));
expect(terminals).toEqual(["completed"]);
expect(getInspectionCounters().frameCapOverflows).toBe(1);
});
});
Loading