Sanitize replayed turns before cross-provider request builds

src/provider/inference-dependencies.ts

@@ -7,6 +7,7 @@ import * as bifrostAdapter from "./bifrost-adapter.js";
 import * as openaiResponses from "./openai-responses-adapter.js";
 import { CODEX_RESPONSES_PROVIDER, withCodexContentTypeRepair } from "./codex-responses-adapter.js";
 import { GROK_RESPONSES_PROVIDER } from "./grok-responses-adapter.js";
+import { withReplaySanitizer } from "./replay-sanitizer.js";
 import { BIFROST_PROVIDER } from "./bifrost-adapter.js";
 import { OPENAI_RESPONSES_PROVIDER } from "./openai-responses-adapter.js";
 
@@ -59,6 +60,7 @@ export function createInferenceDependencies(): Promise<Dependencies> {
     cached = loadAdapterRegistry(manifest, {
       import: (specifier) => Promise.resolve(localModules[specifier]),
     })
+      .then(withReplaySanitizer)
       .then(createDependencies)
       .then((deps) => ({
         ...deps,

src/provider/replay-sanitizer.test.ts

@@ -0,0 +1,189 @@
+import { describe, expect, it } from "bun:test";
+import { createBuiltinRegistry } from "@intx/inference/providers";
+import type { ConversationTurn, LastCycleSource } from "@intx/types/runtime";
+import { sanitizeReplayTurns, withReplaySanitizer } from "./replay-sanitizer.js";
+
+const GROK_SIGNATURE = "grok-opaque-signature-blob";
+
+function grokThinkingHistory(): ConversationTurn[] {
+  return [
+    {
+      role: "user",
+      content: [{ type: "text", text: "hello" }],
+      timestamp: 1,
+    },
+    {
+      role: "assistant",
+      model: "grok-4",
+      content: [
+        { type: "thinking", thinking: "pondering", signature: GROK_SIGNATURE },
+        { type: "text", text: "answer", signature: GROK_SIGNATURE },
+      ],
+      timestamp: 2,
+    },
+    {
+      role: "user",
+      content: [{ type: "text", text: "continue" }],
+      timestamp: 3,
+    },
+  ];
+}
+
+function resolveSanitized(source: LastCycleSource) {
+  return withReplaySanitizer(createBuiltinRegistry()).resolve(source);
+}
+
+describe("sanitizeReplayTurns", () => {
+  it("strips foreign thinking blocks and signatures", () => {
+    const turns = sanitizeReplayTurns(grokThinkingHistory(), "claude-opus-4");
+    const assistant = turns.find((t) => t.role === "assistant");
+    expect(assistant).toBeDefined();
+    expect(assistant?.content.some((b) => b.type === "thinking")).toBe(false);
+    expect(JSON.stringify(turns)).not.toContain(GROK_SIGNATURE);
+  });
+
+  it("keeps thinking blocks for same-model replay", () => {
+    const turns = sanitizeReplayTurns(grokThinkingHistory(), "grok-4");
+    const assistant = turns.find((t) => t.role === "assistant");
+    expect(assistant?.content.some((b) => b.type === "thinking")).toBe(true);
+  });
+
+  it("converts foreign refusal blocks to text", () => {
+    const turns = sanitizeReplayTurns(
+      [
+        {
+          role: "assistant",
+          model: "gpt-5",
+          content: [{ type: "refusal", reason: "cannot comply" }],
+          timestamp: 1,
+        },
+      ],
+      "claude-opus-4",
+    );
+    expect(turns[0]?.content).toEqual([{ type: "text", text: "cannot comply" }]);
+  });
+
+  it("drops foreign redacted_thinking and citation blocks", () => {
+    const turns = sanitizeReplayTurns(
+      [
+        {
+          role: "assistant",
+          model: "claude-opus-4",
+          content: [
+            { type: "redacted_thinking", data: "opaque" },
+            { type: "text", text: "cited answer" },
+            { type: "citation", citedText: "quote", source: {} },
+          ],
+          timestamp: 1,
+        },
+      ],
+      "gemini-2.5-pro",
+    );
+    expect(turns[0]?.content).toEqual([{ type: "text", text: "cited answer" }]);
+  });
+
+  it("answers dangling tool_calls with a synthetic error result", () => {
+    const turns = sanitizeReplayTurns(
+      [
+        {
+          role: "assistant",
+          model: "grok-4",
+          content: [
+            { type: "text", text: "running tool" },
+            { type: "tool_call", id: "call_1", name: "ls", arguments: {} },
+          ],
+          timestamp: 1,
+        },
+      ],
+      "claude-opus-4",
+    );
+    const results = turns.flatMap((t) => t.content.filter((b) => b.type === "tool_result"));
+    expect(results).toHaveLength(1);
+    expect(results[0]).toMatchObject({ callId: "call_1", isError: true });
+  });
+});
+
+describe("withReplaySanitizer", () => {
+  it("builds an Anthropic request from a grok-signed thinking turn", () => {
+    const adapter = resolveSanitized({
+      sourceId: "s1",
+      provider: "anthropic",
+      model: "claude-opus-4",
+    });
+    const request = adapter.buildRequest(grokThinkingHistory(), "claude-opus-4", {});
+    expect(request.body).not.toContain(GROK_SIGNATURE);
+    expect(request.body).not.toContain('"thinking"');
+  });
+
+  it("builds a Google request from a grok-signed thinking turn", () => {
+    const adapter = resolveSanitized({
+      sourceId: "s1",
+      provider: "google-genai",
+      model: "gemini-2.5-pro",
+    });
+    const request = adapter.buildRequest(grokThinkingHistory(), "gemini-2.5-pro", {});
+    expect(request.body).not.toContain(GROK_SIGNATURE);
+    expect(request.body).not.toContain("thoughtSignature");
+  });
+
+  it("builds an Anthropic request from a persisted refusal block", () => {
+    const adapter = resolveSanitized({
+      sourceId: "s1",
+      provider: "anthropic",
+      model: "claude-opus-4",
+    });
+    const request = adapter.buildRequest(
+      [
+        {
+          role: "user",
+          content: [{ type: "text", text: "do it" }],
+          timestamp: 1,
+        },
+        {
+          role: "assistant",
+          model: "gpt-5",
+          content: [{ type: "refusal", reason: "cannot comply" }],
+          timestamp: 2,
+        },
+        {
+          role: "user",
+          content: [{ type: "text", text: "why not" }],
+          timestamp: 3,
+        },
+      ],
+      "claude-opus-4",
+      {},
+    );
+    expect(request.body).toContain("cannot comply");
+  });
+
+  it("builds an Anthropic request from a dangling tool_call", () => {
+    const adapter = resolveSanitized({
+      sourceId: "s1",
+      provider: "anthropic",
+      model: "claude-opus-4",
+    });
+    const request = adapter.buildRequest(
+      [
+        {
+          role: "user",
+          content: [{ type: "text", text: "list files" }],
+          timestamp: 1,
+        },
+        {
+          role: "assistant",
+          model: "grok-4",
+          content: [
+            { type: "text", text: "running tool" },
+            { type: "tool_call", id: "call_1", name: "ls", arguments: {} },
+          ],
+          timestamp: 2,
+        },
+      ],
+      "claude-opus-4",
+      {},
+    );
+    expect(request.body).toContain("tool_result");
+    expect(request.body).toContain("call_1");
+  });
+});

src/provider/replay-sanitizer.ts

@@ -0,0 +1,77 @@
+import { transformMessages, type AdapterRegistry } from "@intx/inference";
+import type { ContentBlock, ConversationTurn } from "@intx/types/runtime";
+
+// Repairs persisted history at the request-build boundary so a turn produced
+// by one provider replays safely against another. transformMessages (vendored)
+// strips thinking blocks for foreign-model turns, rewrites safety_rating to
+// text, and answers dangling tool_calls with synthetic error results — the
+// same tool_result/isError shape the reactor's gate-timeout path appends.
+// What it does not cover, this module handles first: output-only block types
+// with no cross-provider wire shape (refusal, citation, redacted_thinking,
+// audio, video, code execution) that make adapter builders throw, and opaque
+// provider signatures that a foreign provider rejects when echoed back.
+
+// Output-only shapes a foreign provider cannot round-trip; adapter builders
+// throw on them, so they are dropped from foreign-model turns before build.
+const FOREIGN_UNMAPPABLE_TYPES = new Set<ContentBlock["type"]>([
+  "redacted_thinking",
+  "citation",
+  "audio",
+  "video",
+  "code_execution_request",
+  "code_execution_result",
+]);
+
+function stripForeignBlocks(turn: ConversationTurn): ConversationTurn {
+  const content = turn.content.flatMap((block): ContentBlock[] => {
+    if (block.type === "refusal") {
+      return [{ type: "text", text: block.reason }];
+    }
+    if (FOREIGN_UNMAPPABLE_TYPES.has(block.type)) {
+      return [];
+    }
+    // Signatures authenticate a block to the provider that signed it; a
+    // foreign provider 400s when one is echoed back (Gemini replays them
+    // as thoughtSignature verbatim).
+    if ("signature" in block && block.signature !== undefined) {
+      const unsigned = { ...block };
+      delete unsigned.signature;
+      return [unsigned];
+    }
+    return [block];
+  });
+  return { ...turn, content };
+}
+
+/**
+ * Repair persisted turns for replay against `targetModel`. Assistant turns
+ * produced by a different model lose blocks the target provider cannot
+ * accept; dangling tool_calls are answered with synthetic error results.
+ */
+export function sanitizeReplayTurns(
+  turns: ConversationTurn[],
+  targetModel: string,
+): ConversationTurn[] {
+  const repaired = turns.map((turn) =>
+    turn.role === "assistant" && turn.model !== targetModel ? stripForeignBlocks(turn) : turn,
+  );
+  return transformMessages(repaired, { targetModel });
+}
+
+/**
+ * Wrap an adapter registry so every resolved adapter sanitizes replayed
+ * turns before building its request.
+ */
+export function withReplaySanitizer(adapters: AdapterRegistry): AdapterRegistry {
+  return {
+    has: (provider) => adapters.has(provider),
+    resolve(source, quirks) {
+      const adapter = adapters.resolve(source, quirks);
+      return {
+        ...adapter,
+        buildRequest: (turns, model, options) =>
+          adapter.buildRequest(sanitizeReplayTurns(turns, model), model, options),
+      };
+    },
+  };
+}