fix: harden Converse handlers with metadata events, text filtering, and stream flag

- Add optional logger to converseToCompletionRequest for warnings
- Filter empty-string text blocks in all conversion paths
- Warn on unexpected message roles
- Change textContent || null to ?? null to preserve empty strings
- Set completionReq.stream = true in handleConverseStream
- Add metadata events (usage + latencyMs) to all 3 stream builders
- Remove duplicate contentBlockIndex from stream event payloads
- Add webSearches warning on tool-call-only responses
- Add JSDoc to converseUsage explaining zero-default behavior
- Fix error type || to ?? for consistency
- Update test expectations and add safeResolve guard in postPartialBinary
- Import shared test helpers from helpers/mock-res.ts

Author: jpr5

src/__tests__/bedrock-stream.test.ts

@@ -1,15 +1,15 @@
 import { describe, it, expect, afterEach } from "vitest";
 import * as http from "node:http";
 import { crc32 } from "node:zlib";
-import type { Fixture, HandlerDefaults } from "../types.js";
+import type { Fixture } from "../types.js";
 import { createServer, type ServerInstance } from "../server.js";
 import {
   converseToCompletionRequest,
   handleConverse,
   handleConverseStream,
 } from "../bedrock-converse.js";
 import { Journal } from "../journal.js";
-import { Logger } from "../logger.js";
+import { createMockReq, createMockRes, createDefaults } from "./helpers/mock-res.js";
 
 // --- helpers ---
 
@@ -161,6 +161,13 @@ function postPartialBinary(
     const parsed = new URL(url);
     const chunks: Buffer[] = [];
     let aborted = false;
+    let resolved = false;
+    const safeResolve = (value: { body: Buffer; aborted: boolean }) => {
+      if (!resolved) {
+        resolved = true;
+        resolve(value);
+      }
+    };
     const req = http.request(
       {
         hostname: parsed.hostname,
@@ -175,7 +182,7 @@ function postPartialBinary(
       (res) => {
         res.on("data", (c: Buffer) => chunks.push(c));
         res.on("end", () => {
-          resolve({ body: Buffer.concat(chunks), aborted });
+          safeResolve({ body: Buffer.concat(chunks), aborted });
         });
         res.on("error", () => {
           aborted = true;
@@ -184,13 +191,13 @@ function postPartialBinary(
           aborted = true;
         });
         res.on("close", () => {
-          resolve({ body: Buffer.concat(chunks), aborted });
+          safeResolve({ body: Buffer.concat(chunks), aborted });
         });
       },
     );
     req.on("error", () => {
       aborted = true;
-      resolve({ body: Buffer.concat(chunks), aborted });
+      safeResolve({ body: Buffer.concat(chunks), aborted });
     });
     req.write(data);
     req.end();
@@ -1472,7 +1479,7 @@ describe("converseToCompletionRequest (edge cases)", () => {
       },
       "model",
     );
-    expect(result.messages[0]).toEqual({ role: "assistant", content: null });
+    expect(result.messages[0]).toEqual({ role: "assistant", content: "" });
   });
 
   it("handles user tool result with missing text in content items (text ?? '' fallback)", () => {
@@ -1583,8 +1590,8 @@ describe("converseToCompletionRequest (edge cases)", () => {
       "model",
     );
     expect(result.messages[0].tool_calls).toHaveLength(1);
-    // Empty text → content is null (falsy)
-    expect(result.messages[0].content).toBeNull();
+    // Empty text → content is "" (nullish coalescing preserves empty string)
+    expect(result.messages[0].content).toBe("");
   });
 });
 
@@ -1723,50 +1730,6 @@ describe("POST /model/{modelId}/invoke-with-response-stream (error fixture no ex
 
 // ─── Direct handler tests for req.method/req.url fallback branches ──────────
 
-function createMockReq(overrides: Partial<http.IncomingMessage> = {}): http.IncomingMessage {
-  return {
-    method: undefined,
-    url: undefined,
-    headers: {},
-    ...overrides,
-  } as unknown as http.IncomingMessage;
-}
-
-function createMockRes(): http.ServerResponse & { _written: string; _status: number } {
-  const res = {
-    _written: "",
-    _status: 0,
-    writableEnded: false,
-    statusCode: 0,
-    writeHead(status: number) {
-      res._status = status;
-      res.statusCode = status;
-    },
-    setHeader() {},
-    write(data: string) {
-      res._written += data;
-      return true;
-    },
-    end(data?: string) {
-      if (data) res._written += data;
-      res.writableEnded = true;
-    },
-    destroy() {
-      res.writableEnded = true;
-    },
-  };
-  return res as unknown as http.ServerResponse & { _written: string; _status: number };
-}
-
-function createDefaults(overrides: Partial<HandlerDefaults> = {}): HandlerDefaults {
-  return {
-    latency: 0,
-    chunkSize: 100,
-    logger: new Logger("silent"),
-    ...overrides,
-  };
-}
-
 describe("handleConverse (direct handler call, method/url fallbacks)", () => {
   it("uses fallback for text response with undefined method/url", async () => {
     const fixture: Fixture = {

src/bedrock-converse.ts

@@ -75,6 +75,14 @@ function converseStopReason(
   return overrideFinishReason;
 }
 
+/**
+ * Build Converse-format usage from fixture overrides.
+ *
+ * When no overrides are provided (the common case for mocks), all token
+ * counts default to zero.  This is intentional — aimock is a mock server
+ * and does not perform real tokenisation.  Callers that need non-zero
+ * usage should supply explicit `usage` overrides in their fixture.
+ */
 function converseUsage(overrides?: ResponseOverrides): {
   inputTokens: number;
   outputTokens: number;
@@ -113,15 +121,13 @@ function buildBedrockStreamTextEvents(
     events.push({
       eventType: "contentBlockStart",
       payload: {
-        contentBlockIndex: blockIndex,
         contentBlockStart: { contentBlockIndex: blockIndex, start: { type: "thinking" } },
       },
     });
     for (let i = 0; i < reasoning.length; i += chunkSize) {
       events.push({
         eventType: "contentBlockDelta",
         payload: {
-          contentBlockIndex: blockIndex,
           contentBlockDelta: {
             contentBlockIndex: blockIndex,
             delta: { type: "thinking_delta", thinking: reasoning.slice(i, i + chunkSize) },
@@ -136,15 +142,13 @@ function buildBedrockStreamTextEvents(
   events.push({
     eventType: "contentBlockStart",
     payload: {
-      contentBlockIndex: textBlockIndex,
       contentBlockStart: { contentBlockIndex: textBlockIndex, start: { type: "text" } },
     },
   });
   for (let i = 0; i < content.length; i += chunkSize) {
     events.push({
       eventType: "contentBlockDelta",
       payload: {
-        contentBlockIndex: textBlockIndex,
         contentBlockDelta: {
           contentBlockIndex: textBlockIndex,
           delta: { type: "text_delta", text: content.slice(i, i + chunkSize) },
@@ -157,6 +161,11 @@ function buildBedrockStreamTextEvents(
     eventType: "messageStop",
     payload: { stopReason: converseStopReason(overrides?.finishReason, "end_turn") },
   });
+  const usage = converseUsage(overrides);
+  events.push({
+    eventType: "metadata",
+    payload: { metadata: { usage, metrics: { latencyMs: 0 } } },
+  });
   return events;
 }
 
@@ -172,15 +181,16 @@ function buildBedrockStreamContentWithToolCallsEvents(
     ...overrides,
     finishReason: "stop",
   });
-  events.pop();
+  // Remove trailing metadata + messageStop events — we re-emit them after tool blocks
+  events.pop(); // metadata
+  events.pop(); // messageStop
   let blockIndex = reasoning ? 2 : 1;
 
   for (const tc of toolCalls) {
     const toolUseId = tc.id || generateToolUseId();
     events.push({
       eventType: "contentBlockStart",
       payload: {
-        contentBlockIndex: blockIndex,
         contentBlockStart: {
           contentBlockIndex: blockIndex,
           start: { toolUse: { toolUseId, name: tc.name } },
@@ -192,7 +202,6 @@ function buildBedrockStreamContentWithToolCallsEvents(
       events.push({
         eventType: "contentBlockDelta",
         payload: {
-          contentBlockIndex: blockIndex,
           contentBlockDelta: {
             contentBlockIndex: blockIndex,
             delta: { toolUse: { input: argsStr.slice(i, i + chunkSize) } },
@@ -207,6 +216,11 @@ function buildBedrockStreamContentWithToolCallsEvents(
     eventType: "messageStop",
     payload: { stopReason: converseStopReason(overrides?.finishReason, "tool_use") },
   });
+  const usage = converseUsage(overrides);
+  events.push({
+    eventType: "metadata",
+    payload: { metadata: { usage, metrics: { latencyMs: 0 } } },
+  });
   return events;
 }
 
@@ -226,7 +240,6 @@ function buildBedrockStreamToolCallEvents(
     events.push({
       eventType: "contentBlockStart",
       payload: {
-        contentBlockIndex: tcIdx,
         contentBlockStart: {
           contentBlockIndex: tcIdx,
           start: { toolUse: { toolUseId, name: tc.name } },
@@ -238,7 +251,6 @@ function buildBedrockStreamToolCallEvents(
       events.push({
         eventType: "contentBlockDelta",
         payload: {
-          contentBlockIndex: tcIdx,
           contentBlockDelta: {
             contentBlockIndex: tcIdx,
             delta: { toolUse: { input: argsStr.slice(i, i + chunkSize) } },
@@ -252,6 +264,11 @@ function buildBedrockStreamToolCallEvents(
     eventType: "messageStop",
     payload: { stopReason: converseStopReason(overrides?.finishReason, "tool_use") },
   });
+  const usage = converseUsage(overrides);
+  events.push({
+    eventType: "metadata",
+    payload: { metadata: { usage, metrics: { latencyMs: 0 } } },
+  });
   return events;
 }
 
@@ -260,6 +277,7 @@ function buildBedrockStreamToolCallEvents(
 export function converseToCompletionRequest(
   req: ConverseRequest,
   modelId: string,
+  logger?: Logger,
 ): ChatCompletionRequest {
   const messages: ChatMessage[] = [];
 
@@ -275,7 +293,9 @@ export function converseToCompletionRequest(
     if (msg.role === "user") {
       // Check for toolResult blocks
       const toolResults = msg.content.filter((b) => b.toolResult);
-      const textBlocks = msg.content.filter((b) => b.text !== undefined && !b.toolResult);
+      const textBlocks = msg.content.filter(
+        (b) => b.text !== undefined && b.text !== "" && !b.toolResult,
+      );
 
       if (toolResults.length > 0) {
         for (const block of toolResults) {
@@ -298,21 +318,21 @@ export function converseToCompletionRequest(
 
       // Plain user message
       const text = msg.content
-        .filter((b) => b.text !== undefined)
+        .filter((b) => b.text !== undefined && b.text !== "")
         .map((b) => b.text ?? "")
         .join("");
       messages.push({ role: "user", content: text });
     } else if (msg.role === "assistant") {
       const toolUseBlocks = msg.content.filter((b) => b.toolUse);
       const textContent = msg.content
-        .filter((b) => b.text !== undefined)
+        .filter((b) => b.text !== undefined && b.text !== "")
         .map((b) => b.text ?? "")
         .join("");
 
       if (toolUseBlocks.length > 0) {
         messages.push({
           role: "assistant",
-          content: textContent || null,
+          content: textContent ?? null,
           tool_calls: toolUseBlocks.map((b) => ({
             id: b.toolUse!.toolUseId,
             type: "function" as const,
@@ -323,7 +343,12 @@ export function converseToCompletionRequest(
           })),
         });
       } else {
-        messages.push({ role: "assistant", content: textContent || null });
+        messages.push({ role: "assistant", content: textContent ?? null });
+      }
+    } else {
+      const warnMsg = `Unexpected message role "${msg.role}" in Converse request — skipping`;
+      if (logger) {
+        logger.warn(warnMsg);
       }
     }
   }
@@ -518,7 +543,7 @@ export async function handleConverse(
     return;
   }
 
-  const completionReq = converseToCompletionRequest(converseReq, modelId);
+  const completionReq = converseToCompletionRequest(converseReq, modelId, logger);
   completionReq._endpointType = "chat";
 
   const testId = getTestId(req);
@@ -623,7 +648,7 @@ export async function handleConverse(
     const errBody = {
       type: "error",
       error: {
-        type: response.error.type || "invalid_request_error",
+        type: response.error.type ?? "invalid_request_error",
         message: response.error.message,
       },
     };
@@ -681,6 +706,11 @@ export async function handleConverse(
 
   // Tool call response
   if (isToolCallResponse(response)) {
+    if ("webSearches" in response) {
+      logger.warn(
+        "webSearches in fixture response are not supported for Bedrock Converse API — ignoring",
+      );
+    }
     const overrides = extractOverrides(response);
     journal.add({
       method: req.method ?? "POST",
@@ -775,7 +805,8 @@ export async function handleConverseStream(
     return;
   }
 
-  const completionReq = converseToCompletionRequest(converseReq, modelId);
+  const completionReq = converseToCompletionRequest(converseReq, modelId, logger);
+  completionReq.stream = true;
   completionReq._endpointType = "chat";
 
   const testId = getTestId(req);
@@ -882,7 +913,7 @@ export async function handleConverseStream(
     const errBody = {
       type: "error",
       error: {
-        type: response.error.type || "invalid_request_error",
+        type: response.error.type ?? "invalid_request_error",
         message: response.error.message,
       },
     };
@@ -968,6 +999,11 @@ export async function handleConverseStream(
 
   // Tool call response — stream as Event Stream
   if (isToolCallResponse(response)) {
+    if ("webSearches" in response) {
+      logger.warn(
+        "webSearches in fixture response are not supported for Bedrock Converse API — ignoring",
+      );
+    }
     const overrides = extractOverrides(response);
     const journalEntry = journal.add({
       method: req.method ?? "POST",