fix: sweep — 15 fixes across router, realtime, recorder, fal-audio, gemini-interactions, and drift infra

CHANGELOG.md

@@ -2,6 +2,18 @@
 
 ## [Unreleased]
 
+### Fixed
+
+- **Router** — systemMessage array exact-match logic was unsatisfiable for 2+ needles; collapsed to substring matching. Added `elevenlabs-tts` and `translation` to endpoint compatibility filter.
+- **Recorder** — Content-Type empty-string fallback (`??` → `||`), derived `EndpointType` from `FixtureMatch` instead of duplicate union, negative guards on Gemini Interactions outputs detection, scoped `turnIndex`/`hasToolResult` to chat endpoints only.
+- **WS-Realtime** — session.update rollback now captures full snapshot instead of just model/type. Added Beta flat fields for noise reduction, transcription, and turn_detection. Joined all text content parts in `realtimeItemsToMessages`. Added try-catch with debug logging around `sendEvent` for WebSocket close race safety.
+- **WS-Gemini-Live** — replaced deterministic `call_gemini_${name}_${i}` tool call IDs with random `generateToolCallId()` to prevent cross-turn collisions. Pre-computed `resolvedToolCalls` for wire/history ID consistency. Added unrecognized-role warning and ws.send try-catch with debug logging.
+- **Gemini Interactions** — `interactionsUsage` honors Gemini-native field names (`promptTokenCount`/`candidatesTokenCount`/`totalTokenCount`). `truncateAfterChunks` only counts `content.delta` events. Added `webSearches` warning on tool-call branch.
+- **fal-audio + ElevenLabs** — all journal entries now use `flattenHeaders(req.headers)` instead of `{}`. `handleSyncRun` accepts `RawJSONResponse` fixtures from queue-walk recordings.
+- **Helpers** — extended `resolveUsage` with Gemini-native token fields. Preserved error cause in `resolveResponse` factory rethrow. `buildEmbeddingResponse` accepts optional usage. `extractFormField` escapes regex metacharacters.
+- **Drift test infra** — retry logging with body consumption, broadened `redactUrl` to cover `api_key`/`apikey`/`token`/`access_token` patterns, URL threaded into error messages with redaction, `parseDataOnlySSE` [DONE] filter fix, `parseTypedSSE` multi-line data handling with null guards.
+- **Drift collector** — invoke vitest directly via npx to avoid pnpm stdout prefix breaking JSON parse; classify raw stack traces as infrastructure errors instead of crashing.
+
 ## [1.27.0] - 2026-05-20
 
 ### Added
@@ -12,7 +24,6 @@
 
 - **Walk structured content arrays in `extractLastUserMessage`** — handle multimodal user content (`AGUIMessageContentPart[]`) by joining text parts and skipping non-text. Export `NO_USER_MESSAGE_SENTINEL` constant and `AGUIMessageContentPart` type. ([#231](https://github.com/CopilotKit/aimock/pull/231))
 - **Harden recorder against error responses, double-settle, and broken sentinel persistence** — guard against recording fixtures from non-2xx upstream responses, add `settled` flag to prevent error+end race, skip disk write for predicate fixtures (sentinel was semantically broken on reload), include parse error reason in SSE warning log
-
 ## [1.26.1] - 2026-05-19
 
 ### Added

scripts/drift-report-collector.ts

@@ -300,7 +300,7 @@ function parseVitestOutput(stdout: string, context: string): VitestJsonResult |
 
 function runDriftTests(): VitestJsonResult {
   try {
-    const stdout = execSync("pnpm test:drift --reporter=json", {
+    const stdout = execSync("npx vitest run --config vitest.config.drift.ts --reporter=json", {
       encoding: "utf-8",
       stdio: ["pipe", "pipe", "pipe"],
       maxBuffer: 50 * 1024 * 1024,
@@ -409,6 +409,9 @@ function collectDriftEntries(results: VitestJsonResult): DriftEntry[] {
       /returned empty body/i,
       /waitUntil timeout/i,
       /AssertionError/i,
+      /^Error:/m,
+      /at\s+\S+\s+\(file:/,
+      /STACK_TRACE_ERROR/,
     ];
     const driftLikeIndicators = [/drift/i, /mismatch/i, /expected.*but/i, /LLMOCK DRIFT/i];
 

src/__tests__/drift/providers.ts

@@ -39,6 +39,10 @@ async function fetchWithRetry(url: string, init: RequestInit, maxRetries = 3): P
     try {
       const res = await fetch(url, init);
       if (RETRYABLE_STATUSES.has(res.status) && attempt < maxRetries - 1) {
+        console.warn(
+          `Retry ${attempt + 1}/${maxRetries} after ${res.status} for ${url.slice(0, 80)}`,
+        );
+        await res.text(); // consume body to free socket
         const backoff = Math.pow(2, attempt) * 1000;
         await new Promise((r) => setTimeout(r, backoff));
         continue;
@@ -59,15 +63,21 @@ async function fetchWithRetry(url: string, init: RequestInit, maxRetries = 3): P
 // Response parsing
 // ---------------------------------------------------------------------------
 
-function assertOk(raw: string, status: number, context: string): void {
+/** Redact API keys from query parameters in URLs for safe error messages */
+function redactUrl(url: string): string {
+  return url.replace(/([?&])(api[-_]?key|key|token|access_token)=[^&]+/gi, "$1$2=REDACTED");
+}
+
+function assertOk(raw: string, status: number, context: string, url?: string): void {
   if (status >= 400) {
-    throw new Error(`${context}: API returned ${status}: ${raw.slice(0, 300)}`);
+    const urlSuffix = url ? ` (${redactUrl(url)})` : "";
+    throw new Error(`${context}: API returned ${status}${urlSuffix}: ${raw.slice(0, 300)}`);
   }
 }
 
-function parseJsonResponse(raw: string, status: number, context: string): unknown {
+function parseJsonResponse(raw: string, status: number, context: string, url?: string): unknown {
   if (!raw) throw new Error(`${context}: empty response (status ${status})`);
-  assertOk(raw, status, context);
+  assertOk(raw, status, context, url);
   try {
     return JSON.parse(raw);
   } catch {
@@ -88,14 +98,18 @@ function normalizeLineEndings(text: string): string {
 function parseDataOnlySSE(text: string): { data: unknown }[] {
   return normalizeLineEndings(text)
     .split("\n\n")
-    .filter((block) => block.startsWith("data: ") && !block.includes("[DONE]"))
+    .filter((block) => block.startsWith("data: ") && block.trim() !== "data: [DONE]")
     .map((block) => {
       // Rejoin continuation lines (data split across lines)
       const json = block
         .split("\n")
         .map((line) => (line.startsWith("data: ") ? line.slice(6) : line))
         .join("");
-      return { data: JSON.parse(json) };
+      try {
+        return { data: JSON.parse(json) };
+      } catch (err) {
+        throw new Error(`Malformed SSE JSON in frame: ${json.slice(0, 100)}`, { cause: err });
+      }
     });
 }
 
@@ -105,12 +119,27 @@ function parseTypedSSE(text: string): { type: string; data: unknown }[] {
     .split("\n\n")
     .filter((block) => block.includes("event: ") && block.includes("data: "))
     .map((block) => {
-      const eventMatch = block.match(/^event: (.+)$/m);
-      const dataMatch = block.match(/^data: (.+)$/m);
-      return {
-        type: eventMatch![1],
-        data: JSON.parse(dataMatch![1]),
-      };
+      const eventMatch = block.match(/^event: (.*)$/m);
+      if (!eventMatch) {
+        throw new Error("Malformed SSE block: " + block.slice(0, 100));
+      }
+      // Handle multi-line data: collect all data lines and join them
+      const json = block
+        .split("\n")
+        .filter((line) => line.startsWith("data: "))
+        .map((line) => line.slice(6))
+        .join("");
+      if (!json) {
+        throw new Error("Malformed SSE block (no data): " + block.slice(0, 100));
+      }
+      try {
+        return {
+          type: eventMatch[1],
+          data: JSON.parse(json),
+        };
+      } catch (err) {
+        throw new Error(`Malformed SSE JSON in frame: ${json.slice(0, 100)}`, { cause: err });
+      }
     });
 }
 
@@ -339,7 +368,7 @@ export async function geminiNonStreaming(
   });
 
   const raw = await res.text();
-  return { status: res.status, body: parseJsonResponse(raw, res.status, "Gemini"), raw };
+  return { status: res.status, body: parseJsonResponse(raw, res.status, "Gemini", url), raw };
 }
 
 export async function geminiStreaming(
@@ -361,7 +390,7 @@ export async function geminiStreaming(
   });
 
   const raw = await res.text();
-  assertOk(raw, res.status, "Gemini streaming");
+  assertOk(raw, res.status, "Gemini streaming", url);
   const parsed = parseDataOnlySSE(raw);
   const rawEvents = parsed.map((p) => ({
     type: "gemini.chunk",
@@ -590,13 +619,11 @@ export async function listAnthropicModels(apiKey: string): Promise<string[]> {
 }
 
 export async function listGeminiModels(apiKey: string): Promise<string[]> {
-  const res = await fetchWithRetry(
-    `https://generativelanguage.googleapis.com/v1beta/models?key=${apiKey}`,
-    { method: "GET" },
-  );
+  const url = `https://generativelanguage.googleapis.com/v1beta/models?key=${apiKey}`;
+  const res = await fetchWithRetry(url, { method: "GET" });
 
   const raw = await res.text();
-  const json = parseJsonResponse(raw, res.status, "Gemini model list") as {
+  const json = parseJsonResponse(raw, res.status, "Gemini model list", url) as {
     models: { name: string }[];
   };
   // Gemini returns "models/gemini-2.5-flash" — strip prefix

src/__tests__/ws-gemini-live.test.ts

@@ -181,7 +181,7 @@ describe("WebSocket Gemini Live BidiGenerateContent", () => {
     expect(msg.toolCall.functionCalls).toHaveLength(1);
     expect(msg.toolCall.functionCalls[0].name).toBe("get_weather");
     expect(msg.toolCall.functionCalls[0].args).toEqual({ city: "NYC" });
-    expect(msg.toolCall.functionCalls[0].id).toBe("call_gemini_get_weather_0");
+    expect(msg.toolCall.functionCalls[0].id).toMatch(/^call_/);
 
     // Separate turnComplete message follows the toolCall
     const turnCompleteMsg = JSON.parse(raw[2]);
@@ -907,7 +907,7 @@ describe("WebSocket Gemini Live BidiGenerateContent", () => {
   it("handles user turn with functionResponse that has string response", async () => {
     // Fixture that matches a tool call id
     const toolResultFixtureStr: Fixture = {
-      match: { toolCallId: "call_gemini_search_0" },
+      match: { toolCallId: "call_search_1" },
       response: { content: "Result processed" },
     };
     instance = await createServer([toolResultFixtureStr]);
@@ -917,13 +917,22 @@ describe("WebSocket Gemini Live BidiGenerateContent", () => {
     await ws.waitForMessages(1); // setupComplete
 
     // Send clientContent with functionResponse where response is a string
+    // Provide explicit id so it matches the fixture's toolCallId
     ws.send(
       JSON.stringify({
         clientContent: {
           turns: [
             {
               role: "user",
-              parts: [{ functionResponse: { name: "search", response: "string-result" } }],
+              parts: [
+                {
+                  functionResponse: {
+                    name: "search",
+                    response: "string-result",
+                    id: "call_search_1",
+                  },
+                },
+              ],
             },
           ],
           turnComplete: true,
@@ -941,7 +950,7 @@ describe("WebSocket Gemini Live BidiGenerateContent", () => {
   it("handles toolResponse with fallback id and string response", async () => {
     // Fixture matching on tool call id
     const toolResultFixture3: Fixture = {
-      match: { toolCallId: "call_gemini_lookup_0" },
+      match: { toolCallId: "call_lookup_1" },
       response: { content: "Lookup done" },
     };
     instance = await createServer([toolResultFixture3]);
@@ -950,11 +959,13 @@ describe("WebSocket Gemini Live BidiGenerateContent", () => {
     ws.send(setupMsg());
     await ws.waitForMessages(1); // setupComplete
 
-    // Send toolResponse without id (relies on fallback) and with string response
+    // Send toolResponse with explicit id and string response
     ws.send(
       JSON.stringify({
         toolResponse: {
-          functionResponses: [{ name: "lookup", response: "string-response-value" }],
+          functionResponses: [
+            { name: "lookup", response: "string-response-value", id: "call_lookup_1" },
+          ],
         },
       }),
     );
@@ -966,6 +977,87 @@ describe("WebSocket Gemini Live BidiGenerateContent", () => {
     ws.close();
   });
 
+  it("generates random call_ ID when toolResponse functionResponse has no id", async () => {
+    instance = await createServer(allFixtures);
+    const ws = await connectWebSocket(instance.url, GEMINI_WS_PATH);
+
+    ws.send(setupMsg());
+    await ws.waitForMessages(1); // setupComplete
+
+    // Send toolResponse WITHOUT an id field — exercises the generateToolCallId() fallback
+    ws.send(toolResponseMsg("get_weather", { temp: "72F" }));
+
+    // No fixture will match the random ID, so we get a "No fixture matched" error
+    const raw = await ws.waitForMessages(2);
+    const msg = JSON.parse(raw[1]);
+    expect(msg.error).toBeDefined();
+    expect(msg.error.message).toBe("No fixture matched");
+
+    // Small pause to ensure journal write completed
+    await new Promise((r) => setTimeout(r, 50));
+
+    // Inspect the journal to verify the generated tool_call_id starts with call_
+    const entry = instance.journal.getLast();
+    expect(entry).not.toBeNull();
+    const messages = entry!.body!.messages;
+    const toolMsg = messages.find((m) => m.role === "tool");
+    expect(toolMsg).toBeDefined();
+    expect(toolMsg!.tool_call_id).toMatch(/^call_/);
+    // Verify it's a random ID (not the old deterministic format)
+    expect(toolMsg!.tool_call_id).not.toMatch(/^call_gemini_/);
+
+    ws.close();
+  });
+
+  it("generates random call_ ID when clientContent functionResponse has no id", async () => {
+    const afterToolFixture: Fixture = {
+      match: { userMessage: "continue-after-tool" },
+      response: { content: "Continued" },
+    };
+    instance = await createServer([afterToolFixture]);
+    const ws = await connectWebSocket(instance.url, GEMINI_WS_PATH);
+
+    ws.send(setupMsg());
+    await ws.waitForMessages(1); // setupComplete
+
+    // Send clientContent with a functionResponse lacking an id, followed by user text
+    ws.send(
+      JSON.stringify({
+        clientContent: {
+          turns: [
+            {
+              role: "user",
+              parts: [
+                { functionResponse: { name: "search", response: { results: [] } } },
+                { text: "continue-after-tool" },
+              ],
+            },
+          ],
+          turnComplete: true,
+        },
+      }),
+    );
+
+    const raw = await ws.waitForMessages(3); // setupComplete + content + turnComplete
+    const contentMsg = JSON.parse(raw[1]);
+    expect(contentMsg.serverContent).toBeDefined();
+
+    // Small pause to ensure journal write completed
+    await new Promise((r) => setTimeout(r, 50));
+
+    // Inspect the journal to verify the generated tool_call_id starts with call_
+    const entry = instance.journal.getLast();
+    expect(entry).not.toBeNull();
+    const messages = entry!.body!.messages;
+    const toolMsg = messages.find((m) => m.role === "tool");
+    expect(toolMsg).toBeDefined();
+    expect(toolMsg!.tool_call_id).toMatch(/^call_/);
+    // Verify it's a random ID (not the old deterministic format)
+    expect(toolMsg!.tool_call_id).not.toMatch(/^call_gemini_/);
+
+    ws.close();
+  });
+
   it("handles setup with tools that have empty functionDeclarations", async () => {
     instance = await createServer(allFixtures);
     const ws = await connectWebSocket(instance.url, GEMINI_WS_PATH);

src/elevenlabs-audio.ts

@@ -122,7 +122,7 @@ export async function handleElevenLabsTTS(
       journal.add({
         method,
         path,
-        headers: {},
+        headers: flattenHeaders(req.headers),
         body: syntheticReq,
         response: {
           status: 503,
@@ -375,7 +375,7 @@ export async function handleElevenLabsAudio(
       journal.add({
         method,
         path,
-        headers: {},
+        headers: flattenHeaders(req.headers),
         body: syntheticReq,
         response: {
           status: 503,