Skip to content

fix: finish passthrough SSE after terminal events #1055

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
itling wants to merge 1 commit into from
Closed

fix: finish passthrough SSE after terminal events #1055

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
20 changes: 19 additions & 1 deletion src/server/relay.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -517,11 +517,14 @@ export function relaySseWithHeartbeat(
if (!body) return null;
const reader = body.getReader();
const decoder = new TextDecoder();
const heartbeat = new TextEncoder().encode(": opencodex keepalive\n\n");
const encoder = new TextEncoder();
const heartbeat = encoder.encode(": opencodex keepalive\n\n");
const doneSentinel = encoder.encode("data: [DONE]\n\n");
let timer: ReturnType<typeof setInterval> | undefined;
let closed = false;
let clientCancelled = false;
let terminalReported = false;
let doneSeen = false;
let buffer = "";

const reportTerminal = (status: ResponsesTerminalStatus) => {
Expand All @@ -532,6 +535,10 @@ export function relaySseWithHeartbeat(

const inspectPayload = (payload: string | null) => {
if (!payload) return;
if (payload === "[DONE]") {
doneSeen = true;
return;
}
const status = terminalStatusFromSsePayload(payload);
if (status) reportTerminal(status);
};
Expand Down Expand Up @@ -571,13 +578,24 @@ export function relaySseWithHeartbeat(
if (done) {
buffer += decoder.decode();
if (buffer.trim()) inspectPayload(sseDataPayload(buffer));
if (terminalReported && !doneSeen) {
controller.enqueue(doneSentinel);
Comment on lines +581 to +582

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 5, 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 Separate synthetic DONE from unterminated terminal frames

If the upstream closes after a terminal frame that lacks the final blank SSE delimiter, the relay already forwarded those raw bytes during the previous pull and this EOF path then appends data: [DONE]\n\n directly after them. That turns the terminal payload and [DONE] into one malformed SSE block (data: <json>\ndata: [DONE]) instead of dispatching the terminal event followed by a separate DONE sentinel; insert a delimiter before the synthetic sentinel or use the frame-aware terminal boundary for this path.

Useful? React with 👍 / 👎.

doneSeen = true;
}
if (!terminalReported && !clientCancelled) reportTerminal("incomplete");
cleanup();
controller.close();
return;
}
inspectChunk(value);
controller.enqueue(value);
if (terminalReported && !doneSeen) {

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 5, 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 Close the relay even when DONE is already present

When an upstream batches a Responses terminal event and data: [DONE] in the same chunk but keeps the HTTP connection open, inspectChunk() sets both terminalReported and doneSeen, so this new branch is skipped and the relay never runs cleanup() or controller.close(). That leaves the heartbeat timer active and onDone uncalled despite the terminal event already being forwarded; only the synthetic DONE enqueue should be guarded by !doneSeen, while close/cleanup should still happen whenever a terminal was reported.

Useful? React with 👍 / 👎.

controller.enqueue(doneSentinel);
doneSeen = true;
cleanup();
controller.close();
reader.cancel("Responses terminal event received").catch(() => {});

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 5, 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 Drop coalesced frames after the terminal event

If a gateway writes response.completed and another SSE block in the same upstream chunk, this branch closes only after controller.enqueue(value) has already forwarded the whole chunk, so post-terminal deltas or metadata can reach the client before the synthetic [DONE]. The terminal-aware relays elsewhere use an SSE frame boundary to forward through the terminal block and drop later blocks; this path should do the same before cancelling the reader.

Useful? React with 👍 / 👎.

}
Comment on lines 590 to +598

@coderabbitai coderabbitai Bot Aug 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift

Do not forward bytes after the terminal SSE block.

Line 590 can leave an incomplete later block in buffer, but Line 591 forwards the entire value. If one upstream read contains a complete response.completed block followed by the first bytes of a real data: [DONE] block, doneSeen remains false. Lines 592-598 then append a second sentinel and cancel before the next read can complete the original block. The client receives adjacent data: [DONE] lines, which an SSE parser reads as one payload instead of [DONE].

Return the terminal block boundary from the inspector and enqueue only bytes through that boundary. Discard post-terminal bytes before emitting the synthetic sentinel. Add a split-chunk regression in tests/passthrough-abort.test.ts around Lines 224-238.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/server/relay.ts` around lines 590 - 598, The relay stream handling around
inspectChunk must return the byte boundary of the terminal SSE block, then
enqueue only the input through that boundary and discard any following bytes
before emitting the synthetic doneSentinel. Update terminal detection so
doneSeen reflects the inspected terminal block even when a later [DONE] block is
split across the same read, and add a split-chunk regression covering this case
in passthrough-abort.test.ts.

} catch (err) {
if (!clientCancelled) reportTerminal("incomplete");
cleanup();
Expand Down
15 changes: 15 additions & 0 deletions tests/passthrough-abort.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -221,6 +221,21 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => {
expect(terminals).toEqual(["completed"]);
});

test("SSE passthrough appends DONE after a terminal payload without upstream DONE", async () => {
const enc = new TextEncoder();
const ac = new AbortController();
const terminals: string[] = [];
const relayed = relaySseWithHeartbeat(streamFromChunks([
enc.encode('event: response.completed\ndata: {"type":"response.completed","response":{"id":"r1","status":"completed"}}\n\n'),
]), ac, 15_000, status => terminals.push(status))!;
Comment on lines +228 to +230

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 5, 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 Exercise the relay used by native passthrough

This regression constructs relaySseWithHeartbeat() directly, but this same test file asserts that the real native Responses passthrough branch does not call relaySseWithHeartbeat() and production now routes through relaySseWithFailedTail() instead. As a result, the new test can pass while the described passthrough path still hangs or formats DONE incorrectly; cover relaySseWithFailedTail() or a server-level passthrough stream so the focused regression protects the changed behavior.

AGENTS.md reference: AGENTS.md:L228-L230

Useful? React with 👍 / 👎.


const raw = await readAll(relayed);

expect(raw).toContain('event: response.completed\ndata: {"type":"response.completed","response":{"id":"r1","status":"completed"}}\n\n');
expect(raw).toEndWith("data: [DONE]\n\n");
expect(terminals).toEqual(["completed"]);
});

test("SSE passthrough treats DONE without a terminal as incomplete", async () => {
const enc = new TextEncoder();
const ac = new AbortController();
Expand Down
Loading