fix(adapters): thread abort signal into image normalization call sites

src/adapters/anthropic-image-codec.ts

@@ -52,6 +52,8 @@ const PASSTHROUGH_MEDIA = new Set(["image/jpeg", "image/png", "image/gif", "imag
 export interface NormalizeOptions {
   /** Shift every image's starting ladder position down (413 retry tightening; 030). */
   tierBias?: number;
+  /** Cancels queued native decode work and stops pulling more images. */
+  abortSignal?: AbortSignal;
   /** Test seam: replaces the Bun.Image encode path (audit round 1, blocker 6). */
   encode?: EncodeFn;
   /** Test seam: replaces the pass-through decode validation (C-gate round 1, blocker 1). */

src/adapters/anthropic-image-normalize.ts

@@ -111,8 +111,6 @@ export interface NormalizeTarget {
 }
 
 export interface NormalizeTargetsOptions extends NormalizeOptions {
-  /** Cancels queued native decode work and stops pulling more images. */
-  abortSignal?: AbortSignal;
   /** Total base64 budget across all targets. Default: TOTAL_IMAGE_BASE64_BUDGET. */
   budget?: number;
   /**
@@ -144,6 +142,9 @@ export async function normalizeImageTargets(targets: NormalizeTarget[], options:
     if (abortSignal?.aborted) throw abortSignal.reason ?? new DOMException("Aborted", "AbortError");
     const leave = await enterImageDecode(abortSignal);
     try {
+      // Re-check after admission: an abort landing between dequeue and decode-start
+      // must not begin decoding (an in-flight native decode cannot be interrupted).
+      if (abortSignal?.aborted) throw abortSignal.reason ?? new DOMException("Aborted", "AbortError");
       const result = await processAt(b64, pos, mediaType, encode, validate);
       if (abortSignal?.aborted) throw abortSignal.reason ?? new DOMException("Aborted", "AbortError");
       return result;

src/adapters/anthropic.ts

@@ -945,7 +945,10 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti
       // Primary image layer: resize/re-encode to fit Anthropic limits without dropping
       // (anthropic-image-normalize.ts); the guard below remains the deterministic backstop.
       // imageTierBias > 0 = upstream-413 tightened retry (030): start every image one tier lower.
-      await normalizeAnthropicImages(messages, { tierBias: incoming?.imageTierBias ?? 0 });
+      await normalizeAnthropicImages(messages, {
+        tierBias: incoming?.imageTierBias ?? 0,
+        abortSignal: incoming?.abortSignal,
+      });
       // Anthropic rejects many-image requests (>20 images) carrying any image over
       // 2000px per side; see anthropic-image-guard.ts for the full limit policy.
       enforceAnthropicImageLimits(messages);

src/server/claude-messages.ts

@@ -420,7 +420,7 @@ async function anthropicNativePassthrough(
   // count_tokens too — counts must match what the real send will contain, and the 32MB
   // body cap applies to it equally. Non-message bodies pass through untouched.
   if (Array.isArray(body.messages)) {
-    await normalizeAnthropicImages(body.messages);
+    await normalizeAnthropicImages(body.messages, { abortSignal: req.signal });
     enforceAnthropicImageLimits(body.messages);
   }
   const headers = new Headers();

tests/adapters/anthropic/anthropic-image-normalize.test.ts

@@ -419,6 +419,65 @@ describe("bounded parallel first pass (WP170)", () => {
     expect(g.stats().peak).toBe(IMAGE_NORMALIZE_CONCURRENCY);
   });
 
+  test("a pre-aborted request rejects before any decode starts", async () => {
+    const controller = new AbortController();
+    controller.abort(new Error("client disconnected"));
+    let encodeCalls = 0;
+    await expect(normalizeAnthropicImages(
+      [userMsg([imageBlock(fakePngBase64(2100, 1500, 8192))])],
+      {
+        abortSignal: controller.signal,
+        encode: async () => {
+          encodeCalls++;
+          return { data: "unexpected", mediaType: "image/webp" };
+        },
+      },
+    )).rejects.toThrow("client disconnected");
+    expect(encodeCalls).toBe(0);
+  });
+
+  test("a request cancelled between dequeue and decode-start never begins decoding", async () => {
+    // Admission removes the waiter's abort listener, so an abort landing after the
+    // gate hands over a slot is only caught by the re-check before processAt. The
+    // blocker request parks every slot; the queued request is aborted from the
+    // blocker's replace callback — after admission, before its continuation runs.
+    const g = gatedEncoder();
+    const controller = new AbortController();
+    const images = distinctImages(IMAGE_NORMALIZE_CONCURRENCY + 1);
+    const blockerTargets: NormalizeTarget[] = images.slice(0, IMAGE_NORMALIZE_CONCURRENCY).map(b64 => ({
+      base64: b64,
+      mediaType: "image/png",
+      replace: () => controller.abort(new Error("client disconnected")),
+      drop: () => {},
+    }));
+    let queuedEncodes = 0;
+    const queuedTargets: NormalizeTarget[] = [{
+      base64: images[IMAGE_NORMALIZE_CONCURRENCY],
+      mediaType: "image/png",
+      replace: () => {},
+      drop: () => {},
+    }];
+
+    const blocker = normalizeImageTargets(blockerTargets, { encode: g.encode });
+    await g.waitForArrivals(IMAGE_NORMALIZE_CONCURRENCY);
+    const queued = normalizeImageTargets(queuedTargets, {
+      encode: async () => {
+        queuedEncodes++;
+        return { data: "unexpected", mediaType: "image/webp" };
+      },
+      abortSignal: controller.signal,
+    });
+    // Let the queued request's worker reach the decode queue before the gate releases.
+    await Bun.sleep(10);
+    g.release();
+    await blocker;
+    // Construct .rejects only once queued has settled: an unawaited .rejects on a
+    // still-pending promise starves the event loop (Bun busy-waits the assertion),
+    // so Bun.sleep/g.release() would never run — a deadlock, not an assertion failure.
+    await expect(queued).rejects.toThrow("client disconnected");
+    expect(queuedEncodes).toBe(0);
+  });
+
   test("a thrown target callback rejects the call, settles in-flight work, and stops new pulls", async () => {
     // processAt swallows encode/validate throws into {kind:"failed"} (its own catch),
     // so the production escape hatch is a throwing target callback (drop/replace).

tests/adapters/openai/openai-chat-image-normalization.test.ts

@@ -12,6 +12,8 @@ import {
   TIER_SPECS,
   type EncodeFn,
 } from "../../../src/adapters/anthropic-image-normalize";
+import { bunImageEncode, bunImageValidate } from "../../../src/adapters/anthropic-image-codec";
+import { sniffImageDimensions } from "../../../src/adapters/anthropic-image-guard";
 import type { OcxMessage, OcxParsedRequest, OcxProviderConfig } from "../../../src/types";
 import { createTestTranslatorBudget } from "../../helpers/translator-budget";
 import { phaseTimer } from "../../helpers/phase-timing";
@@ -101,6 +103,20 @@ function headerOnlyPngB64(width: number, height: number, base64Length: number):
   return bytes.toString("base64");
 }
 
+/** BMP header claiming the given dimensions; sniffImageDimensions has no BMP branch. */
+function headerOnlyBmpB64(width: number, height: number): string {
+  const bytes = Buffer.alloc(54);
+  bytes.write("BM", 0);
+  bytes.writeUInt32LE(bytes.length, 2);
+  bytes.writeUInt32LE(54, 10);
+  bytes.writeUInt32LE(40, 14);
+  bytes.writeInt32LE(width, 18);
+  bytes.writeInt32LE(height, 22);
+  bytes.writeUInt16LE(1, 26);
+  bytes.writeUInt16LE(24, 28);
+  return bytes.toString("base64");
+}
+
 
 /** Wrap raw base64 as a data URL, the only image form this wire normalizes. */
 function dataUrl(b64: string, mediaType = "image/png"): string {
@@ -273,6 +289,29 @@ describe("openai-chat inline image normalization", () => {
     expect(imageParts(messages as ChatMsg[])[0]?.image_url?.url).toBe(original);
   });
 
+  test("an oversized image whose header cannot be sniffed is rejected by the decode metadata bound", async () => {
+    // sniffImageDimensions reads only PNG/JPEG/GIF/WebP headers, so this header-only
+    // BMP claiming 5000x4000 (20MPx > MAX_INPUT_PIXELS) clears the pre-decode gates
+    // unsized. The bound that stops it is the metadata check inside the decode path
+    // itself: bunImageValidate on the pass-through branch, bunImageEncode elsewhere.
+    const bmp = headerOnlyBmpB64(5_000, 4_000);
+    expect(sniffImageDimensions(bmp)).toBeNull();
+    const input = Uint8Array.from(Buffer.from(bmp, "base64"));
+    await expect(bunImageValidate(input)).rejects.toThrow("image dimensions exceed the safe decode limit");
+    await expect(bunImageEncode(input, TIER_SPECS[0], 80)).rejects.toThrow("image dimensions exceed the safe decode limit");
+
+    // End to end the normalizer drops it after one rejected decode attempt, and this
+    // wire retains the original bytes on drop.
+    const original = dataUrl(bmp, "image/bmp");
+    const messages = [{
+      role: "user",
+      content: [{ type: "image_url", image_url: { url: original } }],
+    }];
+    await normalizeOpenAIChatImages(messages);
+    expect(getNormalizeStatsForTests().encodeCalls).toBe(1);
+    expect(imageParts(messages as ChatMsg[])[0]?.image_url?.url).toBe(original);
+  });
+
   test("an already-cancelled oversized build does not start normalization", async () => {
     const big = headerOnlyPngB64(1000, 1000, OPENAI_CHAT_IMAGE_BASE64_BUDGET + 4);
     const controller = new AbortController();