diff --git a/src/lib/bun-stream-caps.ts b/src/lib/bun-stream-caps.ts index 09395aab6..e86df8a55 100644 --- a/src/lib/bun-stream-caps.ts +++ b/src/lib/bun-stream-caps.ts @@ -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, @@ -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") { 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; } diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index e8bc1e7e8..539617208 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -2135,6 +2135,8 @@ async function handleResponsesInner( needsClientRewrite, config.streamMode ?? "auto", ); + const inlineEagerRewrite = needsClientRewrite + && (win32EagerRewrite || eagerPath?.useEagerRelay === true); if (eagerPath?.useEagerRelay || win32EagerRewrite) { const turnAc = new AbortController(); linkAbortSignal(upstream, turnAc.signal); @@ -2191,11 +2193,11 @@ async function handleResponsesInner( }, onClientCancel: () => options.onNativePassthroughCancel?.(), onDone: () => unregisterTurn(turnAc), - }, win32EagerRewrite ? { rewriteBudget: translatorBudget } : undefined); + }, inlineEagerRewrite ? { rewriteBudget: translatorBudget } : undefined); // 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, { diff --git a/tests/bun-stream-caps.test.ts b/tests/bun-stream-caps.test.ts index 2a7ad0b4c..93c79808d 100644 --- a/tests/bun-stream-caps.test.ts +++ b/tests/bun-stream-caps.test.ts @@ -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", () => { diff --git a/tests/relay-eager.test.ts b/tests/relay-eager.test.ts index 9840ead4b..e8e30409c 100644 --- a/tests/relay-eager.test.ts +++ b/tests/relay-eager.test.ts @@ -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(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 | undefined; + await Promise.race([ + done, + new Promise((_, 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 () => { diff --git a/tests/responses-snapshot-repair-server.test.ts b/tests/responses-snapshot-repair-server.test.ts index ddfba352d..8e5c55bf5 100644 --- a/tests/responses-snapshot-repair-server.test.ts +++ b/tests/responses-snapshot-repair-server.test.ts @@ -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"; @@ -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);