Skip to content

fix(streaming): relay Darwin rewrites eagerly (#1127) #1142

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

Merged
lidge-jun merged 1 commit into from
Aug 7, 2026
+106 −48
Merged

fix(streaming): relay Darwin rewrites eagerly (#1127) #1142

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
6 changes: 3 additions & 3 deletions src/lib/bun-stream-caps.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,7 @@ export function decideEagerRelay(
* Windows preserves the decision for no-rewrite traffic. Darwin permits only
* explicit config opt-in; `auto` remains tee even on a future fixed runtime.
* Returns the normalized effective decision, or null when platform policy,
* rewrite needs, or a Darwin non-config-eager mode selects tee.
* Windows rewrite needs, or a Darwin non-config-eager mode selects tee.
*/
export function selectEagerPath(
platform: NodeJS.Platform,
Expand All @@ -103,12 +103,12 @@ export function selectEagerPath(
version: string = Bun.version,
minFixed: string | null = MIN_FIXED_BUN_VERSION,
): EagerRelayDecision | null {
if (needsClientRewrite || (platform !== "win32" && platform !== "darwin")) {
if (platform !== "win32" && platform !== "darwin") {

@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 Update the declared Darwin transport contract

For darwin + needsClientRewrite + eager-relay, this selector now returns config-eager, but the module header (src/lib/bun-stream-caps.ts:9-12), the mirror comment (src/server/index.ts:343-345), and the architecture contract (structure/04_transports-and-sidecars.md:51-62) still state that Darwin eager relay is restricted to no-rewrite traffic; the architecture document explicitly requires these descriptions and the platform matrix to remain in lockstep. Update those declarations with this policy change so future maintenance and source-invariant tests do not preserve the obsolete fallback.

AGENTS.md reference: src/AGENTS.md:L10-L11

Useful? React with 👍 / 👎.

return null;
}

const decision = decideEagerRelay(mode, version, minFixed);
if (platform === "win32") return decision;
if (platform === "win32") return needsClientRewrite ? null : decision;
return decision.reason === "config-eager" ? decision : null;
}

Expand Down
10 changes: 6 additions & 4 deletions src/server/responses/core.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2135,6 +2135,8 @@ async function handleResponsesInner(
needsClientRewrite,
config.streamMode ?? "auto",
);
const inlineEagerRewrite = needsClientRewrite
&& (win32EagerRewrite || eagerPath?.useEagerRelay === true);
Comment on lines +2138 to +2139

@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 Emit a failed terminal when an inline terminal rewrite throws

On Darwin with explicit eager-relay, a client rewrite can throw while processing response.completed, for example when item-ID or Copilot retained state exceeds its translator budget. The eager producer feeds the raw chunk to inspector before rewriting it, so sawTerminal() is already true; its catch block consequently suppresses the synthetic failure, closes the client stream without the completed block or a response.failed tail, and leaves accounting recorded as completed. The prior Darwin tee path converted this rewrite error into a typed failed terminal, so base failure suppression on whether a terminal was emitted to the client rather than whether the raw inspector saw one.

AGENTS.md reference: src/AGENTS.md:L17-L19

Useful? React with 👍 / 👎.

if (eagerPath?.useEagerRelay || win32EagerRewrite) {
const turnAc = new AbortController();
linkAbortSignal(upstream, turnAc.signal);
Expand Down Expand Up @@ -2191,11 +2193,11 @@ async function handleResponsesInner(
},
onClientCancel: () => options.onNativePassthroughCancel?.(),
onDone: () => unregisterTurn(turnAc),
}, win32EagerRewrite ? { rewriteBudget: translatorBudget } : undefined);
}, inlineEagerRewrite ? { rewriteBudget: translatorBudget } : undefined);

@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.

P1 Badge Bound partial frames before entering the Darwin eager rewrite

For Darwin with explicit eager-relay and any client rewrite, supplying the translator budget here does not bound partial SSE blocks: relaySseEagerBounded calls terminalBoundary.feed(value) before rewriteOutbound, and that boundary accumulates an incomplete event in an unmetered string. The previous Darwin tee path passed every fragment through relaySseWithBlockRewrite, whose reservations fail at the 32 MiB turn limit; after this change, a broken gateway sending an arbitrarily large or never-delimited data: event can grow proxy RSS without limit. Account this framing in the rewrite budget or impose an equivalent cap at the terminal boundary.

AGENTS.md reference: src/AGENTS.md:L15-L19

Useful? React with 👍 / 👎.

// When selected, this relay closes response.completed even if upstream
// keeps the connection alive. Windows rewrite traffic applies its
// payload transform inline — never via the Bun#32111-unsafe
// tee()+JS-pull chain (#864).
// keeps the connection alive. Windows forced-rewrite traffic and Darwin
// explicit eager traffic apply client rewrites inline rather than via
// the tee()+JS-pull chain.
if (!headers.has("content-type")) headers.set("content-type", "text/event-stream");
return markEagerRelaySseResponse(
markNativePassthroughSseResponse(new Response(eagerBody, {
Expand Down
72 changes: 34 additions & 38 deletions tests/bun-stream-caps.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,44 +98,40 @@ describe("decideEagerRelay (activation scenarios)", () => {
});

describe("selectEagerPath (platform policy matrix)", () => {
test("win32 + no rewrite + config-eager → eager", () => {
expect(selectEagerPath("win32", false, "eager-relay", "1.3.14", null))
.toEqual({ useEagerRelay: true, reason: "config-eager" });
});

test("win32 + no rewrite + auto known-bad → tee", () => {
expect(selectEagerPath("win32", false, "auto", "1.3.14", null))
.toEqual({ useEagerRelay: false, reason: "auto-known-bad" });
});

test("win32 + no rewrite + auto fixed runtime → eager", () => {
expect(selectEagerPath("win32", false, "auto", "1.4.0", "1.4.0"))
.toEqual({ useEagerRelay: true, reason: "auto-fixed-runtime" });
});

test("darwin + no rewrite + config-eager → eager", () => {
expect(selectEagerPath("darwin", false, "eager-relay", "1.3.14", null))
.toEqual({ useEagerRelay: true, reason: "config-eager" });
});

test("darwin + no rewrite + auto fixed runtime → tee with no eager decision", () => {
expect(selectEagerPath("darwin", false, "auto", "1.4.0", "1.4.0")).toBeNull();
});

test("darwin + rewrite + config-eager → tee", () => {
expect(selectEagerPath("darwin", true, "eager-relay", "1.3.14", null)).toBeNull();
});

test("linux + config-eager → tee", () => {
expect(selectEagerPath("linux", false, "eager-relay", "1.3.14", null)).toBeNull();
});

test("legacy-tee is a Windows decision and null on ineligible platforms", () => {
expect(selectEagerPath("win32", false, "legacy-tee", "9.9.9", "1.4.0"))
.toEqual({ useEagerRelay: false, reason: "config-legacy" });
expect(selectEagerPath("darwin", false, "legacy-tee", "9.9.9", "1.4.0")).toBeNull();
expect(selectEagerPath("linux", false, "legacy-tee", "9.9.9", "1.4.0")).toBeNull();
});
const configLegacy = { useEagerRelay: false, reason: "config-legacy" } as const;
const configEager = { useEagerRelay: true, reason: "config-eager" } as const;
const autoFixed = { useEagerRelay: true, reason: "auto-fixed-runtime" } as const;
const cases: Array<{
platform: NodeJS.Platform;
mode: "auto" | "legacy-tee" | "eager-relay";
rewrite: boolean;
expected: typeof configLegacy | typeof configEager | typeof autoFixed | null;
}> = [
{ platform: "win32", mode: "legacy-tee", rewrite: false, expected: configLegacy },
{ platform: "win32", mode: "eager-relay", rewrite: false, expected: configEager },
{ platform: "win32", mode: "auto", rewrite: false, expected: autoFixed },
{ platform: "win32", mode: "legacy-tee", rewrite: true, expected: null },
{ platform: "win32", mode: "eager-relay", rewrite: true, expected: null },
{ platform: "win32", mode: "auto", rewrite: true, expected: null },
{ platform: "darwin", mode: "legacy-tee", rewrite: false, expected: null },
{ platform: "darwin", mode: "eager-relay", rewrite: false, expected: configEager },
{ platform: "darwin", mode: "auto", rewrite: false, expected: null },
{ platform: "darwin", mode: "legacy-tee", rewrite: true, expected: null },
{ platform: "darwin", mode: "eager-relay", rewrite: true, expected: configEager },
{ platform: "darwin", mode: "auto", rewrite: true, expected: null },
{ platform: "linux", mode: "legacy-tee", rewrite: false, expected: null },
{ platform: "linux", mode: "eager-relay", rewrite: false, expected: null },
{ platform: "linux", mode: "auto", rewrite: false, expected: null },
{ platform: "linux", mode: "legacy-tee", rewrite: true, expected: null },
{ platform: "linux", mode: "eager-relay", rewrite: true, expected: null },
{ platform: "linux", mode: "auto", rewrite: true, expected: null },
];

for (const { platform, mode, rewrite, expected } of cases) {
test(`${platform} + ${mode} + rewrite=${rewrite}`, () => {
expect(selectEagerPath(platform, rewrite, mode, "1.4.0", "1.4.0")).toEqual(expected);
});
}
});

describe("isStreamMode", () => {
Expand Down
22 changes: 19 additions & 3 deletions tests/relay-eager.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -270,18 +270,34 @@ describe("relaySseEagerBounded — inline payload rewrite (#864)", () => {
const budget = createTranslatorBudget();
const up = controlledUpstream();
const ac = new AbortController();
const { hooks } = makeHooks();
const { hooks, rec } = makeHooks();
let resolveDone!: () => void;
const done = new Promise<void>(resolve => { resolveDone = resolve; });
const previousOnDone = hooks.onDone;
hooks.onDone = () => {
previousOnDone();
resolveDone();
};
hooks.rewritePayload = (payload: string) => payload;
relaySseEagerBounded(up.stream, ac, hooks, { rewriteBudget: budget });

up.push(enc.encode(`data: {"type":"unterminated"`));
await settle();
// The shared terminal boundary now owns incomplete SSE framing, so the
// downstream rewrite stage never retains an unterminated block.
expect(budget.snapshot().currentBytes).toBe(0);
ac.abort(new Error("test abort"));
await settle();
let timeout: ReturnType<typeof setTimeout> | undefined;
await Promise.race([
done,
Comment on lines +290 to +291

@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 Wait for the pushed fragment before aborting

The new done promise only confirms teardown after ac.abort(); it does not confirm that the fragment pushed on line 284 reached terminalBoundary.feed(). Since enqueue() resolves the pending reader.read() asynchronously, the abort can be observed first, causing the producer to exit before exercising the rewrite boundary, while both zero-byte assertions still pass. Await a deterministic inspection/processing signal before the first assertion and abort so this remains a focused regression test rather than a vacuous cleanup check.

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

Useful? React with 👍 / 👎.

new Promise<never>((_, reject) => {
timeout = setTimeout(() => reject(new Error("relay cleanup timed out")), 2_000);
}),
]).finally(() => {
if (timeout) clearTimeout(timeout);
});
expect(budget.snapshot().currentBytes).toBe(0);
expect(rec.dones).toBe(1);
budget.dispose();
});

test("blocks without a data field pass through untouched before the terminal", async () => {
Expand Down
44 changes: 44 additions & 0 deletions tests/responses-snapshot-repair-server.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@ import { tmpdir } from "node:os";
import { join } from "node:path";
import { saveConfig } from "../src/config";
import { startServer } from "../src/server";
import { handleResponses } from "../src/server/responses";
import { isEagerRelaySseResponse } from "../src/server/relay";
import type { OcxConfig } from "../src/types";
import { installIsolatedCodexHome, type IsolatedCodexHome } from "./helpers/isolated-codex-home";

Expand Down Expand Up @@ -63,6 +65,48 @@ afterEach(async () => {
});

describe("responsesSnapshotRepair through /v1/responses", () => {
test.skipIf(process.platform !== "darwin")(
"Darwin eager-relay applies snapshot repair inline before bytes reach the client",
async () => {
const gateway = "https://sparse-darwin-eager.example.test";
stubSparseGateway(gateway);
const config = {
port: 0,
streamMode: "eager-relay",
defaultProvider: "sparse",
providers: {
sparse: {
adapter: "openai-responses",
baseUrl: `${gateway}/v1`,
authMode: "key",
apiKey: "test-key",
responsesSnapshotRepair: true,
},
},
} as OcxConfig;

const response = await handleResponses(
new Request("http://localhost/v1/responses", {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({ model: "sparse-model", input: "hi", stream: true }),
}),
config,
{ model: "", provider: "" },
);

expect(isEagerRelaySseResponse(response)).toBe(true);
const text = await response.text();
expect(text).toContain("response.content_part.added");
expect(text).toContain("response.output_text.done");
expect(text).toContain("response.output_item.done");
const completedLine = text.split("\n").find(line => line.includes('"response.completed"'));
expect(completedLine).toBeDefined();
const completed = JSON.parse(completedLine!.replace(/^data: /, "")) as { response: { output: { id: string }[] } };
expect(completed.response.output[0]?.id).toBe("msg_sparse");
},
);

test("an opt-in gateway's sparse stream reaches the client as the full canonical lifecycle", async () => {
const gateway = "https://sparse.example.test";
stubSparseGateway(gateway);
Expand Down
Loading