Codebase quality sweep: error handling, drift detection, test hygiene (#176)

## Summary

Codebase-wide quality sweep addressing 6 categories of issues across 53
files:

- **Error handling hardening** — wrap unguarded stream writes in
try/catch (recorder.ts, agui-recorder.ts), add response termination for
headers-already-sent paths (server.ts, a2a-mock.ts, mcp-mock.ts), fix
vector-mock double body consumption
- **Parse error detail surfacing** — capture and include the actual
JSON.parse error message in all 25+ bare catch blocks across every
provider handler, WebSocket handler, and stream-collapse function
- **Drift detection gaps** — `compareSSESequences` now compares ALL
events per type (not just the first), catching shape variations like
Anthropic thinking deltas; Ollama drift tests use synchronous env-var
gate instead of broken async `describe.skipIf`
- **Test hygiene** — fix 12 unrestored spy/mock leaks across 6 test
files, correct misleading "toolCalls should win" comments/assertions,
move prototype-level spy to afterEach, update 17 test assertion strings
for new error format

## Test plan

- [x] 2867 tests pass (79 files, 37 skipped)
- [x] TypeScript typecheck clean
- [x] Prettier + ESLint clean (pre-commit hooks pass)
- [x] 7-agent CR Round 1: 0 new bucket (a) findings from changes; 2
minor items fixed (comment + confirmed non-issue)
- [x] 10-slot blitz execution with zero merge conflicts

Author: jpr5

CHANGELOG.md

@@ -1,5 +1,23 @@
 # @copilotkit/aimock
 
+## [Unreleased]
+
+### Fixed
+
+- **Progressive relay for NDJSON and Bedrock binary event streams** — Ollama NDJSON and Bedrock binary event streams were fully buffered before relay, triggering downstream idle timeouts; now relayed progressively as chunks arrive
+- **JSON.parse error detail in bare catch blocks** — capture and surface parse-error detail in all bare catch blocks across 25+ provider/WebSocket/stream-collapse handlers instead of swallowing context
+- **Unguarded stream write/end calls** — wrap stream write/end in try/catch (recorder.ts, agui-recorder.ts) to prevent unhandled exceptions on client disconnect
+- **Response termination for headers-already-sent paths** — add response termination in error paths where headers were already sent (server.ts, a2a-mock.ts, mcp-mock.ts), preventing connection hangs
+- **Vector-mock double body consumption** — fix route passthrough consuming the request body twice, causing empty-body forwarding
+- **Drift detection compared only first event per type** — `compareSSESequences` now compares ALL events per type, not just the first, catching previously invisible divergences
+- **Ollama drift tests used broken async describe.skipIf** — replaced with synchronous env-var gate so tests are correctly skipped or executed
+- **12 unrestored spy/mock leaks and misleading assertions** — fix spy/mock leaks across test files and correct assertions that passed for the wrong reasons
+
+### Changed
+
+- **Anti-buffering headers on all progressive stream relay paths** — standard headers (Cache-Control, Connection, X-Accel-Buffering) added to all progressive stream relay paths to prevent intermediate proxy buffering
+- **Stream-collapse returns firstDroppedSample** — stream-collapse functions now return the first dropped sample for forensic debugging of collapsed streams
+
 ## [1.21.0] - 2026-05-11
 
 ### Added

src/__tests__/api-conformance.test.ts

@@ -1037,7 +1037,7 @@ describe("OpenAI Embeddings API conformance", () => {
       );
       expect(res.status).toBe(400);
       const json = JSON.parse(res.body);
-      expect(json.error.message).toBe("Malformed JSON");
+      expect(json.error.message).toMatch(/^Malformed JSON body: /);
     });
 
     it("Content-Type is application/json", async () => {

src/__tests__/bedrock-stream.test.ts

@@ -1442,7 +1442,7 @@ describe("POST /model/{modelId}/converse (malformed JSON)", () => {
 
     expect(res.status).toBe(400);
     const body = JSON.parse(res.body);
-    expect(body.error.message).toBe("Malformed JSON");
+    expect(body.error.message).toMatch(/^Malformed JSON: /);
   });
 });
 
@@ -1569,7 +1569,7 @@ describe("POST /model/{modelId}/converse-stream (malformed JSON)", () => {
 
     expect(res.status).toBe(400);
     const body = JSON.parse(res.body);
-    expect(body.error.message).toBe("Malformed JSON");
+    expect(body.error.message).toMatch(/^Malformed JSON: /);
   });
 });
 

src/__tests__/bedrock.test.ts

@@ -243,7 +243,7 @@ describe("POST /model/{modelId}/invoke (error handling)", () => {
 
     expect(res.status).toBe(400);
     const body = JSON.parse(res.body);
-    expect(body.error.message).toBe("Malformed JSON");
+    expect(body.error.message).toMatch(/^Malformed JSON: /);
   });
 });
 

src/__tests__/chaos.test.ts

@@ -590,6 +590,10 @@ describe("fixture-level chaos on non-OpenAI provider", () => {
 // ---------------------------------------------------------------------------
 
 describe("chaos with logLevel silent: invalid header is ignored gracefully", () => {
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
   it("proceeds normally and does not throw when x-aimock-chaos-drop is not a number", async () => {
     const fixtures: Fixture[] = [
       { match: { userMessage: "hello" }, response: { content: "Hi there" } },

src/__tests__/cohere.test.ts

@@ -575,7 +575,7 @@ describe("POST /v2/chat (validation)", () => {
 
     expect(res.status).toBe(400);
     const body = JSON.parse(res.body);
-    expect(body.error.message).toBe("Malformed JSON");
+    expect(body.error.message).toMatch(/^Malformed JSON body: /);
   });
 
   it("returns 404 when no fixture matches", async () => {

src/__tests__/drift/ollama.drift.ts

@@ -1,10 +1,10 @@
 /**
  * Ollama drift tests.
  *
- * Compares aimock's Ollama endpoint output shapes against a real local
- * Ollama instance. Skips automatically if Ollama is not reachable.
+ * Compares aimock's Ollama endpoint output shapes against a real Ollama
+ * instance. Skips unless OLLAMA_HOST is set in the environment.
  *
- * Requires: local Ollama running at http://localhost:11434
+ * Requires: OLLAMA_HOST env var (e.g. http://localhost:11434)
  */
 
 import { describe, it, expect, beforeAll, afterAll } from "vitest";
@@ -13,21 +13,10 @@ import { extractShape, triangulate, formatDriftReport } from "./schema.js";
 import { httpPost, startDriftServer, stopDriftServer } from "./helpers.js";
 
 // ---------------------------------------------------------------------------
-// Connectivity check
+// Environment-based opt-in (consistent with other drift files)
 // ---------------------------------------------------------------------------
 
-let OLLAMA_REACHABLE = false;
-
-async function checkOllamaConnectivity(): Promise<boolean> {
-  try {
-    const res = await fetch("http://localhost:11434/api/tags", {
-      signal: AbortSignal.timeout(3000),
-    });
-    return res.ok;
-  } catch {
-    return false;
-  }
-}
+const OLLAMA_HOST = process.env.OLLAMA_HOST ?? "http://localhost:11434";
 
 // ---------------------------------------------------------------------------
 // Server lifecycle
@@ -36,7 +25,6 @@ async function checkOllamaConnectivity(): Promise<boolean> {
 let instance: ServerInstance;
 
 beforeAll(async () => {
-  OLLAMA_REACHABLE = await checkOllamaConnectivity();
   instance = await startDriftServer();
 });
 
@@ -119,7 +107,7 @@ function parseNDJSON(body: string): object[] {
     .map((line) => JSON.parse(line) as object);
 }
 
-describe.skipIf(!OLLAMA_REACHABLE)("Ollama drift", () => {
+describe.skipIf(!process.env.OLLAMA_HOST)("Ollama drift", () => {
   it("/api/chat response shape matches", async () => {
     const sdkShape = ollamaChatResponseShape();
 
@@ -130,7 +118,7 @@ describe.skipIf(!OLLAMA_REACHABLE)("Ollama drift", () => {
     };
 
     const [realRes, mockRes] = await Promise.all([
-      httpPost("http://localhost:11434/api/chat", body),
+      httpPost(`${OLLAMA_HOST}/api/chat`, body),
       httpPost(`${instance.url}/api/chat`, body),
     ]);
 
@@ -161,7 +149,7 @@ describe.skipIf(!OLLAMA_REACHABLE)("Ollama drift", () => {
     };
 
     const [realRes, mockRes] = await Promise.all([
-      httpPost("http://localhost:11434/api/chat", body),
+      httpPost(`${OLLAMA_HOST}/api/chat`, body),
       httpPost(`${instance.url}/api/chat`, body),
     ]);
 
@@ -199,7 +187,7 @@ describe.skipIf(!OLLAMA_REACHABLE)("Ollama drift", () => {
     };
 
     const [realRes, mockRes] = await Promise.all([
-      httpPost("http://localhost:11434/api/generate", body),
+      httpPost(`${OLLAMA_HOST}/api/generate`, body),
       httpPost(`${instance.url}/api/generate`, body),
     ]);
 

src/__tests__/drift/schema.ts

@@ -424,25 +424,23 @@ export function compareSSESequences(
     }
   }
 
-  // Compare shapes of matching event types
+  // Compare shapes of matching event types — collect ALL events per type and
+  // merge their shapes so that variant payloads (e.g. text_delta vs
+  // thinking_delta on the same Anthropic event type) are all represented.
   for (const type of realTypeSet) {
     if (!mockTypeSet.has(type)) continue;
-    const realEvent = real.find((e) => e.type === type);
-    const mockEvent = mock.find((e) => e.type === type);
-    const sdkEvent = sdk.find((e) => e.type === type);
-
-    if (realEvent && mockEvent) {
-      const eventDiffs = triangulate(
-        sdkEvent?.dataShape ?? null,
-        realEvent.dataShape,
-        mockEvent.dataShape,
-      );
-      for (const d of eventDiffs) {
-        diffs.push({
-          ...d,
-          path: `SSE:${type}.${d.path}`,
-        });
-      }
+
+    const realEvents = real.filter((e) => e.type === type);
+    const mockEvents = mock.filter((e) => e.type === type);
+    const sdkEvents = sdk.filter((e) => e.type === type);
+
+    const realMerged = mergeShapes(realEvents.map((e) => e.dataShape));
+    const mockMerged = mergeShapes(mockEvents.map((e) => e.dataShape));
+    const sdkMerged = sdkEvents.length > 0 ? mergeShapes(sdkEvents.map((e) => e.dataShape)) : null;
+
+    const eventDiffs = triangulateAt(`SSE:${type}`, sdkMerged, realMerged, mockMerged);
+    for (const d of eventDiffs) {
+      diffs.push(d);
     }
   }
 

src/__tests__/embeddings.test.ts

@@ -425,7 +425,7 @@ describe("POST /v1/embeddings (error handling)", () => {
 
     expect(res.status).toBe(400);
     const body = JSON.parse(res.body);
-    expect(body.error.message).toBe("Malformed JSON");
+    expect(body.error.message).toMatch(/^Malformed JSON body: /);
     expect(body.error.code).toBe("invalid_json");
   });
 });

src/__tests__/fixture-loader.test.ts

@@ -49,6 +49,12 @@ describe("loadFixtureFile", () => {
   });
 
   afterEach(() => {
+    // Restore console spies that individual tests create via vi.spyOn(console, "warn").
+    // We cannot use vi.restoreAllMocks() here because the top-level vi.mock("node:fs")
+    // overrides would also be wiped, breaking subsequent tests.
+    if ("mockRestore" in console.warn) {
+      (console.warn as ReturnType<typeof vi.spyOn>).mockRestore();
+    }
     rmSync(tmpDir, { recursive: true, force: true });
   });
 
@@ -360,6 +366,12 @@ describe("loadFixturesFromDir", () => {
   });
 
   afterEach(() => {
+    // Restore console spies that individual tests create via vi.spyOn(console, "warn").
+    // We cannot use vi.restoreAllMocks() here because the top-level vi.mock("node:fs")
+    // overrides would also be wiped, breaking subsequent tests.
+    if ("mockRestore" in console.warn) {
+      (console.warn as ReturnType<typeof vi.spyOn>).mockRestore();
+    }
     rmSync(tmpDir, { recursive: true, force: true });
   });