fix(adapters): validate buffered response and ndjson frame shapes (#2531)

src/adapters/anthropic.ts

@@ -265,6 +265,39 @@ export function formatAnthropicErrorBody(status: number, _headers: Headers, payl
   return redactSecretString(detail).slice(0, 400);
 }
 
+function isAnthropicRecord(value: unknown): value is Record<string, unknown> {
+  return value !== null && typeof value === "object" && !Array.isArray(value);
+}
+
+function anthropicStructuralValueType(value: unknown): string {
+  if (value === null) return "null";
+  return Array.isArray(value) ? "array" : typeof value;
+}
+
+interface InvalidAnthropicShapeDiagnostic {
+  reason: "content_not_array" | "content_block_not_object";
+  blockIndex?: number;
+  valueType: string;
+}
+
+/**
+ * Structured refusal for a malformed buffered response body, shaped like the google adapter's
+ * `invalidGoogleShapeEvent` so an operator reading a log can tell which rung failed: the content
+ * container, or one block inside an otherwise well-formed container.
+ */
+function invalidAnthropicShapeEvent(
+  diagnostic: InvalidAnthropicShapeDiagnostic,
+): Extract<AdapterEvent, { type: "error" }> {
+  const at = diagnostic.blockIndex !== undefined ? `; blockIndex=${diagnostic.blockIndex}` : "";
+  const subject = diagnostic.reason === "content_not_array" ? "content" : "content block";
+  return {
+    type: "error",
+    message: `anthropic response contained invalid ${subject} (${diagnostic.reason}${at}; valueType=${diagnostic.valueType})`,
+    status: 502,
+    errorType: "upstream_error",
+  };
+}
+
 function extractAnthropicErrorDetail(parsed: unknown): string | undefined {
   if (typeof parsed === "string") return parsed.trim() || undefined;
   if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) return undefined;
@@ -1279,12 +1312,55 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti
     },
 
     async parseResponse(response: Response, budget: TranslatorBudget): Promise<AdapterEvent[]> {
-      const json = await response.json() as Record<string, unknown>;
+      const parsed: unknown = await response.json();
+      // `response.json()` resolves a body of `null` to `null` without throwing, so the cast below
+      // used to reach `json.content` on it — the #1219 defect at the buffered body root. The
+      // streaming parser has skipped a non-record frame since #1240, but a buffered body has no
+      // next frame to recover into, so this fails closed instead.
+      if (!isAnthropicRecord(parsed)) {
+        return [{
+          type: "error",
+          message: `anthropic response was not a JSON object (${anthropicStructuralValueType(parsed)})`,
+          status: 502,
+          errorType: "upstream_error",
+        }];
+      }
+      const json = parsed;
       const responseBytes = new TextEncoder().encode(JSON.stringify(json)).byteLength;
       budget.chargeRetained(responseBytes, { kind: "retained_collectors" });
       try {
       const events: AdapterEvent[] = [];
-      const content = json.content as { type: string; text?: string; id?: string; name?: string; input?: unknown; thinking?: string; reasoning?: string; signature?: string; data?: string }[] | undefined;
+      // Same retain-then-return tail every other terminal branch in this function uses; naming it
+      // keeps the two shape guards below from drifting out of step with the accounting.
+      const finishWithEvents = (batch: AdapterEvent[]): AdapterEvent[] => {
+        retainTranslatedEventBatch(batch, budget);
+        return batch;
+      };
+      const rawContent: unknown = json.content;
+      // `content` is claimed model output inside a well-formed response, so it is governed by the
+      // #1332 nested-shape rule (fail closed) rather than #1240's root-frame padding rule (skip).
+      // Absence stays legal in both encodings. A present non-array was silently accepted and is
+      // the dangerous case: a string is iterable, so `for (const block of "text")` walked it one
+      // CHARACTER at a time, read `undefined` from every `block.type`, and completed the turn as a
+      // successful empty response — the #2231 failure mode on a different adapter.
+      if (rawContent !== undefined && rawContent !== null && !Array.isArray(rawContent)) {
+        return finishWithEvents([invalidAnthropicShapeEvent({
+          reason: "content_not_array",
+          valueType: anthropicStructuralValueType(rawContent),
+        })]);
+      }
+      if (Array.isArray(rawContent)) {
+        for (let blockIndex = 0; blockIndex < rawContent.length; blockIndex++) {
+          if (!isAnthropicRecord(rawContent[blockIndex])) {
+            return finishWithEvents([invalidAnthropicShapeEvent({
+              reason: "content_block_not_object",
+              blockIndex,
+              valueType: anthropicStructuralValueType(rawContent[blockIndex]),
+            })]);
+          }
+        }
+      }
+      const content = rawContent as { type: string; text?: string; id?: string; name?: string; input?: unknown; thinking?: string; reasoning?: string; signature?: string; data?: string }[] | undefined;
       if (content) {
         for (const block of content) {
           if (block.type === "text" && block.text) {

src/adapters/command-code.ts

@@ -7,6 +7,7 @@ import { isAllowedToolChoice, namespacedToolName, resolveToolChoiceWireName, too
 import type { AdapterFetchContext, AdapterRequest, ProviderAdapter } from "./base";
 import type { TranslatorBudget } from "../lib/translator-budget";
 import { readBoundedResponseBody } from "../lib/bounded-body";
+import { debugDroppedFrame } from "../lib/debug";
 import { configuredReasoningEfforts } from "../reasoning-effort";
 import { commandCodeReasoningEfforts, refreshCommandCodeReasoningEfforts } from "../providers/command-code-efforts";
 import { identifyRoutedModel } from "./identity";
@@ -365,7 +366,7 @@ async function*ndjson(response: Response, budget: TranslatorBudget): AsyncGenera
       let newline = buffer.indexOf("\n");
       while (newline >= 0) {
         const line = buffer.slice(0, newline).trim(); buffer = buffer.slice(newline + 1);
-        if (line) { try { yield JSON.parse(stripEventFrame(line)) as Record<string, unknown>; } catch { /* ignore non-events */ } }
+        if (line) yield* decodeEventLine(line);
         newline = buffer.indexOf("\n");
       }
       const residualBytes = encoder.encode(buffer).byteLength;
@@ -376,14 +377,49 @@ async function*ndjson(response: Response, budget: TranslatorBudget): AsyncGenera
       if (done) break;
     }
     const final = buffer.trim();
-    if (final) { try { yield JSON.parse(stripEventFrame(final)) as Record<string, unknown>; } catch { /* ignore */ } }
+    if (final) yield* decodeEventLine(final);
   } finally {
     budget.releaseRetained(bufferBytes, { kind: "live_transient" });
     try { await reader.cancel(); } catch { /* already closed */ }
     reader.releaseLock();
   }
 }
 
+/**
+ * Yield one NDJSON line as an event record, or nothing.
+ *
+ * `JSON.parse("null")` returns `null` instead of throwing, so the `try/catch` around the parse
+ * cannot see it and the `event.type` read in parseStream crashed the turn — the #1219 defect, on
+ * the one streaming transport the #1240 audit did not cover because it is NDJSON rather than SSE.
+ *
+ * A frame that does not parse to a record is padding, not an event: drop it and continue exactly
+ * as an unparseable line is already dropped, so a stream whose only frames are junk ends in the
+ * same single terminal `done` as an empty body. Skipping is what preserves an answer whose deltas
+ * have already arrived — the observed #1219 case is `null` padding BETWEEN content deltas, where
+ * terminating would discard a complete response (#1240).
+ *
+ * Note this deliberately makes a junk-only stream a quiet `[done]` where it previously threw. That
+ * throw was an unguarded type assumption, not a designed failure signal, and `[done]` is already
+ * what an empty body, a blank-line-only body and an unparseable-only body all produce here. The
+ * broader question — whether this adapter should report *any* no-valid-event stream as a failure
+ * rather than an empty success — is pre-existing, applies to all four of those inputs equally, and
+ * is deliberately not decided by this change.
+ */
+function* decodeEventLine(line: string): Generator<Record<string, unknown>> {
+  let parsed: unknown;
+  try {
+    parsed = JSON.parse(stripEventFrame(line));
+  } catch {
+    debugDroppedFrame("command-code", line);
+    return;
+  }
+  if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) {
+    debugDroppedFrame("command-code", line);
+    return;
+  }
+  yield parsed as Record<string, unknown>;
+}
+
 /** The endpoint is newline-delimited JSON; defensively strip an SSE `data:` frame if the gateway ever switches shapes. */
 function stripEventFrame(line: string): string {
   return line.startsWith("data:") ? line.slice("data:".length).trim() : line;

src/adapters/google.ts

@@ -1158,7 +1158,18 @@ export function createGoogleAdapter(provider: OcxProviderConfig): ProviderAdapte
       let raw: Record<string, unknown>;
       let rawBytes = 0;
       try {
-        raw = JSON.parse(rawText) as Record<string, unknown>;
+        const parsedRaw: unknown = JSON.parse(rawText);
+        // `JSON.parse("null")` returns null instead of throwing, so the catch below cannot see it
+        // and the `raw.error` read crashed the turn — #1219 at the buffered body root, which #1240
+        // never reached because that audit swept SSE frame parsers only. There is no next frame to
+        // recover into here, so unlike a stream frame this fails closed, matching the
+        // unparseable-body branch just below and the buffered candidate guards added in #2232.
+        if (!isGoogleRecord(parsedRaw)) {
+          budget.releaseRetained(rawTextBytes, { kind: "retained_collectors" });
+          const valueType = googleStructuralValueType(parsedRaw);
+          return [{ type: "error", message: `google response was not a JSON object (${valueType})` }];
+        }
+        raw = parsedRaw;
         rawBytes = new TextEncoder().encode(JSON.stringify(raw)).byteLength;
         const rawReservation = budget.reserveTransient(rawBytes, { kind: "retained_collectors" });
         rawReservation.commitRetained();

src/adapters/openai-chat.ts

@@ -1932,8 +1932,28 @@ export function createOpenAIChatAdapter(provider: OcxProviderConfig): ProviderAd
         const choice = rawChoice;
         if (choice.finish_reason === "error") return [upstreamErrorEvent(choice.error, usage)];
         if (!choice.message) return [{ type: "error", message: "upstream response contained no choices", ...(usage ? { usage } : {}) }];
+        // `!choice.message` splits this input class on TRUTHINESS, not on shape: `null` and `0` fail
+        // closed here, while `"text"`, `true` and `[{...}]` pass and every property read below yields
+        // `undefined` — so a choice claiming an assistant message completed as a SUCCESSFUL EMPTY
+        // turn, stranding any tool call it claimed. The one line above already validates the choice
+        // container this way; its message was left on a truthiness test.
+        //
+        // Every non-record is rejected, arrays included. The google adapter does carve out `[]` for
+        // `content`, but that carve-out is specific to a protobuf-derived wire where a repeated
+        // field can spell an empty message — and `content` is genuinely an ARRAY of blocks there.
+        // `message` is a record on a plain-JSON wire that already has `{}`, so importing the
+        // exception would be an analogy rather than evidence. `[{"content":"…"}]` is the case that
+        // matters: it discards a complete answer, #2232's `content: [{ parts: [...] }]` one adapter over.
+        //
+        // Read through `unknown` rather than the declared type: `choices` is a cast over wire data,
+        // so its `message?: Record<string, unknown>` is an assertion the upstream never made, and
+        // narrowing against it is what let the missing check look type-safe.
+        const rawMessage: unknown = choice.message;
+        if (!isRecord(rawMessage)) {
+          return [invalidChoicesEvent(usage)];
+        }
 
-        const msg = choice.message;
+        const msg = rawMessage as Record<string, unknown>;
         const reasoningText = reasoningTextFrom(msg);
         if (reasoningText !== undefined) events.push({ type: "reasoning_raw_delta", text: reasoningText });
         if (typeof msg.content === "string") events.push({ type: "text_delta", text: msg.content });