fix(sse): keep Responses tool-call ids/args intact and stop replaying output_item.done in PII transform (#15507)

changelog.d/fixes/15507-responses-pii-warn-bytes.md

@@ -0,0 +1 @@
+- fix(sse): keep Responses API tool-call ids/arguments intact and stop replaying output_item.done in the PII stream transform (#15507)

src/lib/sseTextTransform.ts

@@ -43,8 +43,26 @@ export const METADATA_KEYS = new Set([
   "event",
   "provider",
   "format",
+  // OpenAI Responses API protocol identifiers / opaque blobs (#15507): never user text.
+  "call_id",
+  "item_id",
+  "obfuscation",
+  "encrypted_content",
+  "response_id",
+  "previous_response_id",
+  "status",
 ]);
 
+// Responses API events whose `delta` carries tool-call arguments (JSON), not answer text.
+export function isResponsesToolArgsDelta(json: any): boolean {
+  const t = json?.type;
+  return (
+    typeof t === "string" &&
+    t.startsWith("response.") &&
+    (t.includes("_arguments.") || t.startsWith("response.custom_tool_call_input."))
+  );
+}
+
 /**
  * Classify a string field as a PII-processed category, or `null` when it is metadata that
  * must pass through untouched. `parentKey` is the key of the object that directly contains
@@ -218,7 +236,10 @@ export function createSseTextTransform(
             for (const key of Object.keys(obj)) {
               if (typeof obj[key] === "string") {
                 const val = obj[key];
-                const field = classifyField(key, parentKey);
+                const field =
+                  key === "delta" && obj === json && isResponsesToolArgsDelta(json)
+                    ? "toolArgs"
+                    : classifyField(key, parentKey);
                 if (field === null) {
                   continue;
                 }

src/lib/streamingPiiTransform.ts

@@ -220,9 +220,17 @@ export function createPiiSseTransform(options?: PiiTransformOptions): TransformS
 
     // 3. Responses API
     if (typeof lastJson.type === "string" && lastJson.type.startsWith("response.")) {
-      const finalJson = JSON.parse(JSON.stringify(lastJson));
-      const idx = typeof finalJson.output_index === "number" ? finalJson.output_index : 0;
+      const idx = typeof lastJson.output_index === "number" ? lastJson.output_index : 0;
       const buffers = getBuffers(`${idx}_0`);
+      // Never clone a terminal event (output_item.done etc.) — that replays it with the same
+      // sequence_number (#15507). Emit the buffered text as a fresh text delta instead.
+      if (!lastJson.type.endsWith(".delta")) {
+        if (!buffers.content) return null;
+        const delta = buffers.content;
+        buffers.content = "";
+        return { type: "response.output_text.delta", output_index: idx, delta };
+      }
+      const finalJson = JSON.parse(JSON.stringify(lastJson));
       if (buffers.content) {
         finalJson.delta = buffers.content;
         buffers.content = "";

tests/unit/issue-15507-pii-responses-tool-lifecycle.test.ts

@@ -0,0 +1,85 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+
+const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-test-issue-15507-"));
+process.env.DATA_DIR = tmpDir;
+process.env.PII_RESPONSE_SANITIZATION = "true";
+process.env.PII_RESPONSE_SANITIZATION_MODE = "warn";
+
+import { createPiiSseTransform } from "../../src/lib/streamingPiiTransform.ts";
+
+async function run(chunks: string[]): Promise<string> {
+  const t = createPiiSseTransform();
+  const writer = t.writable.getWriter();
+  const reader = t.readable.getReader();
+  const w = (async () => {
+    for (const c of chunks) await writer.write(new TextEncoder().encode(c));
+    await writer.close();
+  })();
+  const out: string[] = [];
+  let r = await reader.read();
+  while (!r.done) {
+    out.push(new TextDecoder().decode(r.value));
+    r = await reader.read();
+  }
+  await w;
+  return out.join("");
+}
+
+test("#15507: warn mode keeps Responses tool lifecycle bytes intact (no id mutation, no replayed done)", async () => {
+  const item = {
+    type: "function_call",
+    id: "fc_probe",
+    call_id: "call_probe",
+    name: "bash",
+    arguments: "",
+  };
+  const args = '{"command":"echo hi"}';
+  const events = [
+    { type: "response.output_item.added", sequence_number: 0, output_index: 0, item },
+    {
+      type: "response.function_call_arguments.delta",
+      sequence_number: 1,
+      output_index: 0,
+      item_id: "fc_probe",
+      delta: args,
+      obfuscation: "random-padding-1",
+    },
+    {
+      type: "response.function_call_arguments.done",
+      sequence_number: 2,
+      output_index: 0,
+      item_id: "fc_probe",
+      arguments: args,
+    },
+    {
+      type: "response.output_item.done",
+      sequence_number: 3,
+      output_index: 0,
+      item: { ...item, arguments: args },
+    },
+    {
+      type: "response.completed",
+      sequence_number: 4,
+      response: { id: "resp_probe", status: "completed", output: [{ ...item, arguments: args }] },
+    },
+  ];
+  const input = events.map((e) => `data: ${JSON.stringify(e)}\n\n`);
+  const output = await run(input);
+  const parsed = output
+    .split("\n")
+    .filter((l) => l.startsWith("data: "))
+    .map((l) => JSON.parse(l.slice(6)));
+
+  const added = parsed.find((e) => e.type === "response.output_item.added");
+  assert.equal(added.item.call_id, "call_probe", "added.item.call_id must be preserved");
+  const delta = parsed.find((e) => e.type === "response.function_call_arguments.delta");
+  assert.equal(delta.item_id, "fc_probe", "arg delta item_id must be preserved");
+  assert.equal(delta.delta, args, "arg delta must be preserved");
+  const dones = parsed.filter((e) => e.type === "response.output_item.done");
+  assert.equal(dones.length, 1, "exactly one output_item.done");
+  assert.equal(output, input.join(""), "warn mode must be byte-preserving");
+});