From 2670f24d4c3ff4a140f8969fafce08a6dea02a97 Mon Sep 17 00:00:00 2001 From: Jingwei <164893774+0xJingwei@users.noreply.github.com> Date: Sun, 2 Aug 2026 22:29:47 +0800 Subject: [PATCH 1/4] fix: repair incomplete Responses snapshots --- src/config.ts | 1 + src/lib/bun-stream-caps.ts | 9 +- src/server/responses-snapshot-repair.ts | 152 ++++++++++++++++++++++++ src/server/responses/core.ts | 38 +++--- src/types.ts | 6 + tests/bun-stream-caps.test.ts | 9 +- tests/config.test.ts | 30 +++++ tests/passthrough-abort.test.ts | 7 +- tests/relay-eager.test.ts | 29 +++++ tests/responses-snapshot-repair.test.ts | 120 +++++++++++++++++++ 10 files changed, 379 insertions(+), 22 deletions(-) create mode 100644 src/server/responses-snapshot-repair.ts create mode 100644 tests/responses-snapshot-repair.test.ts diff --git a/src/config.ts b/src/config.ts index dd09ce8b4..a33865043 100644 --- a/src/config.ts +++ b/src/config.ts @@ -489,6 +489,7 @@ const providerConfigSchema = z.object({ reasoning: z.array(z.string().min(1)).optional(), repairMissingTerminalIds: z.boolean().optional(), }).strict().optional(), + responsesSnapshotRepair: z.boolean().optional(), }).passthrough(); const RESERVED_PROVIDER_NAMES = new Set(["__proto__", "prototype", "constructor"]); diff --git a/src/lib/bun-stream-caps.ts b/src/lib/bun-stream-caps.ts index 188f10b51..6ed7efa8d 100644 --- a/src/lib/bun-stream-caps.ts +++ b/src/lib/bun-stream-caps.ts @@ -92,8 +92,10 @@ export function decideEagerRelay( * Apply the two-platform eager-relay policy to the runtime/config capability. * 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. + * Returns the normalized effective decision, or null when platform policy or a Darwin + * non-config-eager mode selects tee. Windows rewrites remain on tee because the extra JS pull + * wrapper is part of the Bun#32111 crash shape; Darwin's explicit eager mode may compose a + * client-facing rewrite after the single-reader relay. */ export function selectEagerPath( platform: NodeJS.Platform, @@ -102,7 +104,8 @@ export function selectEagerPath( version: string = Bun.version, minFixed: string | null = MIN_FIXED_BUN_VERSION, ): EagerRelayDecision | null { - if (needsClientRewrite || (platform !== "win32" && platform !== "darwin")) { + if ((needsClientRewrite && platform === "win32") + || (platform !== "win32" && platform !== "darwin")) { return null; } diff --git a/src/server/responses-snapshot-repair.ts b/src/server/responses-snapshot-repair.ts new file mode 100644 index 000000000..17d3014b3 --- /dev/null +++ b/src/server/responses-snapshot-repair.ts @@ -0,0 +1,152 @@ +import type { SsePayloadRewrite } from "./sse-payload-rewrite"; + +const RESPONSE_EVENT_STATUSES: Readonly> = { + "response.created": "in_progress", + "response.in_progress": "in_progress", + "response.completed": "completed", + "response.failed": "failed", + "response.incomplete": "incomplete", + "response.queued": "queued", +}; + +function isPlainObject(value: unknown): value is Record { + return !!value && typeof value === "object" && !Array.isArray(value); +} + +function repairOutputTextPart(part: Record): Record { + if (part.type !== "output_text" || Array.isArray(part.annotations)) return part; + return { ...part, annotations: [] }; +} + +function repairSummaryPart(part: Record): Record { + if (part.type !== "summary_text" || typeof part.text === "string") return part; + return { ...part, text: "" }; +} + +function repairOutputItem(item: Record): Record { + if (item.type === "reasoning") { + return Array.isArray(item.summary) ? item : { ...item, summary: [] }; + } + if (item.type !== "message") return item; + + let changed = false; + const content = Array.isArray(item.content) + ? item.content.map((part) => { + if (!isPlainObject(part)) return part; + const repaired = repairOutputTextPart(part); + changed = changed || repaired !== part; + return repaired; + }) + : []; + changed = changed || !Array.isArray(item.content); + const role = item.role === "assistant" ? item.role : "assistant"; + changed = changed || role !== item.role; + return changed ? { ...item, content, role } : item; +} + +function repairResponseSnapshot( + response: Record, + defaultStatus: string, +): Record { + const repaired = { ...response }; + let changed = false; + + if (Array.isArray(repaired.output)) { + const output = repaired.output.map((item) => { + if (!isPlainObject(item)) return item; + const next = repairOutputItem(item); + changed = changed || next !== item; + return next; + }); + if (changed) repaired.output = output; + } else { + repaired.output = []; + changed = true; + } + if (typeof repaired.parallel_tool_calls !== "boolean") { + repaired.parallel_tool_calls = true; + changed = true; + } + if (repaired.tool_choice === undefined || repaired.tool_choice === null) { + repaired.tool_choice = "auto"; + changed = true; + } + if (!Array.isArray(repaired.tools)) { + repaired.tools = []; + changed = true; + } + if (typeof repaired.status !== "string") { + repaired.status = defaultStatus; + changed = true; + } + + return changed ? repaired : response; +} + +/** + * Repair required fields omitted by a few Responses-compatible gateways across lifecycle events. + * Existing upstream values remain authoritative; only absent or structurally invalid fields are + * backfilled. + */ +export function createResponsesSnapshotPayloadRewrite(): SsePayloadRewrite { + return (payload) => { + let event: unknown; + try { + event = JSON.parse(payload); + } catch { + return payload; + } + if (!isPlainObject(event)) return payload; + const type = typeof event.type === "string" ? event.type : ""; + let nextEvent = event; + let changed = false; + + const responseStatus = RESPONSE_EVENT_STATUSES[type]; + if (responseStatus && isPlainObject(event.response)) { + const response = repairResponseSnapshot(event.response, responseStatus); + if (response !== event.response) { + nextEvent = { ...nextEvent, response }; + changed = true; + } + } + + if ((type === "response.output_item.added" || type === "response.output_item.done") + && isPlainObject(event.item)) { + const item = repairOutputItem(event.item); + if (item !== event.item) { + nextEvent = { ...nextEvent, item }; + changed = true; + } + } + + if ((type === "response.content_part.added" || type === "response.content_part.done") + && isPlainObject(event.part)) { + const part = repairOutputTextPart(event.part); + if (part !== event.part) { + nextEvent = { ...nextEvent, part }; + changed = true; + } + } + + if ((type === "response.reasoning_summary_part.added" + || type === "response.reasoning_summary_part.done") && isPlainObject(event.part)) { + const part = repairSummaryPart(event.part); + if (part !== event.part) { + nextEvent = { ...nextEvent, part }; + changed = true; + } + } + + if ((type === "response.output_text.delta" || type === "response.output_text.done") + && !Array.isArray(event.logprobs)) { + nextEvent = { ...nextEvent, logprobs: [] }; + changed = true; + } + + return changed ? JSON.stringify(nextEvent) : payload; + }; +} + +export function hasResponsesSnapshotRepair(enabled: boolean | undefined): enabled is true { + return enabled === true; +} diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index e714b3d81..40c7bb5fd 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -150,6 +150,10 @@ import { createResponsesItemIdPayloadRewrite, hasResponsesItemIdRepair, } from "../responses-item-id-repair"; +import { + createResponsesSnapshotPayloadRewrite, + hasResponsesSnapshotRepair, +} from "../responses-snapshot-repair"; import { createImageGenCallRestoreRewrite, imageGenToolCallAliases, @@ -1750,25 +1754,33 @@ async function handleResponsesInner( // native relay, never enters JS Sink.write); branch[1] is consumed in the // background for terminal-outcome/quota inspection only. // #314 alternative shape: win32 no-rewrite traffic follows the runtime/config - // gate; darwin no-rewrite traffic joins it only for explicit - // `streamMode: "eager-relay"` opt-in. Darwin `auto` always stays tee. The + // gate; Darwin traffic joins it only for explicit `streamMode: "eager-relay"` + // opt-in and may compose a client-facing rewrite after the single reader. + // Darwin `auto` always stays tee. The // eager shape skips tee and uses one bounded reader with inline inspection // (src/server/relay-eager.ts; policy: // devlog/_plan/260731_macos_rss_retention/100_darwin_eager_optin.md). // The bundled known-bad runtime remains on tee by default on both platforms. if (isEventStream && upstreamResponse.body) { const repairConfig = route.provider.responsesItemIdRepair; - const needsClientRewrite = imageGenCallAliases.size > 0 || hasResponsesItemIdRepair(repairConfig); - // Compose opt-in payload rewrites into one parse/stringify pass (image-gen restore first). + const snapshotRepair = route.provider.responsesSnapshotRepair; const payloadRewrites = [ createImageGenCallRestoreRewrite(imageGenCallAliases), hasResponsesItemIdRepair(repairConfig) ? createResponsesItemIdPayloadRewrite(repairConfig!, translatorBudget) : undefined, + hasResponsesSnapshotRepair(snapshotRepair) + ? createResponsesSnapshotPayloadRewrite() + : undefined, ].filter((rewrite): rewrite is NonNullable => rewrite !== undefined); + const needsClientRewrite = payloadRewrites.length > 0; + const payloadRewrite = needsClientRewrite + ? composeSsePayloadRewrites(...payloadRewrites) + : undefined; // #864: win32 rewrite traffic must never enter the tee()+JS-pull chain - // (Bun#32111 JS-sink segfault — text frames pass, the terminal block is - // lost). The eager single reader applies the same rewrites inline. + // (Bun#32111 JS-sink segfault). Darwin's explicit eager mode uses the + // same single-reader relay so snapshot repair cannot reintroduce tee + // retention on the third-party Responses path. const win32EagerRewrite = isWin32EagerRewrite(process.platform, needsClientRewrite); const eagerPath = selectEagerPath( process.platform, @@ -1812,9 +1824,7 @@ async function handleResponsesInner( finishInspection: () => inspector.finish(), disposeInspection: () => inspector.dispose(), sawTerminal: () => inspector.reported(), - ...(win32EagerRewrite - ? { rewritePayload: composeSsePayloadRewrites(...payloadRewrites) } - : {}), + ...(payloadRewrite ? { rewritePayload: payloadRewrite } : {}), onSynthetic: kind => { if (!reportNativeTerminal) return; if (kind === "incomplete") { @@ -1828,10 +1838,9 @@ async function handleResponsesInner( }, onClientCancel: () => options.onNativePassthroughCancel?.(), onDone: () => unregisterTurn(turnAc), - }, win32EagerRewrite ? { rewriteBudget: translatorBudget } : undefined); - // selectEagerPath admits only no-rewrite traffic on both eligible platforms; - // win32 rewrite traffic reaches this relay too, but with the payload rewrite - // applied inline — never via an image/item-id JS pull wrapper (#32111, #864). + }, payloadRewrite ? { rewriteBudget: translatorBudget } : undefined); + // win32 rewrite traffic and Darwin's explicit eager mode apply client + // rewrites inline after raw inspection, never through tee()+JS pull. if (!headers.has("content-type")) headers.set("content-type", "text/event-stream"); return markEagerRelaySseResponse( markNativePassthroughSseResponse(new Response(eagerBody, { @@ -1901,8 +1910,9 @@ async function handleResponsesInner( // win32 must keep the pure native relay (Bun#32111 JS-sink segfault); elsewhere a JS pull // relay is established practice (relayWithAbort, relaySseWithHeartbeat) and lets a // mid-stream reset end with a clean response.failed terminal instead of a raw socket error. + // Compose opt-in payload rewrites into one parse/stringify pass (image-gen restore first). const rewrittenBody = payloadRewrites.length > 0 - ? relaySseWithPayloadRewrite(nativeBody, composeSsePayloadRewrites(...payloadRewrites), translatorBudget) + ? relaySseWithPayloadRewrite(nativeBody, payloadRewrite!, translatorBudget) : nativeBody; const clientBody = process.platform === "win32" && !needsClientRewrite ? nativeBody diff --git a/src/types.ts b/src/types.ts index c4827a8fa..f0589dc73 100644 --- a/src/types.ts +++ b/src/types.ts @@ -1081,6 +1081,12 @@ export interface OcxProviderConfig { * Disabled by default; function_call ids and call_id pairing are never rewritten. */ responsesItemIdRepair?: ResponsesItemIdRepairConfig; + /** + * Provider-local repair for Responses gateways whose response.created / response.in_progress + * snapshots omit required canonical fields such as output, tools, and tool_choice. + * Disabled by default and applied only to the client-facing SSE branch. + */ + responsesSnapshotRepair?: boolean; /** Model ids whose tool_choice only accepts `auto` or `none`; forced/named choices are downgraded. */ autoToolChoiceOnlyModels?: string[]; /** Model ids that expect prior assistant `reasoning_content` to be preserved in chat history. */ diff --git a/tests/bun-stream-caps.test.ts b/tests/bun-stream-caps.test.ts index 2a7ad0b4c..493bbeda9 100644 --- a/tests/bun-stream-caps.test.ts +++ b/tests/bun-stream-caps.test.ts @@ -122,8 +122,13 @@ describe("selectEagerPath (platform policy matrix)", () => { 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("darwin + rewrite + config-eager → eager single-reader relay", () => { + expect(selectEagerPath("darwin", true, "eager-relay", "1.3.14", null)) + .toEqual({ useEagerRelay: true, reason: "config-eager" }); + }); + + test("win32 + rewrite + config-eager → tee", () => { + expect(selectEagerPath("win32", true, "eager-relay", "1.3.14", null)).toBeNull(); }); test("linux + config-eager → tee", () => { diff --git a/tests/config.test.ts b/tests/config.test.ts index b62da331e..38971de7f 100644 --- a/tests/config.test.ts +++ b/tests/config.test.ts @@ -648,6 +648,36 @@ describe("opencodex config defaults", () => { expect(readConfigDiagnostics().error).toContain("responsesItemIdRepair"); }); + test("accepts only a boolean responsesSnapshotRepair opt-in", () => { + writeConfig({ + port: 12345, + providers: { + custom: { + adapter: "openai-responses", + baseUrl: "https://example.test/v1", + responsesSnapshotRepair: true, + }, + }, + defaultProvider: "custom", + }); + expect(readConfigDiagnostics().error).toBeNull(); + expect(readConfigDiagnostics().config.providers.custom.responsesSnapshotRepair).toBe(true); + + writeConfig({ + port: 12345, + providers: { + custom: { + adapter: "openai-responses", + baseUrl: "https://example.test/v1", + responsesSnapshotRepair: { enabled: true }, + }, + }, + defaultProvider: "custom", + }); + expect(readConfigDiagnostics().source).toBe("fallback"); + expect(readConfigDiagnostics().error).toContain("responsesSnapshotRepair"); + }); + test("accepts a relative responsesPath", () => { writeResponsesPathConfig("/responses"); diff --git a/tests/passthrough-abort.test.ts b/tests/passthrough-abort.test.ts index 28592fb0d..fdff182b4 100644 --- a/tests/passthrough-abort.test.ts +++ b/tests/passthrough-abort.test.ts @@ -50,8 +50,9 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => { // win32 must receive the tee'd body untouched when no client rewrite is required — no JS pull // wrapper on the default path (Bun#32111 segfault). expect(sseBranch).toContain("const repairConfig = route.provider.responsesItemIdRepair;"); - expect(sseBranch).toContain("const needsClientRewrite = imageGenCallAliases.size > 0"); - expect(sseBranch).toContain("new Response(eagerBody"); + expect(sseBranch).toContain("const snapshotRepair = route.provider.responsesSnapshotRepair;"); + expect(sseBranch).toContain("const needsClientRewrite = payloadRewrites.length > 0;"); + expect(sseBranch).toContain("rewritePayload: payloadRewrite"); expect(sseBranch).toContain("const rewrittenBody = payloadRewrites.length > 0"); expect(sseBranch).toContain('process.platform === "win32"'); expect(sseBranch).toContain("&& !needsClientRewrite"); @@ -60,7 +61,7 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => { // reader with the payload rewrite applied inline — never the tee()+JS-pull // chain that loses the terminal block on Windows (Bun#32111). expect(sseBranch).toContain("win32EagerRewrite"); - expect(sseBranch).toContain("rewritePayload: composeSsePayloadRewrites(...payloadRewrites)"); + expect(sseBranch).toContain("...(payloadRewrite ? { rewritePayload: payloadRewrite } : {})"); // Elsewhere the failed-tail relay converts mid-stream resets into a clean response.failed. expect(sseBranch).toContain("relaySseWithFailedTail(rewrittenBody, upstream"); expect(sseBranch).toContain("new Response(clientBody"); diff --git a/tests/relay-eager.test.ts b/tests/relay-eager.test.ts index 2084ea8c1..ccbd2c2e9 100644 --- a/tests/relay-eager.test.ts +++ b/tests/relay-eager.test.ts @@ -292,6 +292,35 @@ describe("relaySseEagerBounded — side-effect parity", () => { expect(rec.dones).toBe(1); }); + test("rewrites client-facing SSE inline while inspection keeps the raw upstream bytes", async () => { + const { hooks, rec } = makeHooks(); + const inspected: string[] = []; + const rawInspect = hooks.inspectChunk; + hooks.inspectChunk = chunk => { + inspected.push(new TextDecoder().decode(chunk)); + rawInspect(chunk); + }; + hooks.rewritePayload = payload => { + const event = JSON.parse(payload) as Record; + return JSON.stringify({ ...event, client_repaired: true }); + }; + const up = controlledUpstream(); + const relayed = relaySseEagerBounded(up.stream, new AbortController(), hooks); + const created = sse(JSON.stringify({ type: "response.created", response: {} })); + const completed = sse(COMPLETED); + up.push(created.slice(0, 17)); + up.push(created.slice(17)); + up.push(completed); + up.close(); + + const text = await readAll(relayed); + await settle(); + expect(text).toContain('"client_repaired":true'); + expect(inspected.join("")).not.toContain("client_repaired"); + expect(rec.terminals).toEqual([{ status: "completed", httpStatus: undefined }]); + expect(rec.dones).toBe(1); + }); + test("eager relay backfills missing completed output before passthrough persistence (#334)", async () => { const { hooks, rec } = makeHooks(); const up = controlledUpstream(); diff --git a/tests/responses-snapshot-repair.test.ts b/tests/responses-snapshot-repair.test.ts new file mode 100644 index 000000000..3d95ce49c --- /dev/null +++ b/tests/responses-snapshot-repair.test.ts @@ -0,0 +1,120 @@ +import { describe, expect, test } from "bun:test"; +import { + createResponsesSnapshotPayloadRewrite, + hasResponsesSnapshotRepair, +} from "../src/server/responses-snapshot-repair"; + +describe("Responses passthrough sparse-snapshot repair", () => { + test("backfills canonical lifecycle fields missing from created and in-progress snapshots", () => { + const rewrite = createResponsesSnapshotPayloadRewrite(); + + for (const type of ["response.created", "response.in_progress"] as const) { + const payload = JSON.stringify({ + type, + sequence_number: 0, + response: { + id: "resp_sparse", + created_at: 1, + model: "example-model", + object: "response", + service_tier: "default", + store: false, + }, + }); + + const event = JSON.parse(rewrite(payload)) as { + response: Record; + }; + expect(event.response).toMatchObject({ + id: "resp_sparse", + status: "in_progress", + output: [], + parallel_tool_calls: true, + tool_choice: "auto", + tools: [], + }); + } + }); + + test("repairs sparse output-item, content-part, text, and terminal snapshots", () => { + const rewrite = createResponsesSnapshotPayloadRewrite(); + const cases = [ + { + input: { type: "response.output_item.added", item: { id: "rs_1", type: "reasoning", status: "in_progress" } }, + expected: { item: { summary: [] } }, + }, + { + input: { type: "response.output_item.added", item: { id: "msg_1", type: "message", status: "in_progress" } }, + expected: { item: { role: "assistant", content: [] } }, + }, + { + input: { type: "response.reasoning_summary_part.added", part: { type: "summary_text" } }, + expected: { part: { type: "summary_text", text: "" } }, + }, + { + input: { type: "response.content_part.added", part: { type: "output_text", text: "" } }, + expected: { part: { type: "output_text", text: "", annotations: [] } }, + }, + { + input: { type: "response.output_text.delta", delta: "ok" }, + expected: { logprobs: [] }, + }, + { + input: { + type: "response.completed", + response: { + id: "resp_terminal", + created_at: 1, + model: "model", + object: "response", + status: "completed", + output: [{ id: "msg_1", type: "message", role: "assistant", status: "completed", content: [{ type: "output_text", text: "ok" }] }], + }, + }, + expected: { + response: { + status: "completed", + parallel_tool_calls: true, + tool_choice: "auto", + tools: [], + output: [{ content: [{ type: "output_text", text: "ok", annotations: [] }] }], + }, + }, + }, + ]; + + for (const { input, expected } of cases) { + expect(JSON.parse(rewrite(JSON.stringify(input)))).toMatchObject(expected); + } + }); + + test("preserves upstream values and leaves unrelated events byte-for-byte", () => { + const rewrite = createResponsesSnapshotPayloadRewrite(); + const created = JSON.stringify({ + type: "response.created", + response: { + id: "resp_complete_shape", + created_at: 1, + model: "model", + object: "response", + status: "queued", + output: [], + parallel_tool_calls: false, + tool_choice: "none", + tools: [{ type: "function", name: "lookup" }], + }, + }); + const unrelated = JSON.stringify({ type: "response.reasoning_summary_text.delta", delta: "thinking" }); + const malformed = "{not-json}"; + + expect(JSON.parse(rewrite(created))).toEqual(JSON.parse(created)); + expect(rewrite(unrelated)).toBe(unrelated); + expect(rewrite(malformed)).toBe(malformed); + }); + + test("reports explicit provider opt-in only", () => { + expect(hasResponsesSnapshotRepair(undefined)).toBe(false); + expect(hasResponsesSnapshotRepair(false)).toBe(false); + expect(hasResponsesSnapshotRepair(true)).toBe(true); + }); +}); From d219dbc4ec1607432b207974f8168ef30c3b67be Mon Sep 17 00:00:00 2001 From: Jingwei <164893774+0xJingwei@users.noreply.github.com> Date: Sun, 2 Aug 2026 23:59:17 +0800 Subject: [PATCH 2/4] fix: harden responses snapshot repair --- .../060_impl_roadmap.md | 2 +- .../100_darwin_eager_optin.md | 25 +- .../ja/reference/configuration/providers.md | 4 +- .../ko/reference/configuration/providers.md | 4 +- .../docs/reference/configuration/providers.md | 4 +- .../ru/reference/configuration/providers.md | 4 +- .../reference/configuration/providers.md | 4 +- src/lib/bun-stream-caps.ts | 8 +- src/server/relay-eager.ts | 23 +- src/server/responses-snapshot-repair.ts | 241 ++++++++++-- src/server/responses/core.ts | 9 +- src/server/sse-payload-rewrite.ts | 31 +- src/types.ts | 3 +- tests/bun-stream-caps.test.ts | 4 + tests/relay-eager.test.ts | 28 ++ tests/responses-snapshot-repair.test.ts | 364 ++++++++++++++++++ tests/sse-payload-rewrite.test.ts | 44 +++ 17 files changed, 741 insertions(+), 61 deletions(-) diff --git a/devlog/_plan/260731_macos_rss_retention/060_impl_roadmap.md b/devlog/_plan/260731_macos_rss_retention/060_impl_roadmap.md index 4539ddbad..9387d53ce 100644 --- a/devlog/_plan/260731_macos_rss_retention/060_impl_roadmap.md +++ b/devlog/_plan/260731_macos_rss_retention/060_impl_roadmap.md @@ -34,7 +34,7 @@ the user for this unit's commits. | `070_harness_warm_fix.md` | wp2 | Apply 050's harness diff (monotonic pause clamp, child-exit/duration split) + post-patch remote smoke calibration | — | | `080_inspection_bounds.md` | wp3 | relay.ts inspector caps + clear points + disconnect cancel (both consumers) + parse-once + observability counters on `/api/system/memory` | — | | `090_eager_failed_tail.md` | wp4 | relay-eager.ts synthetic `response.failed` tail on mid-stream reset | 080 (dispose hook touches the eager producer finally) | -| `100_darwin_eager_optin.md` | wp5 | core.ts gate: darwin joins win32 for explicit `eager-relay` no-rewrite traffic; `auto` stays tee; local darwin abort-stress gate | 090 (tail must exist before opt-in is reachable) | +| `100_darwin_eager_optin.md` | wp5 | core.ts gate: darwin joins win32 for explicit `eager-relay`, including inline client rewrites; `auto` stays tee; local darwin abort-stress gate | 090 (tail must exist before opt-in is reachable) | | `110_verify_and_push.md` | wp6 | full-suite/typecheck/privacy gates, devlog closeout, push | 070–100 | Ordering rationale: 070 is independent and unblocks any future macmini re-run. diff --git a/devlog/_plan/260731_macos_rss_retention/100_darwin_eager_optin.md b/devlog/_plan/260731_macos_rss_retention/100_darwin_eager_optin.md index 52863bde8..fb4ea7fdc 100644 --- a/devlog/_plan/260731_macos_rss_retention/100_darwin_eager_optin.md +++ b/devlog/_plan/260731_macos_rss_retention/100_darwin_eager_optin.md @@ -139,8 +139,9 @@ gate execution record is authoritative. ## Not changed -- Rewrite traffic (image-gen aliases / item-id repair) stays on tee on both - platforms (`needsClientRewrite` guard intact). +- Darwin `auto` rewrite traffic stays on tee. An explicit + `streamMode: "eager-relay"` may compose a client-facing rewrite inline on + the single reader; raw inspection still receives the original bytes. - `decideEagerRelay` itself unchanged — the darwin restriction lives at the call site because it is platform policy, not runtime capability. - Linux unchanged (no opt-in until asked; smallest honest scope). @@ -159,7 +160,7 @@ extraction is preferred for clarity, not necessity. Test matrix: 2. win32 + no-rewrite + auto/known-bad → tee (unchanged). 3. darwin + no-rewrite + config-eager → eager (NEW). 4. darwin + no-rewrite + auto (even with minFixed satisfied) → tee. -5. darwin + rewrite + config-eager → tee. +5. darwin + rewrite + config-eager → eager single-reader inline rewrite. 6. linux + anything → tee. Plus one DIRECT `handleResponses` integration test gated to darwin only @@ -251,3 +252,21 @@ Confirmed by the same round: C1-1 ack chain sound, C1-2 stall detection and all-or-nothing unreachable classification internally sound, C1-3 single normalized selector contract with darwin-auto never eager and win32 unchanged. + +## 2026-08-02 client-rewrite amendment + +An opt-in Responses snapshot repair exposed a contradiction in the original +wp5 scope: keeping Darwin rewrite traffic on tee would reintroduce the second +reader and its retention cost precisely when a provider needs client-facing +compatibility repair. Explicit Darwin eager mode now permits inline payload +rewrites after raw inspection; Darwin `auto` and every Linux mode remain +unchanged. + +The handler-level regression test uses `streamMode: "eager-relay"` together +with an active snapshot rewrite and asserts the eager-path marker plus the +repaired terminal. Relay tests separately prove fragmented SSE framing, +client-only failed-tail behavior, upstream abort/cancel on rewrite failure, +and raw-terminal accounting. The 2026-08-01 abort-stress record above did not +exercise payload rewriting, so it remains evidence for the shared +single-reader transport rather than a separate rewrite-stress claim; the +feature remains explicit opt-in with `legacy-tee` as rollback. diff --git a/docs-site/src/content/docs/ja/reference/configuration/providers.md b/docs-site/src/content/docs/ja/reference/configuration/providers.md index c9875cd13..9fedb25e4 100644 --- a/docs-site/src/content/docs/ja/reference/configuration/providers.md +++ b/docs-site/src/content/docs/ja/reference/configuration/providers.md @@ -74,6 +74,7 @@ description: プロバイダー エントリ、認証、エンドポイント、 | `noPenaltyModels?` | `string[]` |存在/周波数ペナルティを拒否するモデル。 | | `parallelToolCalls?` | `boolean` |並列ツール呼び出しを切り替えます。 OpenAI Chat はデフォルトでオンになっています。非チャット アダプターは明示的な `true` でのみアドバタイズします。 | | `responsesItemIdRepair?` | `{ message?: string[]; reasoning?: string[]; repairMissingTerminalIds?: boolean }` |正確なプレースホルダー ID および欠落している端末 ID に対するダウンストリーム SSE 修復はデフォルトで無効になっています。関数呼び出し ID は決して書き換えられません。 | +| `responsesSnapshotRepair?` | `boolean` | デフォルトで無効のクライアント向け修復です。SSE と JSON の Responses ライフサイクルで欠落した status、output、ツールメタデータを補完し、raw 検査と永続化は変更しません。 | | `autoToolChoiceOnlyModels?` | `string[]` | `tool_choice` が `auto` または `none` のみを受け入れるモデル。強制的な選択は格下げされます。 | | `preserveReasoningContentModels?` | `string[]` |チャット履歴に以前のアシスタント `reasoning_content` が必要なモデル。 | | `thinkingToggleModels?` | `string[]` |エフォート ラダーではなく `thinking.enabled` を使用してモデルをチャットします。 | @@ -186,7 +187,8 @@ Anthropic アカウント ポリシーのリスクを理解していない限り "reasoning": ["rs_0"], "message": ["msg_0"], "repairMissingTerminalIds": true - } + }, + "responsesSnapshotRepair": true } } } diff --git a/docs-site/src/content/docs/ko/reference/configuration/providers.md b/docs-site/src/content/docs/ko/reference/configuration/providers.md index b3adefda8..642209aaf 100644 --- a/docs-site/src/content/docs/ko/reference/configuration/providers.md +++ b/docs-site/src/content/docs/ko/reference/configuration/providers.md @@ -74,6 +74,7 @@ description: 공급자 항목, 인증, 엔드포인트, 모델 카탈로그, 할 | `noPenaltyModels?` | `string[]` | presence/frequency penalty를 허용하지 않는 모델입니다. | | `parallelToolCalls?` | `boolean` | 병렬 도구 호출을 켜거나 끕니다. OpenAI Chat은 기본으로 켜져 있고, 비-chat 어댑터는 명시적으로 `true`일 때만 이를 노출합니다. | | `responsesItemIdRepair?` | `{ message?: string[]; reasoning?: string[]; repairMissingTerminalIds?: boolean }` | 기본값이 꺼진 downstream SSE 복구입니다. 정확한 자리표시자 id와 누락된 종료 id를 복구합니다. function-call id는 다시 쓰지 않습니다. | +| `responsesSnapshotRepair?` | `boolean` | 기본값이 꺼진 클라이언트용 복구입니다. SSE와 JSON의 Responses 수명 주기에서 누락된 status, output, 도구 메타데이터를 채우며 raw 검사와 영속화는 변경하지 않습니다. | | `autoToolChoiceOnlyModels?` | `string[]` | `tool_choice`가 `auto` 또는 `none`만 받는 모델입니다. 강제 선택은 낮은 수준으로 바뀝니다. | | `preserveReasoningContentModels?` | `string[]` | chat 기록에서 이전 assistant `reasoning_content`가 필요한 모델입니다. | | `thinkingToggleModels?` | `string[]` | effort 계층 대신 `thinking.enabled`를 쓰는 chat 모델입니다. | @@ -189,7 +190,8 @@ Anthropic 계정 정책 위험을 이해하지 못한다면 이 기능은 꺼두 "reasoning": ["rs_0"], "message": ["msg_0"], "repairMissingTerminalIds": true - } + }, + "responsesSnapshotRepair": true } } } diff --git a/docs-site/src/content/docs/reference/configuration/providers.md b/docs-site/src/content/docs/reference/configuration/providers.md index 5109a9d8e..64e246195 100644 --- a/docs-site/src/content/docs/reference/configuration/providers.md +++ b/docs-site/src/content/docs/reference/configuration/providers.md @@ -85,6 +85,7 @@ differing backup and rewrites known legacy namespaced selected ids to bare ids. | `noPenaltyModels?` | `string[]` | Models that reject presence/frequency penalties. | | `parallelToolCalls?` | `boolean` | Toggle parallel tool calls. OpenAI Chat defaults on; non-chat adapters advertise only on explicit `true`. | | `responsesItemIdRepair?` | `{ message?: string[]; reasoning?: string[]; repairMissingTerminalIds?: boolean }` | Disabled-by-default downstream SSE repair for exact placeholder ids and missing terminal ids. Function-call ids are never rewritten. | +| `responsesSnapshotRepair?` | `boolean` | Disabled-by-default client-facing repair for sparse Responses lifecycle snapshots in SSE and JSON. Fills missing canonical status, output, and tool metadata while raw inspection and persistence remain unchanged. | | `autoToolChoiceOnlyModels?` | `string[]` | Models whose `tool_choice` accepts only `auto` or `none`; forced choices are downgraded. | | `preserveReasoningContentModels?` | `string[]` | Models requiring prior assistant `reasoning_content` in chat history. | | `thinkingToggleModels?` | `string[]` | Chat models using `thinking.enabled` rather than an effort ladder. | @@ -230,7 +231,8 @@ For a broken `openai-responses` gateway, repair belongs on the provider object: "reasoning": ["rs_0"], "message": ["msg_0"], "repairMissingTerminalIds": true - } + }, + "responsesSnapshotRepair": true } } } diff --git a/docs-site/src/content/docs/ru/reference/configuration/providers.md b/docs-site/src/content/docs/ru/reference/configuration/providers.md index 6805a6e55..e2cd7606e 100644 --- a/docs-site/src/content/docs/ru/reference/configuration/providers.md +++ b/docs-site/src/content/docs/ru/reference/configuration/providers.md @@ -90,6 +90,7 @@ cross-route credential fallback не существует. Строки API GPT- | `noPenaltyModels?` | `string[]` | Модели, отвергающие penalty presence/frequency. | | `parallelToolCalls?` | `boolean` | Переключатель parallel tool call'ов. Для OpenAI Chat по умолчанию включено; не-chat adapter'ы рекламируют это только при явном `true`. | | `responsesItemIdRepair?` | `{ message?: string[]; reasoning?: string[]; repairMissingTerminalIds?: boolean }` | По умолчанию выключенная downstream SSE-repair для exact placeholder-id и отсутствующих terminal-id. Function-call id никогда не переписываются. | +| `responsesSnapshotRepair?` | `boolean` | По умолчанию выключенная клиентская repair для неполных lifecycle snapshot'ов Responses в SSE и JSON. Добавляет отсутствующие status, output и tool metadata, не меняя raw inspection и persistence. | | `autoToolChoiceOnlyModels?` | `string[]` | Модели, у которых `tool_choice` принимает только `auto` или `none`; forced choice понижается. | | `preserveReasoningContentModels?` | `string[]` | Модели, которым нужен предыдущий assistant `reasoning_content` в chat history. | | `thinkingToggleModels?` | `string[]` | Chat-модели, использующие `thinking.enabled` вместо effort-ladder. | @@ -240,7 +241,8 @@ Beijing, а `alibaba-token-plan-intl` обслуживает междунаро "reasoning": ["rs_0"], "message": ["msg_0"], "repairMissingTerminalIds": true - } + }, + "responsesSnapshotRepair": true } } } diff --git a/docs-site/src/content/docs/zh-cn/reference/configuration/providers.md b/docs-site/src/content/docs/zh-cn/reference/configuration/providers.md index 71f5b9218..f68ba7426 100644 --- a/docs-site/src/content/docs/zh-cn/reference/configuration/providers.md +++ b/docs-site/src/content/docs/zh-cn/reference/configuration/providers.md @@ -74,6 +74,7 @@ description: 提供者条目、身份验证、端点、模型目录、配额、 | `noPenaltyModels?` | `string[]` | 会拒绝 presence/frequency penalty 的模型。 | | `parallelToolCalls?` | `boolean` | 切换并行工具调用。OpenAI Chat 默认开启；非 chat 适配器只有显式 `true` 时才会声明支持。 | | `responsesItemIdRepair?` | `{ message?: string[]; reasoning?: string[]; repairMissingTerminalIds?: boolean }` | 默认关闭的下游 SSE 修复，用于精确占位 id 和缺失的终止 id。function-call id 永远不会被重写。 | +| `responsesSnapshotRepair?` | `boolean` | 默认关闭的客户端修复，用于补全 SSE 与 JSON 中稀疏 Responses 生命周期快照缺失的 status、output 和工具元数据；原始检查与持久化保持不变。 | | `autoToolChoiceOnlyModels?` | `string[]` | `tool_choice` 只接受 `auto` 或 `none` 的模型；强制选择会被降级。 | | `preserveReasoningContentModels?` | `string[]` | 需要在聊天历史中保留先前 assistant `reasoning_content` 的模型。 | | `thinkingToggleModels?` | `string[]` | 使用 `thinking.enabled` 而不是 effort 阶梯的 chat 模型。 | @@ -183,7 +184,8 @@ affinity。这些策略不能规避 provider enforcement。 "reasoning": ["rs_0"], "message": ["msg_0"], "repairMissingTerminalIds": true - } + }, + "responsesSnapshotRepair": true } } } diff --git a/src/lib/bun-stream-caps.ts b/src/lib/bun-stream-caps.ts index 6ed7efa8d..024257604 100644 --- a/src/lib/bun-stream-caps.ts +++ b/src/lib/bun-stream-caps.ts @@ -92,10 +92,10 @@ export function decideEagerRelay( * Apply the two-platform eager-relay policy to the runtime/config capability. * 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 or a Darwin - * non-config-eager mode selects tee. Windows rewrites remain on tee because the extra JS pull - * wrapper is part of the Bun#32111 crash shape; Darwin's explicit eager mode may compose a - * client-facing rewrite after the single-reader relay. + * Returns the normalized effective decision, or null when platform policy, a Windows rewrite, + * or a Darwin non-config-eager mode excludes this path. Windows rewrites are forced onto the + * eager single-reader path separately by `isWin32EagerRewrite`; Darwin's explicit eager mode may + * compose a client-facing rewrite after the single-reader relay. */ export function selectEagerPath( platform: NodeJS.Platform, diff --git a/src/server/relay-eager.ts b/src/server/relay-eager.ts index e208c0920..2ccb4e9a6 100644 --- a/src/server/relay-eager.ts +++ b/src/server/relay-eager.ts @@ -161,6 +161,10 @@ export function relaySseEagerBounded( let queuedBytes = 0; let cancelled = false; let done = false; + // Raw inspection can observe a terminal before an inline rewrite succeeds. + // Track client delivery separately so a rewrite failure cannot leave the + // client with an empty/truncated stream while persistence records success. + let clientTerminalEnqueued = false; // Pause gate: resolved by client pull, client cancel, or upstream abort so a // paused producer ALWAYS resumes (audit blocker 2 — no deadlock; onDone and // turn unregistration stay reachable, drainAndShutdown never hangs). @@ -192,6 +196,7 @@ export function relaySseEagerBounded( const producer = async () => { let syntheticKind: "incomplete" | "failed" | null = null; + let clientRelayFailed = false; // reader.read() is not intrinsically tied to the upstream AbortController // (a fetch body usually rejects on abort, but that coupling is the fetch // implementation's, not the stream's). Race every read against the abort @@ -212,7 +217,10 @@ export function relaySseEagerBounded( const tail = flushRewriteTail(); if (tail.byteLength > 0 && !cancelled) { queuedBytes += tail.byteLength; - try { controllerRef?.enqueue(tail); } catch { /* client already gone */ } + try { + controllerRef?.enqueue(tail); + if (hooks.sawTerminal()) clientTerminalEnqueued = true; + } catch { /* client already gone */ } } } if (!hooks.sawTerminal() && !cancelled && !upstream.signal.aborted) { @@ -235,6 +243,7 @@ export function relaySseEagerBounded( queuedBytes += outbound.byteLength; try { controllerRef?.enqueue(outbound); + if (hooks.sawTerminal()) clientTerminalEnqueued = true; } catch { // Controller already torn down (client went away without cancel()). cancelled = true; @@ -249,7 +258,7 @@ export function relaySseEagerBounded( } catch (err) { // Upstream read failure. Distinguish genuine mid-stream reset from // abort-driven teardown (shutdown/cancel-expiry) — audit M3. - if (!hooks.sawTerminal() && !cancelled && !upstream.signal.aborted) { + if (!clientTerminalEnqueued && !cancelled && !upstream.signal.aborted) { // Serializing `err` can run user-defined accessors (Error.message // getters, toString) that re-entrantly cancel the client or abort the // upstream. Build the tail FIRST, then re-check eligibility before @@ -257,8 +266,11 @@ export function relaySseEagerBounded( const tail = new TextEncoder().encode( `\n\nevent: response.failed\ndata: ${buildFailedTailPayload(err)}\n\ndata: [DONE]\n\n`, ); - if (!hooks.sawTerminal() && !cancelled && !upstream.signal.aborted) { - syntheticKind = "failed"; + if (!clientTerminalEnqueued && !cancelled && !upstream.signal.aborted) { + // If raw inspection already recorded a real terminal, the failed + // tail is client-only. Do not overwrite raw terminal accounting. + if (!hooks.sawTerminal()) syntheticKind = "failed"; + clientRelayFailed = true; queuedBytes += tail.byteLength; try { controllerRef?.enqueue(tail); } catch { /* client already torn down */ } try { controllerRef?.close(); } catch { /* client already torn down */ } @@ -276,7 +288,7 @@ export function relaySseEagerBounded( if (cancelled && !hooks.sawTerminal()) { hooks.onClientCancel(); } - if (cancelled || upstream.signal.aborted || syntheticKind === "failed") { + if (cancelled || upstream.signal.aborted || syntheticKind === "failed" || clientRelayFailed) { upstream.abort(); reader.cancel().catch(() => {}); } @@ -284,6 +296,7 @@ export function relaySseEagerBounded( try { controllerRef?.close(); } catch { /* already closed/errored */ } } try { hooks.disposeInspection?.(); } catch { /* inspection teardown must not block lifecycle cleanup */ } + try { rewrite?.dispose?.(); } catch { /* rewrite teardown must not block lifecycle cleanup */ } fireDone(); } }; diff --git a/src/server/responses-snapshot-repair.ts b/src/server/responses-snapshot-repair.ts index 17d3014b3..dd666118e 100644 --- a/src/server/responses-snapshot-repair.ts +++ b/src/server/responses-snapshot-repair.ts @@ -1,3 +1,8 @@ +import type { TranslatorBudget } from "../lib/translator-budget"; +import { + MAX_COMPLETED_OUTPUT_ITEMS, + MAX_COMPLETED_OUTPUT_ITEM_SOURCE_BYTES, +} from "./relay"; import type { SsePayloadRewrite } from "./sse-payload-rewrite"; const RESPONSE_EVENT_STATUSES: Readonly> = { @@ -9,13 +14,47 @@ const RESPONSE_EVENT_STATUSES: Readonly> = { "response.queued": "queued", }; +type RequestDefaults = { + parallelToolCalls: boolean; + toolChoice: unknown; + tools: unknown[]; +}; + +type RetainedOutputItem = { + item: Record; + sourceBytes: number; +}; + function isPlainObject(value: unknown): value is Record { return !!value && typeof value === "object" && !Array.isArray(value); } +function isStructurallyValidToolChoice(value: unknown): boolean { + return (typeof value === "string" && value.trim().length > 0) + || (isPlainObject(value) && typeof value.type === "string" && value.type.trim().length > 0); +} + +function requestDefaults(requestBody: unknown): RequestDefaults { + const request = isPlainObject(requestBody) ? requestBody : {}; + return { + parallelToolCalls: typeof request.parallel_tool_calls === "boolean" + ? request.parallel_tool_calls + : true, + toolChoice: isStructurallyValidToolChoice(request.tool_choice) ? request.tool_choice : "auto", + tools: Array.isArray(request.tools) ? request.tools : [], + }; +} + function repairOutputTextPart(part: Record): Record { - if (part.type !== "output_text" || Array.isArray(part.annotations)) return part; - return { ...part, annotations: [] }; + if (part.type !== "output_text") return part; + const needsText = typeof part.text !== "string"; + const needsAnnotations = !Array.isArray(part.annotations); + if (!needsText && !needsAnnotations) return part; + return { + ...part, + ...(needsText ? { text: "" } : {}), + ...(needsAnnotations ? { annotations: [] } : {}), + }; } function repairSummaryPart(part: Record): Record { @@ -23,38 +62,56 @@ function repairSummaryPart(part: Record): Record): Record { +function repairOutputItem( + item: Record, + inferredStatus?: string, +): Record { + let repaired = item; + let changed = false; + if (item.type === "reasoning") { - return Array.isArray(item.summary) ? item : { ...item, summary: [] }; + const rawSummary = item.summary; + const summary = Array.isArray(rawSummary) + ? rawSummary.map((part) => isPlainObject(part) ? repairSummaryPart(part) : part) + : []; + changed = !Array.isArray(rawSummary) + || summary.some((part, index) => part !== rawSummary[index]); + if (changed) repaired = { ...repaired, summary }; + } else if (item.type === "message") { + const rawContent = item.content; + const content = Array.isArray(rawContent) + ? rawContent.map((part) => isPlainObject(part) ? repairOutputTextPart(part) : part) + : []; + changed = !Array.isArray(rawContent) + || content.some((part, index) => part !== rawContent[index]); + const role = item.role === "assistant" ? item.role : "assistant"; + changed = changed || role !== item.role; + if (changed) repaired = { ...repaired, content, role }; } - if (item.type !== "message") return item; - let changed = false; - const content = Array.isArray(item.content) - ? item.content.map((part) => { - if (!isPlainObject(part)) return part; - const repaired = repairOutputTextPart(part); - changed = changed || repaired !== part; - return repaired; - }) - : []; - changed = changed || !Array.isArray(item.content); - const role = item.role === "assistant" ? item.role : "assistant"; - changed = changed || role !== item.role; - return changed ? { ...item, content, role } : item; + if (inferredStatus && typeof repaired.status !== "string") { + repaired = { ...repaired, status: inferredStatus }; + } + return repaired; } function repairResponseSnapshot( response: Record, defaultStatus: string, + defaults: RequestDefaults, + reconstructedOutput?: Record[], ): Record { const repaired = { ...response }; let changed = false; + const outputStatus = defaultStatus === "completed" ? "completed" : undefined; - if (Array.isArray(repaired.output)) { + if (reconstructedOutput) { + repaired.output = reconstructedOutput; + changed = true; + } else if (Array.isArray(repaired.output)) { const output = repaired.output.map((item) => { if (!isPlainObject(item)) return item; - const next = repairOutputItem(item); + const next = repairOutputItem(item, outputStatus); changed = changed || next !== item; return next; }); @@ -64,15 +121,15 @@ function repairResponseSnapshot( changed = true; } if (typeof repaired.parallel_tool_calls !== "boolean") { - repaired.parallel_tool_calls = true; + repaired.parallel_tool_calls = defaults.parallelToolCalls; changed = true; } - if (repaired.tool_choice === undefined || repaired.tool_choice === null) { - repaired.tool_choice = "auto"; + if (!isStructurallyValidToolChoice(repaired.tool_choice)) { + repaired.tool_choice = defaults.toolChoice; changed = true; } if (!Array.isArray(repaired.tools)) { - repaired.tools = []; + repaired.tools = defaults.tools; changed = true; } if (typeof repaired.status !== "string") { @@ -88,8 +145,64 @@ function repairResponseSnapshot( * Existing upstream values remain authoritative; only absent or structurally invalid fields are * backfilled. */ -export function createResponsesSnapshotPayloadRewrite(): SsePayloadRewrite { - return (payload) => { +export function createResponsesSnapshotPayloadRewrite( + requestBody?: unknown, + budget?: TranslatorBudget, +): SsePayloadRewrite { + const defaults = requestDefaults(requestBody); + const completedItems = new Map(); + const encoder = new TextEncoder(); + let aggregateItemBytes = 0; + let reconstructionTainted = false; + + const clearCompletedItems = (): void => { + if (aggregateItemBytes > 0) { + budget?.releaseRetained(aggregateItemBytes, { kind: "retained_collectors" }); + } + completedItems.clear(); + aggregateItemBytes = 0; + reconstructionTainted = false; + }; + + const retainCompletedItem = ( + index: number, + item: Record, + sourceBytes: number, + ): void => { + const previous = completedItems.get(index); + if (sourceBytes > MAX_COMPLETED_OUTPUT_ITEM_SOURCE_BYTES) { + if (previous) { + completedItems.delete(index); + aggregateItemBytes -= previous.sourceBytes; + budget?.releaseRetained(previous.sourceBytes, { kind: "retained_collectors" }); + } + reconstructionTainted = true; + return; + } + const retainedDelta = sourceBytes - (previous?.sourceBytes ?? 0); + if (retainedDelta > 0) { + budget?.chargeRetained(retainedDelta, { kind: "retained_collectors" }); + } else if (retainedDelta < 0) { + budget?.releaseRetained(-retainedDelta, { kind: "retained_collectors" }); + } + completedItems.set(index, { item, sourceBytes }); + aggregateItemBytes += retainedDelta; + while (completedItems.size > MAX_COMPLETED_OUTPUT_ITEMS + || aggregateItemBytes > MAX_COMPLETED_OUTPUT_ITEM_SOURCE_BYTES) { + let highestIndex = -1; + for (const retainedIndex of completedItems.keys()) { + if (retainedIndex > highestIndex) highestIndex = retainedIndex; + } + const evicted = completedItems.get(highestIndex); + if (!evicted) break; + completedItems.delete(highestIndex); + aggregateItemBytes -= evicted.sourceBytes; + budget?.releaseRetained(evicted.sourceBytes, { kind: "retained_collectors" }); + reconstructionTainted = true; + } + }; + + const rewrite = ((payload: string): string => { let event: unknown; try { event = JSON.parse(payload); @@ -101,24 +214,59 @@ export function createResponsesSnapshotPayloadRewrite(): SsePayloadRewrite { let nextEvent = event; let changed = false; - const responseStatus = RESPONSE_EVENT_STATUSES[type]; - if (responseStatus && isPlainObject(event.response)) { - const response = repairResponseSnapshot(event.response, responseStatus); - if (response !== event.response) { - nextEvent = { ...nextEvent, response }; - changed = true; - } - } - if ((type === "response.output_item.added" || type === "response.output_item.done") && isPlainObject(event.item)) { - const item = repairOutputItem(event.item); + const itemStatus = type === "response.output_item.done" ? "completed" : "in_progress"; + const item = repairOutputItem(event.item, itemStatus); if (item !== event.item) { nextEvent = { ...nextEvent, item }; changed = true; } + if (type === "response.output_item.done") { + if (Number.isInteger(event.output_index) + && (event.output_index as number) >= 0 + && typeof item.type === "string") { + retainCompletedItem( + event.output_index as number, + item, + encoder.encode(JSON.stringify(item)).byteLength, + ); + } else { + reconstructionTainted = true; + } + } } + const responseStatus = Object.prototype.hasOwnProperty.call(RESPONSE_EVENT_STATUSES, type) + ? RESPONSE_EVENT_STATUSES[type] + : undefined; + if (responseStatus && isPlainObject(event.response)) { + const hasAuthoritativeOutput = Array.isArray(event.response.output) + && event.response.output.length > 0; + const reconstructedOutput = type === "response.completed" + && !hasAuthoritativeOutput + && !reconstructionTainted + && completedItems.size > 0 + ? [...completedItems.entries()] + .sort(([left], [right]) => left - right) + .map(([, retained]) => retained.item) + : undefined; + const response = repairResponseSnapshot( + event.response, + responseStatus, + defaults, + reconstructedOutput, + ); + if (response !== event.response) { + nextEvent = { ...nextEvent, response }; + changed = true; + } + } + + const clearsCompletedItems = type === "response.completed" + || type === "response.failed" + || type === "response.incomplete"; + if ((type === "response.content_part.added" || type === "response.content_part.done") && isPlainObject(event.part)) { const part = repairOutputTextPart(event.part); @@ -143,8 +291,25 @@ export function createResponsesSnapshotPayloadRewrite(): SsePayloadRewrite { changed = true; } - return changed ? JSON.stringify(nextEvent) : payload; - }; + const result = changed ? JSON.stringify(nextEvent) : payload; + if (clearsCompletedItems) clearCompletedItems(); + return result; + }) as SsePayloadRewrite; + rewrite.dispose = clearCompletedItems; + return rewrite; +} + +/** Repair a non-streaming Responses JSON object without changing the raw persisted payload. */ +export function repairResponsesSnapshotJson(payload: string, requestBody?: unknown): string { + let response: unknown; + try { + response = JSON.parse(payload); + } catch { + return payload; + } + if (!isPlainObject(response)) return payload; + const repaired = repairResponseSnapshot(response, "completed", requestDefaults(requestBody)); + return repaired === response ? payload : JSON.stringify(repaired); } export function hasResponsesSnapshotRepair(enabled: boolean | undefined): enabled is true { diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 40c7bb5fd..1e5c510f7 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -153,6 +153,7 @@ import { import { createResponsesSnapshotPayloadRewrite, hasResponsesSnapshotRepair, + repairResponsesSnapshotJson, } from "../responses-snapshot-repair"; import { createImageGenCallRestoreRewrite, @@ -1770,7 +1771,7 @@ async function handleResponsesInner( ? createResponsesItemIdPayloadRewrite(repairConfig!, translatorBudget) : undefined, hasResponsesSnapshotRepair(snapshotRepair) - ? createResponsesSnapshotPayloadRewrite() + ? createResponsesSnapshotPayloadRewrite(parsed._rawBody, translatorBudget) : undefined, ].filter((rewrite): rewrite is NonNullable => rewrite !== undefined); const needsClientRewrite = payloadRewrites.length > 0; @@ -1930,7 +1931,11 @@ async function handleResponsesInner( rememberPassthroughResponse(JSON.parse(text) as { id?: unknown; output?: unknown; status?: unknown }); } catch { /* non-JSON despite content-type; recording is best-effort */ } } - return new Response(restoreImageGenCallsInJson(text, imageGenCallAliases), { + const restoredText = restoreImageGenCallsInJson(text, imageGenCallAliases); + const clientText = hasResponsesSnapshotRepair(route.provider.responsesSnapshotRepair) + ? repairResponsesSnapshotJson(restoredText, parsed._rawBody) + : restoredText; + return new Response(clientText, { status: upstreamResponse.status, statusText: upstreamResponse.statusText, headers, diff --git a/src/server/sse-payload-rewrite.ts b/src/server/sse-payload-rewrite.ts index 712853cee..693e382d7 100644 --- a/src/server/sse-payload-rewrite.ts +++ b/src/server/sse-payload-rewrite.ts @@ -7,7 +7,10 @@ import type { TranslatorBudget } from "../lib/translator-budget"; * parse/stringify pass so a tee'd stream is not re-framed twice per event. */ -export type SsePayloadRewrite = (payload: string) => string; +export type SsePayloadRewrite = ((payload: string) => string) & { + /** Release state retained across events when the stream ends or is cancelled. */ + dispose?: () => void; +}; /** Split one complete SSE event block while retaining its original blank-line delimiter. */ export function nextSseBlock(buffer: string): { block: string; delimiter: string; rest: string } | null { @@ -54,11 +57,26 @@ export function replaceSseDataPayload(block: string, payload: string): string { export function composeSsePayloadRewrites(...rewrites: SsePayloadRewrite[]): SsePayloadRewrite { if (rewrites.length === 0) return (payload) => payload; if (rewrites.length === 1) return rewrites[0]!; - return (payload) => { + const composed = ((payload: string): string => { let next = payload; for (const rewrite of rewrites) next = rewrite(next); return next; + }) as SsePayloadRewrite; + let disposed = false; + composed.dispose = () => { + if (disposed) return; + disposed = true; + let firstError: unknown; + for (const rewrite of rewrites) { + try { + rewrite.dispose?.(); + } catch (error) { + firstError ??= error; + } + } + if (firstError !== undefined) throw firstError; }; + return composed; } /** @@ -122,6 +140,12 @@ export function relaySseWithPayloadRewrite( buffer = ""; bufferBytes = 0; }; + let rewriteDisposed = false; + const disposeRewrite = (): void => { + if (rewriteDisposed) return; + rewriteDisposed = true; + try { rewrite.dispose?.(); } catch { /* rewrite teardown must not block stream cleanup */ } + }; const emitProcessedBlocks = ( controller: ReadableStreamDefaultController, @@ -156,6 +180,7 @@ export function relaySseWithPayloadRewrite( appendBuffer(decoder.decode()); emitProcessedBlocks(controller, true); releaseBuffer(); + disposeRewrite(); controller.close(); return; } @@ -163,12 +188,14 @@ export function relaySseWithPayloadRewrite( emitProcessedBlocks(controller); } catch (error) { releaseBuffer(); + disposeRewrite(); try { await reader.cancel(error); } catch { /* already closed */ } controller.error(error); } }, cancel(reason) { releaseBuffer(); + disposeRewrite(); reader.cancel(reason).catch(() => {}); }, }); diff --git a/src/types.ts b/src/types.ts index f0589dc73..f48cb5f28 100644 --- a/src/types.ts +++ b/src/types.ts @@ -1084,7 +1084,8 @@ export interface OcxProviderConfig { /** * Provider-local repair for Responses gateways whose response.created / response.in_progress * snapshots omit required canonical fields such as output, tools, and tool_choice. - * Disabled by default and applied only to the client-facing SSE branch. + * Disabled by default and applied only to client-facing SSE/JSON; raw inspection and + * persistence remain authoritative. */ responsesSnapshotRepair?: boolean; /** Model ids whose tool_choice only accepts `auto` or `none`; forced/named choices are downgraded. */ diff --git a/tests/bun-stream-caps.test.ts b/tests/bun-stream-caps.test.ts index 493bbeda9..956c0ff68 100644 --- a/tests/bun-stream-caps.test.ts +++ b/tests/bun-stream-caps.test.ts @@ -127,6 +127,10 @@ describe("selectEagerPath (platform policy matrix)", () => { .toEqual({ useEagerRelay: true, reason: "config-eager" }); }); + test("darwin + rewrite + auto known-bad runtime → tee", () => { + expect(selectEagerPath("darwin", true, "auto", "1.3.14", null)).toBeNull(); + }); + test("win32 + rewrite + config-eager → tee", () => { expect(selectEagerPath("win32", true, "eager-relay", "1.3.14", null)).toBeNull(); }); diff --git a/tests/relay-eager.test.ts b/tests/relay-eager.test.ts index ccbd2c2e9..9c2b2b56c 100644 --- a/tests/relay-eager.test.ts +++ b/tests/relay-eager.test.ts @@ -67,8 +67,10 @@ function controlledUpstream(): { close: () => void; fail: (err: Error) => void; pullCount: () => number; + cancelCount: () => number; } { let pulls = 0; + let cancels = 0; const pending: Array<{ resolve: (r: ReadableStreamReadResult) => void }> = []; const queue: Array<{ kind: "chunk"; value: Uint8Array } | { kind: "close" } | { kind: "fail"; err: Error }> = []; const controllerQueue: Uint8Array[] = []; @@ -84,6 +86,7 @@ function controlledUpstream(): { const stream = new ReadableStream({ start(controller) { controllerRef = controller; flush(); }, pull() { pulls += 1; }, + cancel() { cancels += 1; }, }, { highWaterMark: 0 }); return { stream, @@ -91,6 +94,7 @@ function controlledUpstream(): { close: () => { closed = true; flush(); }, fail: err => { failure = err; flush(); }, pullCount: () => pulls, + cancelCount: () => cancels, }; } @@ -493,6 +497,30 @@ describe("relaySseEagerBounded — #44 cancel semantics", () => { }); describe("relaySseEagerBounded — error paths", () => { + test("rewrite overflow after a raw terminal emits a client failed tail", async () => { + const { hooks, rec } = makeHooks(); + hooks.rewritePayload = payload => payload; + const budget = createTranslatorBudget({ maxTurnBytes: 32 }); + const up = controlledUpstream(); + const upstreamAc = new AbortController(); + const relayed = relaySseEagerBounded(up.stream, upstreamAc, hooks, { + rewriteBudget: budget, + }); + up.push(sse(COMPLETED)); + + const out = await readAll(relayed); + await settle(); + expect(rec.terminals).toEqual([{ status: "completed", httpStatus: undefined }]); + expect(out).toContain(FAILED_EVENT_MARKER); + expect(out).toContain("data: [DONE]"); + expect(rec.synthetics).toEqual([]); + expect(rec.dones).toBe(1); + expect(upstreamAc.signal.aborted).toBe(true); + expect(up.cancelCount()).toBe(1); + expect(budget.snapshot().currentBytes).toBe(0); + budget.dispose(); + }); + test("(e/090-1) mid-stream upstream error → clean failed tail + onSynthetic/onDone once", async () => { const { hooks, rec } = makeHooks(); const inspectChunk = hooks.inspectChunk; diff --git a/tests/responses-snapshot-repair.test.ts b/tests/responses-snapshot-repair.test.ts index 3d95ce49c..d49fd0e91 100644 --- a/tests/responses-snapshot-repair.test.ts +++ b/tests/responses-snapshot-repair.test.ts @@ -3,6 +3,18 @@ import { createResponsesSnapshotPayloadRewrite, hasResponsesSnapshotRepair, } from "../src/server/responses-snapshot-repair"; +import { handleResponses } from "../src/server/responses"; +import { createTranslatorBudget } from "../src/lib/translator-budget"; +import { isEagerRelaySseResponse } from "../src/server/relay"; +import type { OcxConfig } from "../src/types"; + +function parsedSseEvents(text: string): Array> { + return text.split(/\r?\n\r?\n/) + .map(block => block.split(/\r?\n/).find(line => line.startsWith("data:"))) + .map(line => line?.slice(5).trim()) + .filter((payload): payload is string => !!payload && payload !== "[DONE]") + .map(payload => JSON.parse(payload) as Record); +} describe("Responses passthrough sparse-snapshot repair", () => { test("backfills canonical lifecycle fields missing from created and in-progress snapshots", () => { @@ -88,6 +100,230 @@ describe("Responses passthrough sparse-snapshot repair", () => { } }); + test("backfills both required output_text fields when a sparse gateway omits them", () => { + const rewrite = createResponsesSnapshotPayloadRewrite(); + const event = JSON.parse(rewrite(JSON.stringify({ + type: "response.content_part.added", + part: { type: "output_text" }, + }))) as { part: Record }; + + expect(event.part).toEqual({ type: "output_text", text: "", annotations: [] }); + }); + + test("repairs output-item status and nested reasoning summary parts", () => { + const rewrite = createResponsesSnapshotPayloadRewrite(); + for (const invalidStatus of [undefined, null, 42]) { + const added = JSON.parse(rewrite(JSON.stringify({ + type: "response.output_item.added", + output_index: 0, + item: { + id: "msg_1", + type: "message", + ...(invalidStatus === undefined ? {} : { status: invalidStatus }), + }, + }))) as { item: Record }; + expect(added.item.status).toBe("in_progress"); + } + const done = JSON.parse(rewrite(JSON.stringify({ + type: "response.output_item.done", + output_index: 1, + item: { id: "rs_1", type: "reasoning", summary: [{ type: "summary_text" }] }, + }))) as { item: { status?: unknown; summary?: unknown } }; + + expect(done.item.status).toBe("completed"); + expect(done.item.summary).toEqual([{ type: "summary_text", text: "" }]); + }); + + test("preserves protocol-valid response and output-item statuses", () => { + for (const status of ["queued", "in_progress", "completed", "failed", "incomplete", "cancelled", "future_status"]) { + const rewrite = createResponsesSnapshotPayloadRewrite(); + const event = JSON.parse(rewrite(JSON.stringify({ + type: "response.created", + response: { id: "resp_status", object: "response", status }, + }))) as { response: Record }; + expect(event.response.status).toBe(status); + } + for (const status of ["in_progress", "completed", "incomplete"]) { + const rewrite = createResponsesSnapshotPayloadRewrite(); + const event = JSON.parse(rewrite(JSON.stringify({ + type: "response.output_item.added", + item: { id: "msg_status", type: "message", status }, + }))) as { item: Record }; + expect(event.item.status).toBe(status); + } + for (const item of [ + { type: "web_search_call", status: "searching" }, + { type: "image_generation_call", status: "generating" }, + { type: "code_interpreter_call", status: "interpreting" }, + ]) { + const rewrite = createResponsesSnapshotPayloadRewrite(); + const event = JSON.parse(rewrite(JSON.stringify({ + type: "response.output_item.added", + item, + }))) as { item: Record }; + expect(event.item.status).toBe(item.status); + } + }); + + test("replaces structurally invalid response statuses with the event status", () => { + for (const invalidStatus of [null, 42]) { + const rewrite = createResponsesSnapshotPayloadRewrite(); + const event = JSON.parse(rewrite(JSON.stringify({ + type: "response.completed", + response: { id: "resp_invalid_status", object: "response", status: invalidStatus }, + }))) as { response: Record }; + expect(event.response.status).toBe("completed"); + } + }); + + test("does not invent nested statuses for a mixed in-progress snapshot", () => { + const rewrite = createResponsesSnapshotPayloadRewrite(); + const event = JSON.parse(rewrite(JSON.stringify({ + type: "response.in_progress", + response: { + id: "resp_mixed", + object: "response", + output: [ + { id: "msg_done", type: "message", status: "completed" }, + { id: "msg_active", type: "message" }, + ], + }, + }))) as { response: { output: Array> } }; + + expect(event.response.output[0]?.status).toBe("completed"); + expect(event.response.output[1]?.status).toBeUndefined(); + }); + + test("reconstructs a sparse completed output from prior done items", () => { + for (const sparseOutput of [undefined, null, [], "invalid"]) { + const rewrite = createResponsesSnapshotPayloadRewrite(); + rewrite(JSON.stringify({ + type: "response.output_item.done", + output_index: 0, + item: { + id: "msg_1", + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "hello" }], + }, + })); + const terminal = JSON.parse(rewrite(JSON.stringify({ + type: "response.completed", + response: { + id: "resp_1", + object: "response", + status: "completed", + ...(sparseOutput === undefined ? {} : { output: sparseOutput }), + }, + }))) as { response: { output?: unknown } }; + + expect(terminal.response.output).toEqual([{ + id: "msg_1", + type: "message", + role: "assistant", + status: "completed", + content: [{ type: "output_text", text: "hello", annotations: [] }], + }]); + } + }); + + test("uses request tool metadata when sparse snapshots omit or corrupt it", () => { + const requestDefaults = { + parallel_tool_calls: false, + tool_choice: { type: "function", name: "lookup" }, + tools: [{ type: "function", name: "lookup", parameters: {} }], + }; + const rewrite = createResponsesSnapshotPayloadRewrite(requestDefaults); + + for (const invalidChoice of [undefined, null, 42, "", " ", [], {}, { type: "" }, { type: 42 }]) { + const response = JSON.parse(rewrite(JSON.stringify({ + type: "response.created", + response: { + id: "resp_sparse", + object: "response", + ...(invalidChoice === undefined ? {} : { tool_choice: invalidChoice }), + }, + }))) as { response: Record }; + expect(response.response.parallel_tool_calls).toBe(false); + expect(response.response.tool_choice).toEqual(requestDefaults.tool_choice); + expect(response.response.tools).toEqual(requestDefaults.tools); + } + + const upstreamChoice = { type: "function", name: "upstream-choice" }; + const preserved = JSON.parse(rewrite(JSON.stringify({ + type: "response.created", + response: { + id: "resp_complete", + object: "response", + parallel_tool_calls: true, + tool_choice: upstreamChoice, + tools: [{ type: "function", name: "upstream-choice" }], + }, + }))) as { response: Record }; + expect(preserved.response.parallel_tool_calls).toBe(true); + expect(preserved.response.tool_choice).toEqual(upstreamChoice); + expect(preserved.response.tools).toEqual([{ type: "function", name: "upstream-choice" }]); + + const futureChoice = { type: "web_search_preview_2025_03_11" }; + const futureRewrite = createResponsesSnapshotPayloadRewrite({ tool_choice: futureChoice }); + const future = JSON.parse(futureRewrite(JSON.stringify({ + type: "response.created", + response: { id: "resp_future", object: "response", tool_choice: futureChoice }, + }))) as { response: Record }; + expect(future.response.tool_choice).toEqual(futureChoice); + }); + + test("ignores inherited object keys as lifecycle event types", () => { + const rewrite = createResponsesSnapshotPayloadRewrite(); + const payload = JSON.stringify({ type: "__proto__", response: { id: "resp_proto" } }); + expect(rewrite(payload)).toBe(payload); + }); + + test("charges retained output items and releases them on terminal or dispose", () => { + for (const end of ["terminal", "dispose"] as const) { + const budget = createTranslatorBudget(); + const rewrite = createResponsesSnapshotPayloadRewrite(undefined, budget); + const retainedEvent = JSON.parse(rewrite(JSON.stringify({ + type: "response.output_item.done", + output_index: 0, + item: { + id: "msg_budget", + type: "message", + content: [{ type: "output_text", text: "x".repeat(4096) }], + }, + }))) as { item: Record }; + const retainedItemBytes = new TextEncoder().encode(JSON.stringify(retainedEvent.item)).byteLength; + expect(budget.snapshot().currentBytes).toBeGreaterThanOrEqual(retainedItemBytes); + if (end === "terminal") { + rewrite(JSON.stringify({ + type: "response.completed", + response: { id: "resp_budget", object: "response" }, + })); + } else { + rewrite.dispose?.(); + } + expect(budget.snapshot().currentBytes).toBe(0); + budget.dispose(); + } + }); + + test("does not reconstruct a partial terminal after the retained-item cap is exceeded", () => { + const rewrite = createResponsesSnapshotPayloadRewrite(); + for (let outputIndex = 0; outputIndex <= 256; outputIndex += 1) { + rewrite(JSON.stringify({ + type: "response.output_item.done", + output_index: outputIndex, + item: { id: `msg_${outputIndex}`, type: "message" }, + })); + } + const terminal = JSON.parse(rewrite(JSON.stringify({ + type: "response.completed", + response: { id: "resp_capped", object: "response" }, + }))) as { response: { output?: unknown } }; + + expect(terminal.response.output).toEqual([]); + }); + test("preserves upstream values and leaves unrelated events byte-for-byte", () => { const rewrite = createResponsesSnapshotPayloadRewrite(); const created = JSON.stringify({ @@ -117,4 +353,132 @@ describe("Responses passthrough sparse-snapshot repair", () => { expect(hasResponsesSnapshotRepair(false)).toBe(false); expect(hasResponsesSnapshotRepair(true)).toBe(true); }); + + test("handleResponses keeps the repair default-off and applies an explicit opt-in", async () => { + const savedFetch = globalThis.fetch; + const doneItem = { + id: "msg_1", + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "hello" }], + }; + const upstream = [ + `event: response.output_item.done\ndata: ${JSON.stringify({ + type: "response.output_item.done", + output_index: 0, + item: doneItem, + })}\n\n`, + `event: response.completed\ndata: ${JSON.stringify({ + type: "response.completed", + response: { id: "resp_1", object: "response", status: "completed" }, + })}\n\n`, + "data: [DONE]\n\n", + ].join(""); + globalThis.fetch = (async () => new Response(upstream, { + headers: { "content-type": "text/event-stream" }, + })) as typeof fetch; + try { + for (const enabled of [false, true]) { + const config = { + port: 0, + streamMode: "eager-relay", + defaultProvider: "fixture", + providers: { + fixture: { + adapter: "openai-responses", + baseUrl: "https://fixture.test/v1", + authMode: "key", + apiKey: "fixture-key", + ...(enabled ? { 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: "fixture/example-model", + stream: true, + input: "hello", + parallel_tool_calls: false, + tool_choice: { type: "function", name: "lookup" }, + tools: [{ type: "function", name: "lookup", parameters: {} }], + }), + }), config, { model: "", provider: "" }); + const events = parsedSseEvents(await response.text()); + expect(isEagerRelaySseResponse(response)).toBe( + process.platform === "darwin" || process.platform === "win32", + ); + const terminal = events.find(event => event.type === "response.completed") as { + response: Record; + } | undefined; + if (!enabled) { + expect(terminal?.response).toEqual({ id: "resp_1", object: "response", status: "completed" }); + continue; + } + expect(terminal?.response.parallel_tool_calls).toBe(false); + expect(terminal?.response.tool_choice).toEqual({ type: "function", name: "lookup" }); + expect(terminal?.response.tools).toEqual([{ type: "function", name: "lookup", parameters: {} }]); + expect(terminal?.response.output).toEqual([{ + ...doneItem, + status: "completed", + content: [{ type: "output_text", text: "hello", annotations: [] }], + }]); + } + } finally { + globalThis.fetch = savedFetch; + } + }); + + test("handleResponses repairs non-streaming JSON only for an opted-in provider", async () => { + const savedFetch = globalThis.fetch; + globalThis.fetch = (async () => new Response(JSON.stringify({ + id: "resp_json", + object: "response", + output: [{ id: "msg_json", type: "message", content: [{ type: "output_text" }] }], + }), { headers: { "content-type": "application/json" } })) as typeof fetch; + + try { + for (const enabled of [false, true]) { + const config = { + port: 0, + defaultProvider: "fixture", + providers: { + fixture: { + adapter: "openai-responses", + baseUrl: "https://fixture.test/v1", + authMode: "key", + apiKey: "fixture-key", + ...(enabled ? { 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: "fixture/example-model", input: "hello" }), + }), config, { model: "", provider: "" }); + const body = await response.json() as { status?: unknown; output: Array> }; + if (!enabled) { + expect(body.status).toBeUndefined(); + expect(body.output[0]).toEqual({ + id: "msg_json", + type: "message", + content: [{ type: "output_text" }], + }); + continue; + } + expect(body.status).toBe("completed"); + expect(body.output[0]).toEqual({ + id: "msg_json", + type: "message", + role: "assistant", + status: "completed", + content: [{ type: "output_text", text: "", annotations: [] }], + }); + } + } finally { + globalThis.fetch = savedFetch; + } + }); }); diff --git a/tests/sse-payload-rewrite.test.ts b/tests/sse-payload-rewrite.test.ts index c6c72f615..3c86380f6 100644 --- a/tests/sse-payload-rewrite.test.ts +++ b/tests/sse-payload-rewrite.test.ts @@ -103,6 +103,50 @@ describe("SSE payload rewrite composition", () => { expect(composeSsePayloadRewrites()('{"a":1}')).toBe('{"a":1}'); }); + test("composed rewrite teardown disposes every child exactly once", async () => { + const disposals = [0, 0]; + const first = Object.assign((payload: string) => payload, { + dispose: () => { disposals[0] += 1; }, + }); + const second = Object.assign((payload: string) => payload, { + dispose: () => { disposals[1] += 1; }, + }); + const composed = composeSsePayloadRewrites(first, second); + const budget = createTestTranslatorBudget(); + + await readAll(relaySseWithPayloadRewrite(streamFromText("data: {}\n\n"), composed, budget)); + composed.dispose?.(); + + expect(disposals).toEqual([1, 1]); + budget.dispose(); + }); + + test("teardown attempts every child and cannot block upstream cancellation", async () => { + let secondDisposals = 0; + const throwing = Object.assign((payload: string) => payload, { + dispose: () => { throw new Error("dispose boom"); }, + }); + const second = Object.assign((payload: string) => payload, { + dispose: () => { secondDisposals += 1; }, + }); + const composed = composeSsePayloadRewrites(throwing, second); + expect(() => composed.dispose?.()).toThrow("dispose boom"); + expect(secondDisposals).toBe(1); + + let upstreamCancels = 0; + const upstream = new ReadableStream({ + cancel() { upstreamCancels += 1; }, + }); + const rewrite = Object.assign((payload: string) => payload, { + dispose: () => { throw new Error("dispose boom"); }, + }); + const budget = createTestTranslatorBudget(); + const reader = relaySseWithPayloadRewrite(upstream, rewrite, budget).getReader(); + await reader.cancel("test cancel"); + expect(upstreamCancels).toBe(1); + budget.dispose(); + }); + test("unterminated rewrite accumulation closes through a typed failed tail", async () => { const budget = createTestTranslatorBudget({ maxTurnBytes: 64 }); const upstream = new AbortController(); From 4fcdcd128f201417634d80f443aa8c4ea0ecd9dd Mon Sep 17 00:00:00 2001 From: Jingwei <164893774+0xJingwei@users.noreply.github.com> Date: Mon, 3 Aug 2026 00:17:04 +0800 Subject: [PATCH 3/4] fix: close responses repair review gaps --- .../060_impl_roadmap.md | 13 +++++++------ src/server/responses-snapshot-repair.ts | 7 ++++--- src/server/sse-payload-rewrite.ts | 6 +++++- tests/responses-snapshot-repair.test.ts | 18 ++++++++++++++---- 4 files changed, 30 insertions(+), 14 deletions(-) diff --git a/devlog/_plan/260731_macos_rss_retention/060_impl_roadmap.md b/devlog/_plan/260731_macos_rss_retention/060_impl_roadmap.md index 9387d53ce..8cb65a37c 100644 --- a/devlog/_plan/260731_macos_rss_retention/060_impl_roadmap.md +++ b/devlog/_plan/260731_macos_rss_retention/060_impl_roadmap.md @@ -82,12 +82,13 @@ tail loses reset semantics for opted-in clients). 110 is the terminal gate. queue. Suppress when `sawTerminal()` is true. Do not feed the tail through the inspector. -## Design decisions locked for 100 (from 052 risk table) +## Design decisions locked for 100 (from 052 risk table; rewrite scope superseded by 100) -- Gate change only: `(win32 || darwin) && !needsClientRewrite`, and darwin - additionally requires the EXPLICIT `streamMode: "eager-relay"` decision - (`reason: "config-eager"`). `auto` on darwin stays tee while - `MIN_FIXED_BUN_VERSION` is null. No rewrite-path widening. +- Windows keeps its existing routing policy. Darwin requires the EXPLICIT + `streamMode: "eager-relay"` decision (`reason: "config-eager"`) for both + no-rewrite and client-rewrite traffic; rewrites run inline after the eager + single reader. `auto` on darwin stays tee while `MIN_FIXED_BUN_VERSION` is + null. See `100_darwin_eager_optin.md` for the superseding gate matrix. - Bun#32111 posture per 052: eager avoids the known reproducer shape but is unproven on 1.3.14 — opt-in only, never default. @@ -95,6 +96,6 @@ tail loses reset semantics for opted-in clients). 110 is the terminal gate. - Allocator residual (053): documented NOOP until a stable Bun ships the allocator train. No forced GC, no FFI purge, no restart-as-fix. -- Rewrite-traffic eager migration (052 steps 4–6): future unit. +- Default eager migration beyond the explicit Darwin opt-in: future unit. - `auto` default flip: blocked on a released Bun carrying #32120 (UNSAFE boundary per goalplan). diff --git a/src/server/responses-snapshot-repair.ts b/src/server/responses-snapshot-repair.ts index dd666118e..385ff3f5f 100644 --- a/src/server/responses-snapshot-repair.ts +++ b/src/server/responses-snapshot-repair.ts @@ -84,12 +84,13 @@ function repairOutputItem( : []; changed = !Array.isArray(rawContent) || content.some((part, index) => part !== rawContent[index]); - const role = item.role === "assistant" ? item.role : "assistant"; + // Responses output-message roles are the literal "assistant"; input roles are not valid here. + const role = "assistant"; changed = changed || role !== item.role; if (changed) repaired = { ...repaired, content, role }; } - if (inferredStatus && typeof repaired.status !== "string") { + if (inferredStatus && (typeof repaired.status !== "string" || repaired.status.trim().length === 0)) { repaired = { ...repaired, status: inferredStatus }; } return repaired; @@ -132,7 +133,7 @@ function repairResponseSnapshot( repaired.tools = defaults.tools; changed = true; } - if (typeof repaired.status !== "string") { + if (typeof repaired.status !== "string" || repaired.status.trim().length === 0) { repaired.status = defaultStatus; changed = true; } diff --git a/src/server/sse-payload-rewrite.ts b/src/server/sse-payload-rewrite.ts index 693e382d7..f93d5f156 100644 --- a/src/server/sse-payload-rewrite.ts +++ b/src/server/sse-payload-rewrite.ts @@ -8,7 +8,11 @@ import type { TranslatorBudget } from "../lib/translator-budget"; */ export type SsePayloadRewrite = ((payload: string) => string) & { - /** Release state retained across events when the stream ends or is cancelled. */ + /** + * Release state retained across events when the stream ends or is cancelled. Implementations + * are idempotent. A composed rewrite attempts every child and then throws the first child error, + * so callers on stream-teardown paths must guard this call. + */ dispose?: () => void; }; diff --git a/tests/responses-snapshot-repair.test.ts b/tests/responses-snapshot-repair.test.ts index d49fd0e91..8ceae850d 100644 --- a/tests/responses-snapshot-repair.test.ts +++ b/tests/responses-snapshot-repair.test.ts @@ -5,7 +5,7 @@ import { } from "../src/server/responses-snapshot-repair"; import { handleResponses } from "../src/server/responses"; import { createTranslatorBudget } from "../src/lib/translator-budget"; -import { isEagerRelaySseResponse } from "../src/server/relay"; +import { isEagerRelaySseResponse, MAX_COMPLETED_OUTPUT_ITEMS } from "../src/server/relay"; import type { OcxConfig } from "../src/types"; function parsedSseEvents(text: string): Array> { @@ -112,7 +112,7 @@ describe("Responses passthrough sparse-snapshot repair", () => { test("repairs output-item status and nested reasoning summary parts", () => { const rewrite = createResponsesSnapshotPayloadRewrite(); - for (const invalidStatus of [undefined, null, 42]) { + for (const invalidStatus of [undefined, null, 42, ""]) { const added = JSON.parse(rewrite(JSON.stringify({ type: "response.output_item.added", output_index: 0, @@ -166,7 +166,7 @@ describe("Responses passthrough sparse-snapshot repair", () => { }); test("replaces structurally invalid response statuses with the event status", () => { - for (const invalidStatus of [null, 42]) { + for (const invalidStatus of [null, 42, ""]) { const rewrite = createResponsesSnapshotPayloadRewrite(); const event = JSON.parse(rewrite(JSON.stringify({ type: "response.completed", @@ -194,6 +194,16 @@ describe("Responses passthrough sparse-snapshot repair", () => { expect(event.response.output[1]?.status).toBeUndefined(); }); + test("repairs non-assistant output message roles to the canonical Responses role", () => { + const rewrite = createResponsesSnapshotPayloadRewrite(); + const event = JSON.parse(rewrite(JSON.stringify({ + type: "response.output_item.added", + item: { id: "msg_role", type: "message", role: "user", content: [] }, + }))) as { item: Record }; + + expect(event.item.role).toBe("assistant"); + }); + test("reconstructs a sparse completed output from prior done items", () => { for (const sparseOutput of [undefined, null, [], "invalid"]) { const rewrite = createResponsesSnapshotPayloadRewrite(); @@ -309,7 +319,7 @@ describe("Responses passthrough sparse-snapshot repair", () => { test("does not reconstruct a partial terminal after the retained-item cap is exceeded", () => { const rewrite = createResponsesSnapshotPayloadRewrite(); - for (let outputIndex = 0; outputIndex <= 256; outputIndex += 1) { + for (let outputIndex = 0; outputIndex <= MAX_COMPLETED_OUTPUT_ITEMS; outputIndex += 1) { rewrite(JSON.stringify({ type: "response.output_item.done", output_index: outputIndex, From fdbb3b77a8ca2c24af2a0928033b161a1b9fb7d7 Mon Sep 17 00:00:00 2001 From: Jingwei <164893774+0xJingwei@users.noreply.github.com> Date: Mon, 3 Aug 2026 00:36:55 +0800 Subject: [PATCH 4/4] fix: reject gapped response reconstruction --- src/server/relay.ts | 10 +++++-- src/server/responses-snapshot-repair.ts | 19 +++++++++---- tests/responses-snapshot-repair.test.ts | 17 ++++++++++++ tests/responses-state.test.ts | 4 +-- tests/sse-inspector-bounds.test.ts | 37 +++++++++++++++++++++++++ 5 files changed, 76 insertions(+), 11 deletions(-) diff --git a/src/server/relay.ts b/src/server/relay.ts index 06d1fec18..dcee3ee55 100644 --- a/src/server/relay.ts +++ b/src/server/relay.ts @@ -711,11 +711,15 @@ export function createSseInspector(handlers: SseInspectorHandlers): SseInspector return; } if (!hasAuthoritativeOutput && completedItemsByOutputIndex!.size > 0) { + const orderedItems = [...completedItemsByOutputIndex!.entries()] + .sort(([left], [right]) => left - right); + if (!orderedItems.every(([index], position) => index === position)) { + clearCompletedItems(); + return; + } response = { ...response, - output: [...completedItemsByOutputIndex!.entries()] - .sort(([left], [right]) => left - right) - .map(([, retained]) => retained.item), + output: orderedItems.map(([, retained]) => retained.item), }; } try { diff --git a/src/server/responses-snapshot-repair.ts b/src/server/responses-snapshot-repair.ts index 385ff3f5f..ea50f7012 100644 --- a/src/server/responses-snapshot-repair.ts +++ b/src/server/responses-snapshot-repair.ts @@ -244,14 +244,21 @@ export function createResponsesSnapshotPayloadRewrite( if (responseStatus && isPlainObject(event.response)) { const hasAuthoritativeOutput = Array.isArray(event.response.output) && event.response.output.length > 0; - const reconstructedOutput = type === "response.completed" + let reconstructedOutput: Record[] | undefined; + if (type === "response.completed" && !hasAuthoritativeOutput && !reconstructionTainted - && completedItems.size > 0 - ? [...completedItems.entries()] - .sort(([left], [right]) => left - right) - .map(([, retained]) => retained.item) - : undefined; + && completedItems.size > 0) { + const orderedItems = [...completedItems.entries()] + .sort(([left], [right]) => left - right); + if (orderedItems.every(([index], position) => index === position)) { + reconstructedOutput = orderedItems.map(([, retained]) => retained.item); + } else { + // A gap means at least one completed item is missing. Never compact later indexes into a + // shorter array that appears complete to persistence or the client. + reconstructionTainted = true; + } + } const response = repairResponseSnapshot( event.response, responseStatus, diff --git a/tests/responses-snapshot-repair.test.ts b/tests/responses-snapshot-repair.test.ts index 8ceae850d..39cfef561 100644 --- a/tests/responses-snapshot-repair.test.ts +++ b/tests/responses-snapshot-repair.test.ts @@ -237,6 +237,23 @@ describe("Responses passthrough sparse-snapshot repair", () => { } }); + test("does not reconstruct terminal output when completed item indexes have a gap", () => { + const rewrite = createResponsesSnapshotPayloadRewrite(); + for (const outputIndex of [0, 2]) { + rewrite(JSON.stringify({ + type: "response.output_item.done", + output_index: outputIndex, + item: { id: `msg_${outputIndex}`, type: "message", content: [] }, + })); + } + const terminal = JSON.parse(rewrite(JSON.stringify({ + type: "response.completed", + response: { id: "resp_gapped", object: "response" }, + }))) as { response: { output?: unknown } }; + + expect(terminal.response.output).toEqual([]); + }); + test("uses request tool metadata when sparse snapshots omit or corrupt it", () => { const requestDefaults = { parallel_tool_calls: false, diff --git a/tests/responses-state.test.ts b/tests/responses-state.test.ts index 337b22910..d15521055 100644 --- a/tests/responses-state.test.ts +++ b/tests/responses-state.test.ts @@ -463,12 +463,12 @@ describe("Responses previous_response_id state", () => { feedInspector(inspector, [ { type: "response.output_item.done", - output_index: 2, + output_index: 0, item: { type: "message", role: "assistant", content: [{ type: "output_text", text: "stale sentinel" }] }, }, { type: "response.output_item.done", - output_index: 2, + output_index: 0, item: { type: "function_call", call_id: "call_final", name: "final_lookup", arguments: "{}" }, }, { type: "response.completed", response: { id: "resp_334_duplicate", status: "completed", output: [] } }, diff --git a/tests/sse-inspector-bounds.test.ts b/tests/sse-inspector-bounds.test.ts index 82c25ad5e..d2d245e5f 100644 --- a/tests/sse-inspector-bounds.test.ts +++ b/tests/sse-inspector-bounds.test.ts @@ -232,6 +232,43 @@ describe("createSseInspector frame bounds", () => { }); describe("createSseInspector completed-item bounds", () => { + test("gapped completed indexes never synthesize a partial response", () => { + const completed: Array<{ output?: unknown }> = []; + const inspector = createSseInspector({ onCompletedResponse: response => completed.push(response) }); + inspector.feed(frame(doneItemEvent(0, { type: "message", id: "zero" }))); + inspector.feed(frame(doneItemEvent(2, { type: "message", id: "two" }))); + inspector.feed(frame(completedEvent("gapped"))); + + expect(completed).toEqual([]); + }); + + test("out-of-order contiguous indexes reconstruct in index order", () => { + const completed: Array<{ output?: unknown }> = []; + const inspector = createSseInspector({ onCompletedResponse: response => completed.push(response) }); + inspector.feed(frame(doneItemEvent(1, { type: "message", id: "one" }))); + inspector.feed(frame(doneItemEvent(0, { type: "message", id: "zero" }))); + inspector.feed(frame(completedEvent("out-of-order"))); + + expect(completed).toEqual([expect.objectContaining({ + output: [ + { type: "message", id: "zero" }, + { type: "message", id: "one" }, + ], + })]); + }); + + test("duplicate completed index replacement remains contiguous and latest-wins", () => { + const completed: Array<{ output?: unknown }> = []; + const inspector = createSseInspector({ onCompletedResponse: response => completed.push(response) }); + inspector.feed(frame(doneItemEvent(0, { type: "message", id: "old" }))); + inspector.feed(frame(doneItemEvent(0, { type: "message", id: "new" }))); + inspector.feed(frame(completedEvent("duplicate"))); + + expect(completed).toEqual([expect.objectContaining({ + output: [{ type: "message", id: "new" }], + })]); + }); + test("300 indexes retain at most the lowest 256 and count every high-index eviction", () => { const inspector = createSseInspector({ onCompletedResponse: () => {} }); for (let index = 0; index < 300; index += 1) {