address review feedback for deepseek responses repair

Make response ids unique, mint missing item ids under rewriteNonCanonicalIds, share rewrite/trailer state through one factory, pass trailers through the Windows eager path, and add focused regression tests. Also link the PR to #938.

Author: mouzhi

src/server/relay-eager.ts

@@ -58,6 +58,8 @@ export type EagerRelayHooks = {
 };
 
 export type EagerRelayOptions = {
+  /** Optional trailer emitted after rewrite flush (e.g. data: [DONE]). */
+  trailer?: string | (() => string | undefined);
   /** Bounded client queue in bytes; producer pauses above it. Default 8 MiB. */
   maxQueueBytes?: number;
   /** Transient-budget owner for the inline-rewrite frame buffer. */
@@ -232,6 +234,12 @@ export function relaySseEagerBounded(
               queuedBytes += tail.byteLength;
               try { controllerRef?.enqueue(tail); } catch { /* client already gone */ }
             }
+            const trailer = typeof opts?.trailer === "function" ? opts.trailer() : opts?.trailer;
+            if (trailer && !cancelled) {
+              const trailerBytes = new TextEncoder().encode(trailer);
+              queuedBytes += trailerBytes.byteLength;
+              try { controllerRef?.enqueue(trailerBytes); } catch { /* client already gone */ }
+            }
           }
           if (!hooks.sawTerminal() && !cancelled && !upstream.signal.aborted) {
             syntheticKind = "incomplete";

src/server/responses-item-id-repair.ts

@@ -86,7 +86,9 @@ function mapRawId(
   }
 
   if (!rawId) {
-    if (!state.repairMissingTerminalIds) return null;
+    // When non-canonical rewrite is active, mint a request-local id even if the upstream
+    // omitted the terminal id, so Codex lifecycle correlation stays stable.
+    if (!state.repairMissingTerminalIds && !state.rewriteNonCanonicalIds) return null;
     const minted = mintCanonicalId(type, state.scope, outputIndex);
     state.budget?.chargeRetained(new TextEncoder().encode(JSON.stringify([outputIndex, minted])).byteLength, { kind: "item_ids" });
     state.outputIds[type].set(outputIndex, minted);
@@ -151,7 +153,8 @@ function mapResponseId(state: ResponsesItemIdRepairState, rawId: string | undefi
   if (rawId.startsWith("resp_")) return rawId;
   const existing = state.responseIdMap.get(rawId);
   if (existing) return existing;
-  const minted = `resp_ocx_${state.scope}`;
+  // Keep each distinct upstream response id unique within the request-local scope.
+  const minted = `resp_ocx_${state.scope}_${state.responseIdMap.size}`;
   state.responseIdMap.set(rawId, minted);
   return minted;
 }
@@ -220,7 +223,7 @@ function rewriteOutputItem(
   if (!mapped) return { item, changed: false };
   const currentId = typeof item.id === "string" ? item.id : undefined;
   if (currentId === mapped) return { item, changed: false };
-  if (currentId === undefined && !state.repairMissingTerminalIds) return { item, changed: false };
+  if (currentId === undefined && !state.repairMissingTerminalIds && !state.rewriteNonCanonicalIds) return { item, changed: false };
   return { item: { ...item, id: mapped }, changed: true };
 }
 
@@ -254,7 +257,7 @@ function rewriteItemIdField(
   }
   if (!mapped) return { event, changed: false };
   if (currentId === mapped) return { event, changed: false };
-  if (currentId === undefined && !state.repairMissingTerminalIds) return { event, changed: false };
+  if (currentId === undefined && !state.repairMissingTerminalIds && !state.rewriteNonCanonicalIds) return { event, changed: false };
   const next: Record<string, unknown> = { ...event, item_id: mapped };
   if (state.rewriteNonCanonicalIds && "logprobs" in next) delete next.logprobs;
   return { event: next, changed: true };
@@ -403,36 +406,45 @@ function repairEventPayload(
  * rewriteNonCanonicalIds rewrites DeepSeek-style UUID item/response ids, converts
  * reasoning_text streams into Codex-friendly encrypted_content reasoning items, and
  * ensures a terminal [DONE] trailer so CLI/TUI turns finish.
+ *
+ * The rewrite and trailer share one request-local state object via closure so
+ * composition cannot lose the trailer state.
  */
-export function createResponsesItemIdPayloadRewrite(
+export interface ResponsesItemIdRepairHandlers {
+  rewrite: SsePayloadRewrite;
+  trailer: () => string | undefined;
+}
+
+export function createResponsesItemIdRepairHandlers(
   config: ResponsesItemIdRepairConfig,
   budget?: TranslatorBudget,
-): SsePayloadRewrite {
+): ResponsesItemIdRepairHandlers {
   const state = createRepairState(config, budget);
-  const rewrite: SsePayloadRewrite = (payload) => repairEventPayload(payload, state);
-  (rewrite as SsePayloadRewrite & { __state?: ResponsesItemIdRepairState }).__state = state;
-  return rewrite;
+  return {
+    rewrite: (payload) => repairEventPayload(payload, state),
+    trailer: () => {
+      if (state.sawCompleted && !state.sawDoneTrailer) return "data: [DONE]\n\n";
+      return undefined;
+    },
+  };
 }
 
-export function createResponsesItemIdDoneTrailer(
-  rewrite: SsePayloadRewrite,
-): () => string | undefined {
-  return () => {
-    const state = (rewrite as SsePayloadRewrite & { __state?: ResponsesItemIdRepairState }).__state;
-    if (!state) return undefined;
-    if (state.sawCompleted && !state.sawDoneTrailer) return "data: [DONE]\n\n";
-    return undefined;
-  };
+/** Backward-compatible helper used by existing tests and callers. */
+export function createResponsesItemIdPayloadRewrite(
+  config: ResponsesItemIdRepairConfig,
+  budget?: TranslatorBudget,
+): SsePayloadRewrite {
+  return createResponsesItemIdRepairHandlers(config, budget).rewrite;
 }
 
 export function relaySseWithResponsesItemIdRepair(
   body: ReadableStream<Uint8Array>,
   config: ResponsesItemIdRepairConfig,
   budget: TranslatorBudget,
 ): ReadableStream<Uint8Array> {
-  const rewrite = createResponsesItemIdPayloadRewrite(config, budget);
-  return relaySseWithPayloadRewrite(body, rewrite, budget, {
-    trailer: createResponsesItemIdDoneTrailer(rewrite),
+  const handlers = createResponsesItemIdRepairHandlers(config, budget);
+  return relaySseWithPayloadRewrite(body, handlers.rewrite, budget, {
+    trailer: handlers.trailer,
   });
 }
 

src/server/responses/core.ts

@@ -147,7 +147,7 @@ import { relaySseEagerBounded } from "../relay-eager";
 import { isWin32EagerRewrite, selectEagerPath } from "../../lib/bun-stream-caps";
 import { cancelBodyOnAbort } from "../../lib/abort";
 import {
-  createResponsesItemIdPayloadRewrite,
+  createResponsesItemIdRepairHandlers,
   hasResponsesItemIdRepair,
 } from "../responses-item-id-repair";
 import {
@@ -1818,12 +1818,13 @@ async function handleResponsesInner(
     if (isEventStream && upstreamResponse.body) {
       const repairConfig = route.provider.responsesItemIdRepair;
       const needsClientRewrite = imageGenCallAliases.size > 0 || hasResponsesItemIdRepair(repairConfig);
+      const itemIdRepair = hasResponsesItemIdRepair(repairConfig)
+        ? createResponsesItemIdRepairHandlers(repairConfig!, translatorBudget)
+        : undefined;
       // Compose opt-in payload rewrites into one parse/stringify pass (image-gen restore first).
       const payloadRewrites = [
         createImageGenCallRestoreRewrite(imageGenCallAliases),
-        hasResponsesItemIdRepair(repairConfig)
-          ? createResponsesItemIdPayloadRewrite(repairConfig!, translatorBudget)
-          : undefined,
+        itemIdRepair?.rewrite,
       ].filter((rewrite): rewrite is NonNullable<typeof rewrite> => rewrite !== 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
@@ -1887,7 +1888,12 @@ async function handleResponsesInner(
           },
           onClientCancel: () => options.onNativePassthroughCancel?.(),
           onDone: () => unregisterTurn(turnAc),
-        }, win32EagerRewrite ? { rewriteBudget: translatorBudget } : undefined);
+        }, win32EagerRewrite
+          ? {
+            rewriteBudget: translatorBudget,
+            ...(itemIdRepair ? { trailer: itemIdRepair.trailer } : {}),
+          }
+          : 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).
@@ -1961,7 +1967,12 @@ async function handleResponsesInner(
       // 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.
       const rewrittenBody = payloadRewrites.length > 0
-        ? relaySseWithPayloadRewrite(nativeBody, composeSsePayloadRewrites(...payloadRewrites), translatorBudget)
+        ? relaySseWithPayloadRewrite(
+          nativeBody,
+          composeSsePayloadRewrites(...payloadRewrites),
+          translatorBudget,
+          itemIdRepair ? { trailer: itemIdRepair.trailer } : undefined,
+        )
         : nativeBody;
       const clientBody = process.platform === "win32" && !needsClientRewrite
         ? nativeBody

src/server/sse-payload-rewrite.ts

@@ -141,7 +141,7 @@ export function relaySseWithPayloadRewrite(
     while ((next = nextSseBlock(buffer))) {
       replaceBuffer(next.rest);
       const payload = sseDataPayload(next.block);
-      if (payload) {
+      if (payload !== null) {
         const rewrittenPayload = rewrite(payload);
         if (rewrittenPayload === null) continue;
         const block = rewrittenPayload !== payload
@@ -156,7 +156,7 @@ export function relaySseWithPayloadRewrite(
     }
     if (flushFinal && buffer.length > 0) {
       const payload = sseDataPayload(buffer);
-      if (payload) {
+      if (payload !== null) {
         const rewrittenPayload = rewrite(payload);
         if (rewrittenPayload !== null) {
           const block = rewrittenPayload !== payload

tests/responses-item-id-repair-deepseek.test.ts

@@ -83,12 +83,53 @@ describe("DeepSeek Responses item-id repair", () => {
     expect(reasoningAdded.id).toMatch(/^rs_ocx_[0-9a-f]+_0$/);
     expect(messageAdded.id).toMatch(/^msg_ocx_[0-9a-f]+_1$/);
     expect(textDelta.item_id).toBe(messageAdded.id);
-    expect(completed.id).toMatch(/^resp_ocx_[0-9a-f]+$/);
+    expect(completed.id).toMatch(/^resp_ocx_[0-9a-f]+_\d+$/);
     expect(completed.output[0].id).toBe(reasoningAdded.id);
     expect(String(completed.output[0].encrypted_content)).toMatch(/^ocxr1:/);
     expect(completed.output[1].id).toBe(messageAdded.id);
   });
 
+
+  test("mints unique response ids for distinct upstream response ids", async () => {
+    const upstream = [
+      `event: response.completed\ndata: {"type":"response.completed","response":{"id":"11111111-1111-1111-1111-111111111111","status":"completed","output":[]}}\n\n`,
+      `event: response.completed\ndata: {"type":"response.completed","response":{"id":"22222222-2222-2222-2222-222222222222","status":"completed","output":[]}}\n\n`,
+    ].join("");
+    const events = await parseSse(await readAll(relaySseWithResponsesItemIdRepair(streamFromText(upstream), {
+      rewriteNonCanonicalIds: true,
+    })));
+    const first = (events[0].response as { id: string }).id;
+    const second = (events[1].response as { id: string }).id;
+    expect(first).toMatch(/^resp_ocx_[0-9a-f]+_0$/);
+    expect(second).toMatch(/^resp_ocx_[0-9a-f]+_1$/);
+    expect(first).not.toBe(second);
+  });
+
+  test("mints reasoning ids when rewriteNonCanonicalIds is enabled without repairMissingTerminalIds", async () => {
+    const upstream = [
+      `event: response.output_item.added\ndata: {"type":"response.output_item.added","output_index":0,"item":{"type":"reasoning","summary":[]}}\n\n`,
+      `event: response.output_item.done\ndata: {"type":"response.output_item.done","output_index":0,"item":{"type":"reasoning","summary":[]}}\n\n`,
+    ].join("");
+    const events = await parseSse(await readAll(relaySseWithResponsesItemIdRepair(streamFromText(upstream), {
+      rewriteNonCanonicalIds: true,
+    })));
+    const added = events[0].item as Record<string, unknown>;
+    const done = events[1].item as Record<string, unknown>;
+    expect(added.id).toMatch(/^rs_ocx_[0-9a-f]+_0$/);
+    expect(done.id).toBe(added.id);
+  });
+
+  test("does not duplicate an upstream [DONE] trailer", async () => {
+    const upstream = [
+      `event: response.completed\ndata: {"type":"response.completed","response":{"id":"58786d76-18be-43d9-b6f5-8922796fbe28","status":"completed","output":[]}}\n\n`,
+      `data: [DONE]\n\n`,
+    ].join("");
+    const repaired = await readAll(relaySseWithResponsesItemIdRepair(streamFromText(upstream), {
+      rewriteNonCanonicalIds: true,
+    }));
+    expect(repaired.match(/data: \[DONE\]/g)).toHaveLength(1);
+  });
+
   test("opt-in flag is recognized", () => {
     expect(hasResponsesItemIdRepair({ rewriteNonCanonicalIds: true })).toBe(true);
   });