fix: structured logger for chaos/stream warnings; EventStream bounds;…

… bedrock SSE; body timeout

- chaos.ts: add optional logger param to resolveChaosConfig/evaluateChaos/applyChaos;
  replace all console.warn calls with logger?.warn
- stream-collapse.ts: logger param on collapseStreamingResponse; replace console.warn;
  add explicit case "bedrock" routing to collapseAnthropicSSE; add bounds check in
  decodeEventStreamFrames — return {frames, truncated:true} when totalLength extends
  past buffer, preventing out-of-bounds reads on malformed/truncated EventStream frames
- recorder.ts: pass defaults.logger to collapseStreamingResponse; add res.setTimeout
  body accumulation timeout (30s) to prevent unbounded memory growth on slow responses
- bedrock.ts: update module docstring to describe all four endpoint families
- all handlers: pass defaults.logger as final arg to all applyChaos call sites

src/bedrock-converse.ts

@@ -283,6 +283,7 @@ export async function handleConverse(
         body: completionReq,
       },
       defaults.registry,
+      defaults.logger,
     )
   )
     return;
@@ -485,6 +486,7 @@ export async function handleConverseStream(
         body: completionReq,
       },
       defaults.registry,
+      defaults.logger,
     )
   )
     return;

src/bedrock.ts

@@ -1,13 +1,20 @@
 /**
- * AWS Bedrock Claude endpoint support.
+ * AWS Bedrock Claude endpoint support — invoke and invoke-with-response-stream.
  *
- * Handles POST /model/{modelId}/invoke and /invoke-with-response-stream
- * requests. Translates incoming Bedrock Claude format into the
- * ChatCompletionRequest format used by the fixture router, and converts
- * fixture responses back into the appropriate Bedrock response format
- * (JSON for invoke, AWS Event Stream binary encoding for streaming).
+ * Handles four Bedrock endpoint families (split across two modules):
  *
- * See bedrock-converse.ts for /converse and /converse-stream support.
+ *   This file (bedrock.ts):
+ *     - POST /model/{modelId}/invoke                  — non-streaming invoke
+ *     - POST /model/{modelId}/invoke-with-response-stream — binary EventStream streaming
+ *
+ *   bedrock-converse.ts:
+ *     - POST /model/{modelId}/converse                — Converse API (non-streaming)
+ *     - POST /model/{modelId}/converse-stream         — Converse API (EventStream streaming)
+ *
+ * Translates incoming Bedrock Claude format into the ChatCompletionRequest
+ * format used by the fixture router, and converts fixture responses back into
+ * the appropriate Bedrock response format (JSON for invoke, AWS Event Stream
+ * binary encoding for streaming).
  */
 
 import type * as http from "node:http";
@@ -322,6 +329,7 @@ export async function handleBedrock(
         body: completionReq,
       },
       defaults.registry,
+      defaults.logger,
     )
   )
     return;
@@ -638,6 +646,7 @@ export async function handleBedrockStream(
         body: completionReq,
       },
       defaults.registry,
+      defaults.logger,
     )
   )
     return;

src/chaos.ts

@@ -11,6 +11,7 @@ import type * as http from "node:http";
 import type { ChaosAction, ChaosConfig, ChatCompletionRequest, Fixture } from "./types.js";
 import { writeErrorResponse } from "./sse-writer.js";
 import type { Journal } from "./journal.js";
+import type { Logger } from "./logger.js";
 import type { MetricsRegistry } from "./metrics.js";
 
 /**
@@ -21,6 +22,7 @@ function resolveChaosConfig(
   fixture: Fixture | null,
   serverDefaults?: ChaosConfig,
   rawHeaders?: http.IncomingHttpHeaders,
+  logger?: Logger,
 ): ChaosConfig {
   const base: ChaosConfig = { ...serverDefaults };
 
@@ -41,23 +43,23 @@ function resolveChaosConfig(
     if (typeof dropHeader === "string") {
       const val = parseFloat(dropHeader);
       if (isNaN(val)) {
-        console.warn(`[chaos] x-llmock-chaos-drop: invalid value "${dropHeader}", ignoring`);
+        logger?.warn(`[chaos] x-llmock-chaos-drop: invalid value "${dropHeader}", ignoring`);
       } else {
         if (val < 0 || val > 1) {
-          console.warn(`[chaos] x-llmock-chaos-drop: value ${val} out of range [0,1], clamping`);
+          logger?.warn(`[chaos] x-llmock-chaos-drop: value ${val} out of range [0,1], clamping`);
         }
         base.dropRate = Math.min(1, Math.max(0, val));
       }
     }
     if (typeof malformedHeader === "string") {
       const val = parseFloat(malformedHeader);
       if (isNaN(val)) {
-        console.warn(
+        logger?.warn(
           `[chaos] x-llmock-chaos-malformed: invalid value "${malformedHeader}", ignoring`,
         );
       } else {
         if (val < 0 || val > 1) {
-          console.warn(
+          logger?.warn(
             `[chaos] x-llmock-chaos-malformed: value ${val} out of range [0,1], clamping`,
           );
         }
@@ -67,12 +69,12 @@ function resolveChaosConfig(
     if (typeof disconnectHeader === "string") {
       const val = parseFloat(disconnectHeader);
       if (isNaN(val)) {
-        console.warn(
+        logger?.warn(
           `[chaos] x-llmock-chaos-disconnect: invalid value "${disconnectHeader}", ignoring`,
         );
       } else {
         if (val < 0 || val > 1) {
-          console.warn(
+          logger?.warn(
             `[chaos] x-llmock-chaos-disconnect: value ${val} out of range [0,1], clamping`,
           );
         }
@@ -100,8 +102,9 @@ export function evaluateChaos(
   fixture: Fixture | null,
   serverDefaults?: ChaosConfig,
   rawHeaders?: http.IncomingHttpHeaders,
+  logger?: Logger,
 ): ChaosAction | null {
-  const config = resolveChaosConfig(fixture, serverDefaults, rawHeaders);
+  const config = resolveChaosConfig(fixture, serverDefaults, rawHeaders, logger);
 
   if (config.dropRate !== undefined && config.dropRate > 0 && Math.random() < config.dropRate) {
     return "drop";
@@ -143,8 +146,9 @@ export function applyChaos(
   journal: Journal,
   context: ChaosJournalContext,
   registry?: MetricsRegistry,
+  logger?: Logger,
 ): boolean {
-  const action = evaluateChaos(fixture, serverDefaults, rawHeaders);
+  const action = evaluateChaos(fixture, serverDefaults, rawHeaders, logger);
   if (!action) return false;
 
   if (registry) {

src/cohere.ts

@@ -485,6 +485,7 @@ export async function handleCohere(
         body: completionReq,
       },
       defaults.registry,
+      defaults.logger,
     )
   )
     return;

src/embeddings.ts

@@ -106,6 +106,7 @@ export async function handleEmbeddings(
         body: syntheticReq,
       },
       defaults.registry,
+      defaults.logger,
     )
   )
     return;

src/gemini.ts

@@ -436,6 +436,7 @@ export async function handleGemini(
         body: completionReq,
       },
       defaults.registry,
+      defaults.logger,
     )
   )
     return;

src/messages.ts

@@ -484,6 +484,7 @@ export async function handleMessages(
         body: completionReq,
       },
       defaults.registry,
+      defaults.logger,
     )
   )
     return;

src/ollama.ts

@@ -362,6 +362,7 @@ export async function handleOllama(
         body: completionReq,
       },
       defaults.registry,
+      defaults.logger,
     )
   )
     return;
@@ -604,6 +605,7 @@ export async function handleOllamaGenerate(
         body: completionReq,
       },
       defaults.registry,
+      defaults.logger,
     )
   )
     return;

src/recorder.ts

@@ -107,6 +107,7 @@ export async function proxyAndRecord(
     ctString,
     providerKey,
     isBinaryStream ? rawBuffer : upstreamBody,
+    defaults.logger,
   );
 
   let fixtureResponse: FixtureResponse;
@@ -214,6 +215,7 @@ function makeUpstreamRequest(
   return new Promise((resolve, reject) => {
     const transport = target.protocol === "https:" ? https : http;
     const UPSTREAM_TIMEOUT_MS = 30_000;
+    const BODY_TIMEOUT_MS = 30_000;
     const req = transport.request(
       target,
       {
@@ -225,6 +227,9 @@ function makeUpstreamRequest(
         },
       },
       (res) => {
+        res.setTimeout(BODY_TIMEOUT_MS, () => {
+          req.destroy(new Error(`Upstream response timed out after ${BODY_TIMEOUT_MS / 1000}s`));
+        });
         const chunks: Buffer[] = [];
         res.on("data", (chunk: Buffer) => chunks.push(chunk));
         res.on("error", reject);

src/responses.ts

@@ -547,6 +547,7 @@ export async function handleResponses(
         body: completionReq,
       },
       defaults.registry,
+      defaults.logger,
     )
   )
     return;

src/server.ts

@@ -202,6 +202,7 @@ async function handleCompletions(
         body,
       },
       defaults.registry,
+      defaults.logger,
     )
   )
     return;

src/stream-collapse.ts

@@ -8,6 +8,7 @@
 
 import { crc32 } from "node:zlib";
 import type { RecordProviderKey, ToolCall } from "./types.js";
+import type { Logger } from "./logger.js";
 
 // ---------------------------------------------------------------------------
 // Result type shared by all collapse functions
@@ -438,6 +439,11 @@ function decodeEventStreamFrames(buf: Buffer): {
     const totalLength = buf.readUInt32BE(offset);
     const headersLength = buf.readUInt32BE(offset + 4);
 
+    // Validate bounds: ensure the full frame is within the buffer
+    if (totalLength < 12 || offset + totalLength > buf.length) {
+      return { frames, truncated: true };
+    }
+
     // Validate prelude CRC
     const preludeCrc = buf.readUInt32BE(offset + 8);
     const computedPreludeCrc = crc32(buf.subarray(offset, offset + 8));
@@ -611,6 +617,7 @@ export function collapseStreamingResponse(
   contentType: string,
   providerKey: RecordProviderKey,
   body: string | Buffer,
+  logger?: Logger,
 ): CollapseResult | null {
   const ct = contentType.toLowerCase();
 
@@ -637,8 +644,10 @@ export function collapseStreamingResponse(
         return collapseGeminiSSE(str);
       case "cohere":
         return collapseCohereSSE(str);
+      case "bedrock":
+        return collapseAnthropicSSE(str);
       default:
-        console.warn(
+        logger?.warn(
           `[stream-collapse] unknown SSE provider "${providerKey}", falling back to OpenAI SSE format`,
         );
         return collapseOpenAISSE(str);