feat(compression): isolate sync engines in bounded worker pool (#11318)

Validated on a 17-PR combined board: compression-worker + colocate-standalone-esm-scope within the board's 287/287, typecheck:core clean, env-doc-sync clean. Offloads eligible sync compression engines into a bounded worker_threads pool with a strict serializable DTO boundary and fail-open on spawn/worker/timeout failure. Closes #11023. Thank you @RaviTharuma!

.env.example

@@ -1027,6 +1027,16 @@ PROVIDER_LIMITS_SYNC_SPACING_MS=1500
 # Used by: open-sse/services/compression/engines/rtk/filterLoader.ts.  Default: 0.
 #OMNIROUTE_RTK_TRUST_PROJECT_FILTERS=0
 
+# Maximum concurrent synchronous compression workers. Excess jobs wait FIFO.
+# Used by: open-sse/services/compression/compressionWorkerPool.ts. Default: 2.
+#OMNI_COMPRESSION_WORKERS=2
+# Per-job worker timeout (ms). A timed-out worker is terminated and the request fails open.
+# Used by: open-sse/services/compression/compressionWorkerPool.ts. Default: 120000.
+#OMNI_COMPRESSION_WORKER_TIMEOUT_MS=120000
+# Terminate idle compression workers after this many milliseconds.
+# Used by: open-sse/services/compression/compressionWorkerPool.ts. Default: 60000.
+#OMNI_COMPRESSION_WORKER_IDLE_MS=60000
+
 # T02 stacked-pipeline engine circuit-breaker (OPT-IN, default off). When enabled, a compression
 # engine that throws repeatedly across requests is skipped (fail-open) for a cooldown.
 # Used by: open-sse/services/compression/pipelineEngineBreaker.ts.

changelog.d/features/11023-compression-worker-pool.md

@@ -0,0 +1,3 @@
+- Run synchronous RTK and Caveman request compression in a bounded worker-thread pool, keeping
+  large `/v1/responses` compression heaps outside the HTTP isolate while preserving strict
+  fail-open behavior and per-engine telemetry.

docs/reference/ENVIRONMENT.md

@@ -531,6 +531,9 @@ detection above).
 | `OMNIROUTE_CONFIG_HOT_RELOAD_MS`                | `5000`                                              | `src/lib/config/hotReload.ts`                               | Polling interval (ms) for config hot-reload. Lower than `1000` is rejected.                                                                                             |
 | `OMNIROUTE_DISABLE_REDIS_AUTH_CACHE`            | _(enabled)_                                         | `src/lib/db/apiKeys.ts`                                     | Set `1` to bypass the Redis-backed API-key auth cache (forces DB reads).                                                                                                |
 | `OMNIROUTE_RTK_TRUST_PROJECT_FILTERS`           | `0`                                                 | `open-sse/services/compression/engines/rtk/filterLoader.ts` | Trust user-managed RTK project filter rules without strict signature checks.                                                                                            |
+| `OMNI_COMPRESSION_WORKERS`                      | `2`                                                 | `open-sse/services/compression/compressionWorkerPool.ts`    | Maximum concurrent synchronous RTK/Caveman workers; excess jobs wait FIFO.                                                                                               |
+| `OMNI_COMPRESSION_WORKER_TIMEOUT_MS`            | `120000`                                            | `open-sse/services/compression/compressionWorkerPool.ts`    | Per-job timeout in milliseconds. Timed-out workers are terminated and the request fails open unchanged.                                                                  |
+| `OMNI_COMPRESSION_WORKER_IDLE_MS`               | `60000`                                             | `open-sse/services/compression/compressionWorkerPool.ts`    | Idle lifetime in milliseconds before an unused compression worker is terminated.                                                                                        |
 | `COMPRESSION_PIPELINE_BREAKER_ENABLED`          | `false`                                             | `open-sse/services/compression/pipelineEngineBreaker.ts`    | T02 stacked-pipeline per-engine circuit-breaker master switch. **Opt-in (default off)** — when on, an engine that throws repeatedly across requests is skipped (fail-open) for a cooldown; off = byte-identical legacy behavior. |
 | `COMPRESSION_PIPELINE_BREAKER_THRESHOLD`        | `3`                                                 | `open-sse/services/compression/pipelineEngineBreaker.ts`    | Consecutive cross-request failures before an engine's breaker opens.                                                                                                    |
 | `COMPRESSION_PIPELINE_BREAKER_COOLDOWN_MS`      | `30000`                                             | `open-sse/services/compression/pipelineEngineBreaker.ts`    | Milliseconds an opened engine stays skipped before a half-open probe.                                                                                                   |

open-sse/services/compression/compressionWorker.ts

@@ -0,0 +1,40 @@
+import { parentPort } from "node:worker_threads";
+import {
+  applyCompression,
+  applyStackedCompression,
+  type StackedCompressionStep,
+} from "./strategySelector.ts";
+import type {
+  CompressionWorkerJob,
+  CompressionWorkerMessage,
+} from "./compressionWorkerProtocol.ts";
+
+if (!parentPort) throw new Error("compressionWorker must run in a worker thread");
+parentPort.on("message", (job: CompressionWorkerJob) => {
+  try {
+    const onEngineStep = (step: StackedCompressionStep) =>
+      parentPort.postMessage({
+        id: job.id,
+        type: "step",
+        step,
+      } satisfies CompressionWorkerMessage);
+    const result =
+      job.mode === "stacked"
+        ? applyStackedCompression(job.body, job.options?.config?.stackedPipeline, {
+            ...job.options,
+            onEngineStep,
+          })
+        : applyCompression(job.body, job.mode, job.options);
+    parentPort.postMessage({
+      id: job.id,
+      type: "result",
+      result,
+    } satisfies CompressionWorkerMessage);
+  } catch (error) {
+    parentPort.postMessage({
+      id: job.id,
+      type: "error",
+      error: error instanceof Error ? error.message : String(error),
+    } satisfies CompressionWorkerMessage);
+  }
+});

open-sse/services/compression/compressionWorkerPool.ts

@@ -0,0 +1,164 @@
+import { existsSync } from "node:fs";
+import { dirname, join } from "node:path";
+import { fileURLToPath } from "node:url";
+import { Worker } from "node:worker_threads";
+import type { CompressionResult } from "./types.ts";
+import type { StackedCompressionStep } from "./strategySelector.ts";
+import type {
+  CompressionWorkerJob,
+  CompressionWorkerMessage,
+  CompressionWorkerOptions,
+} from "./compressionWorkerProtocol.ts";
+
+function positiveInteger(value: string | undefined, fallback: number): number {
+  const parsed = Number(value);
+  return Number.isSafeInteger(parsed) && parsed > 0 ? parsed : fallback;
+}
+function workerUrl(): URL {
+  const dir = dirname(fileURLToPath(import.meta.url));
+  for (const name of ["compressionWorker.js", "compressionWorker.ts"]) {
+    if (existsSync(join(dir, name))) return new URL(name, import.meta.url);
+  }
+  return new URL("compressionWorker.js", import.meta.url);
+}
+function unchanged(body: Record<string, unknown>): CompressionResult {
+  return { body, compressed: false, stats: null };
+}
+interface PendingJob extends CompressionWorkerJob {
+  originalBody: Record<string, unknown>;
+  resolve: (result: CompressionResult) => void;
+  onEngineStep?: (step: StackedCompressionStep) => void;
+}
+interface PoolWorker {
+  worker: Worker;
+  job: PendingJob | null;
+  timeout: NodeJS.Timeout | null;
+  idle: NodeJS.Timeout | null;
+}
+
+export class CompressionWorkerPool {
+  private readonly queue: PendingJob[] = [];
+  private readonly workers = new Set<PoolWorker>();
+  private nextId = 1;
+  private readonly size: number;
+  private readonly timeoutMs: number;
+  private readonly idleMs: number;
+
+  constructor({
+    size = positiveInteger(process.env.OMNI_COMPRESSION_WORKERS, 2),
+    timeoutMs = positiveInteger(process.env.OMNI_COMPRESSION_WORKER_TIMEOUT_MS, 120_000),
+    idleMs = positiveInteger(process.env.OMNI_COMPRESSION_WORKER_IDLE_MS, 60_000),
+  }: { size?: number; timeoutMs?: number; idleMs?: number } = {}) {
+    this.size = Math.max(1, Math.floor(size));
+    this.timeoutMs = Math.max(1, Math.floor(timeoutMs));
+    this.idleMs = Math.max(1, Math.floor(idleMs));
+  }
+
+  run(
+    body: Record<string, unknown>,
+    mode: CompressionWorkerJob["mode"],
+    options?: CompressionWorkerOptions,
+    onEngineStep?: (step: StackedCompressionStep) => void
+  ): Promise<CompressionResult> {
+    return new Promise((resolve) => {
+      this.queue.push({
+        id: this.nextId++,
+        body,
+        mode,
+        options,
+        originalBody: body,
+        resolve,
+        onEngineStep,
+      });
+      this.dispatch();
+    });
+  }
+  async close(): Promise<void> {
+    for (const job of this.queue.splice(0)) job.resolve(unchanged(job.originalBody));
+    await Promise.all([...this.workers].map((slot) => this.remove(slot, true)));
+  }
+  private spawn(): PoolWorker {
+    const slot: PoolWorker = {
+      worker: new Worker(workerUrl()),
+      job: null,
+      timeout: null,
+      idle: null,
+    };
+    this.workers.add(slot);
+    slot.worker.on("message", (message: CompressionWorkerMessage) =>
+      this.handleMessage(slot, message)
+    );
+    slot.worker.on("error", () => this.fail(slot));
+    slot.worker.on("exit", () => {
+      if (this.workers.has(slot)) this.fail(slot);
+    });
+    return slot;
+  }
+  private dispatch(): void {
+    while (this.queue.length) {
+      let slot = [...this.workers].find((candidate) => !candidate.job);
+      if (!slot && this.workers.size < this.size) slot = this.spawn();
+      if (!slot) return;
+      if (slot.idle) clearTimeout(slot.idle);
+      const job = this.queue.shift();
+      if (!job) return;
+      slot.job = job;
+      slot.timeout = setTimeout(() => this.fail(slot!), this.timeoutMs);
+      slot.timeout.unref();
+      const { originalBody: _body, resolve: _resolve, onEngineStep: _step, ...wireJob } = job;
+      slot.worker.postMessage(wireJob);
+    }
+  }
+  private handleMessage(slot: PoolWorker, message: CompressionWorkerMessage): void {
+    const job = slot.job;
+    if (!job || job.id !== message.id) return;
+    if (message.type === "step") {
+      try {
+        job.onEngineStep?.(message.step);
+      } catch {
+        // Telemetry is best-effort.
+      }
+      return;
+    }
+    this.finish(slot, message.type === "result" ? message.result : unchanged(job.originalBody));
+  }
+  private finish(slot: PoolWorker, result: CompressionResult): void {
+    const job = slot.job;
+    if (!job) return;
+    if (slot.timeout) clearTimeout(slot.timeout);
+    slot.timeout = null;
+    slot.job = null;
+    job.resolve(result);
+    slot.idle = setTimeout(() => void this.remove(slot, false), this.idleMs);
+    slot.idle.unref();
+    this.dispatch();
+  }
+  private fail(slot: PoolWorker): void {
+    const job = slot.job;
+    if (job) job.resolve(unchanged(job.originalBody));
+    slot.job = null;
+    void this.remove(slot, true).finally(() => this.dispatch());
+  }
+  private async remove(slot: PoolWorker, terminate: boolean): Promise<void> {
+    if (!this.workers.delete(slot)) return;
+    if (slot.timeout) clearTimeout(slot.timeout);
+    if (slot.idle) clearTimeout(slot.idle);
+    if (terminate) await slot.worker.terminate().catch(() => undefined);
+  }
+}
+
+let pool: CompressionWorkerPool | null = null;
+export function runCompressionInWorker(
+  body: Record<string, unknown>,
+  mode: CompressionWorkerJob["mode"],
+  options?: CompressionWorkerOptions,
+  onEngineStep?: (step: StackedCompressionStep) => void
+): Promise<CompressionResult> {
+  pool ??= new CompressionWorkerPool();
+  return pool.run(body, mode, options, onEngineStep);
+}
+export async function closeCompressionWorkerPoolForTests(): Promise<void> {
+  const active = pool;
+  pool = null;
+  await active?.close();
+}

open-sse/services/compression/compressionWorkerProtocol.ts

@@ -0,0 +1,71 @@
+import type { CompressionConfig, CompressionMode, CompressionResult } from "./types.ts";
+import type { StackedCompressionStep } from "./strategySelector.ts";
+import type {
+  CompressionStage,
+  CompressionWireFormat,
+  ImageTransportFidelity,
+} from "./engines/types.ts";
+
+export interface CompressionWorkerOptions {
+  model?: string;
+  supportsVision?: boolean | null;
+  providerTransport?: "direct" | "aggregator";
+  provider?: string;
+  imageTransportFidelity?: ImageTransportFidelity;
+  sourceFormat?: CompressionWireFormat;
+  targetFormat?: CompressionWireFormat;
+  compressionStage?: CompressionStage;
+  config?: CompressionConfig;
+}
+export interface CompressionWorkerJob {
+  id: number;
+  body: Record<string, unknown>;
+  mode: CompressionMode;
+  options?: CompressionWorkerOptions;
+}
+export type CompressionWorkerMessage =
+  | { id: number; type: "step"; step: StackedCompressionStep }
+  | { id: number; type: "result"; result: CompressionResult }
+  | { id: number; type: "error"; error: string };
+
+function isPlainObject(value: object): value is Record<string, unknown> {
+  const prototype = Object.getPrototypeOf(value);
+  return prototype === Object.prototype || prototype === null;
+}
+export function isStrictlySerializable(value: unknown, seen = new Set<object>()): boolean {
+  if (
+    value === null ||
+    typeof value === "string" ||
+    typeof value === "boolean" ||
+    typeof value === "number"
+  ) {
+    return typeof value !== "number" || Number.isFinite(value);
+  }
+  if (typeof value !== "object" || seen.has(value)) return false;
+  seen.add(value);
+  if (Array.isArray(value)) return value.every((entry) => isStrictlySerializable(entry, seen));
+  if (!isPlainObject(value)) return false;
+  return Object.values(value).every((entry) => isStrictlySerializable(entry, seen));
+}
+
+const WORKER_STACK_ENGINES = new Set(["caveman", "rtk", "standard"]);
+export function isCompressionWorkerEligible(
+  body: Record<string, unknown>,
+  mode: CompressionMode,
+  options?: CompressionWorkerOptions
+): boolean {
+  if (mode !== "standard" && mode !== "rtk" && mode !== "stacked") return false;
+  if (mode === "stacked") {
+    const pipeline = options?.config?.stackedPipeline;
+    if (!Array.isArray(pipeline) || pipeline.length === 0) return false;
+    if (
+      pipeline.some((step) => {
+        const engine = typeof step === "string" ? step : step.engine;
+        return !WORKER_STACK_ENGINES.has(engine);
+      })
+    ) {
+      return false;
+    }
+  }
+  return isStrictlySerializable({ body, mode, ...(options ? { options } : {}) });
+}

open-sse/services/compression/strategySelector.ts

@@ -519,6 +519,28 @@ async function runCompressionAsync(
     cachingContext?: CachingDetectionContext;
   }
 ): Promise<CompressionResult> {
+  const workerOptions = options
+    ? {
+        model: options.model,
+        supportsVision: options.supportsVision,
+        providerTransport: options.providerTransport,
+        provider: options.provider,
+        imageTransportFidelity: options.imageTransportFidelity,
+        sourceFormat: options.sourceFormat,
+        targetFormat: options.targetFormat,
+        compressionStage: options.compressionStage,
+        config: options.config,
+      }
+    : undefined;
+  const { isCompressionWorkerEligible } = await import("./compressionWorkerProtocol.ts");
+  if (isCompressionWorkerEligible(body, mode, workerOptions)) {
+    try {
+      const { runCompressionInWorker } = await import("./compressionWorkerPool.ts");
+      return await runCompressionInWorker(body, mode, workerOptions, options?.onEngineStep);
+    } catch {
+      return { body, compressed: false, stats: null };
+    }
+  }
   if (
     options?.config?.memoizeCompressionResults === true &&
     // Only memoize for an explicit principal — a missing principalId would collapse

scripts/build/colocate-standalone.mjs

@@ -33,6 +33,14 @@ const STANDALONE = process.env.OMNIROUTE_STANDALONE_DIR
 
 const CALL_LOG_WORKER_REL = join("src", "lib", "usage", "callLogArtifactWorker.js");
 const CALL_LOG_WORKER_SRC = join(ROOT, "src", "lib", "usage", "callLogArtifactWorker.ts");
+const COMPRESSION_WORKER_REL = join("open-sse", "services", "compression", "compressionWorker.js");
+const COMPRESSION_WORKER_SRC = join(
+  ROOT,
+  "open-sse",
+  "services",
+  "compression",
+  "compressionWorker.ts"
+);
 const WORKER_REL = join(
   "open-sse",
   "services",
@@ -107,9 +115,26 @@ function main() {
   );
   console.log("[colocate-standalone] ✅ call-log artifact worker bundled");
 
+  const compressionWorkerDest = join(STANDALONE, COMPRESSION_WORKER_REL);
+  mkdirSync(dirname(compressionWorkerDest), { recursive: true });
+  runBuildTool(
+    "esbuild",
+    "esbuild",
+    [
+      COMPRESSION_WORKER_SRC,
+      "--bundle",
+      "--platform=node",
+      "--packages=external",
+      "--format=esm",
+      `--outfile=${compressionWorkerDest}`,
+    ],
+    { stdio: "inherit" }
+  );
+  console.log("[colocate-standalone] ✅ compression worker bundled");
+
   // The call-log worker is always present; scope it to ESM immediately. The
   // optional LLMLingua worker dir is added below only when its deps are installed.
-  const workerDirs = [dirname(callLogWorkerDest)];
+  const workerDirs = [dirname(callLogWorkerDest), dirname(compressionWorkerDest)];
 
   if (!hasOptionals) {
     console.log(

scripts/build/pack-artifact-policy.ts

@@ -45,6 +45,7 @@ export const APP_STAGING_ALLOWED_EXACT_PATHS: string[] = [
   // LLMLingua ONNX worker — esbuild'd standalone .js spawned via worker_threads
   // (the Next.js bundler can't trace the computed Worker path). Kept like the MCP server.
   "open-sse/services/compression/engines/llmlingua/onnxWorker.js",
+  "open-sse/services/compression/compressionWorker.js",
   "src/lib/usage/callLogArtifactWorker.js",
   "package.json",
   "peer-stamp.mjs",

scripts/build/prepare-electron-standalone.mjs

@@ -1,12 +1,13 @@
 #!/usr/bin/env node
 
-import { existsSync, lstatSync, readdirSync, rmSync } from "node:fs";
+import { existsSync, lstatSync, mkdirSync, readdirSync, rmSync } from "node:fs";
 import { basename, dirname, join, relative } from "node:path";
 import { fileURLToPath } from "node:url";
 import { assembleStandalone } from "./assembleStandalone.mjs";
 import { assertSqlitePrebuildExists } from "./electronRebuildPlan.mjs";
 import { pruneElectronRuntimeDocs } from "./electronRuntimeDocs.mjs";
 import { stageOptionalPacks } from "./optionalPackStaging.mjs";
+import { runBuildTool } from "./buildToolRunner.mjs";
 
 const __filename = fileURLToPath(import.meta.url);
 const __dirname = dirname(__filename);
@@ -169,6 +170,27 @@ assembleStandalone({
   // app they would point at the build machine's absolute paths and break on install.
   materializeSymlinks: true,
 });
+const compressionWorkerDest = join(
+  ELECTRON_STANDALONE_DIR,
+  "open-sse",
+  "services",
+  "compression",
+  "compressionWorker.js"
+);
+mkdirSync(dirname(compressionWorkerDest), { recursive: true });
+runBuildTool(
+  "esbuild",
+  "esbuild",
+  [
+    join(ROOT, "open-sse", "services", "compression", "compressionWorker.ts"),
+    "--bundle",
+    "--platform=node",
+    "--packages=external",
+    "--format=esm",
+    `--outfile=${compressionWorkerDest}`,
+  ],
+  { stdio: "inherit" }
+);
 
 const docsPrune = pruneElectronRuntimeDocs(ELECTRON_STANDALONE_DIR);
 if (docsPrune.removedFiles > 0) {

scripts/build/prepublish.ts

@@ -407,6 +407,40 @@ if (existsSync(llmWorkerSrc)) {
   }
 }
 
+// ── Step 8.6b: Bundle synchronous compression worker ──────────────────
+const compressionWorkerSrc = join(
+  ROOT,
+  "open-sse",
+  "services",
+  "compression",
+  "compressionWorker.ts"
+);
+const compressionWorkerDest = join(
+  DIST_DIR,
+  "open-sse",
+  "services",
+  "compression",
+  "compressionWorker.js"
+);
+if (!existsSync(compressionWorkerSrc)) {
+  throw new Error("Required compression worker source is missing");
+}
+console.log("  🔨 Bundling compression worker...");
+mkdirSync(dirname(compressionWorkerDest), { recursive: true });
+runBuildTool(
+  "esbuild",
+  "esbuild",
+  [
+    "open-sse/services/compression/compressionWorker.ts",
+    "--bundle",
+    "--platform=node",
+    "--packages=external",
+    "--format=esm",
+    "--outfile=dist/open-sse/services/compression/compressionWorker.js",
+  ],
+  { cwd: ROOT, stdio: "inherit" }
+);
+
 // ── Step 8.7: Bundle CLI Entrypoint ──────────────────────────
 const cliSrcFile = join(ROOT, "bin", "omniroute.ts");
 const cliDestFile = join(ROOT, "bin", "omniroute.mjs");

tests/unit/build/colocate-standalone-esm-scope.test.ts

@@ -127,3 +127,20 @@ test("scoped layout runs a CJS server.js and an ESM worker.js side by side", ()
     rmSync(root, { recursive: true, force: true });
   }
 });
+
+test("colocate-standalone bundles the required compression worker", () => {
+  const root = mkdtempSync(join(tmpdir(), "colocate-compression-worker-"));
+  try {
+    writeFileSync(join(root, "server.js"), "module.exports = {};\n");
+    execFileSync(process.execPath, ["scripts/build/colocate-standalone.mjs"], {
+      cwd: join(import.meta.dirname, "..", "..", ".."),
+      env: { ...process.env, OMNIROUTE_STANDALONE_DIR: root },
+      stdio: "pipe",
+    });
+    const workerDir = join(root, "open-sse", "services", "compression");
+    assert.equal(existsSync(join(workerDir, "compressionWorker.js")), true);
+    assert.equal(JSON.parse(readFileSync(join(workerDir, "package.json"), "utf8")).type, "module");
+  } finally {
+    rmSync(root, { recursive: true, force: true });
+  }
+});

tests/unit/compression/compression-worker.test.ts

@@ -0,0 +1,161 @@
+import assert from "node:assert/strict";
+import { after, describe, it } from "node:test";
+import {
+  isCompressionWorkerEligible,
+  isStrictlySerializable,
+} from "../../../open-sse/services/compression/compressionWorkerProtocol.ts";
+import {
+  closeCompressionWorkerPoolForTests,
+  CompressionWorkerPool,
+} from "../../../open-sse/services/compression/compressionWorkerPool.ts";
+import {
+  applyCompression,
+  applyCompressionAsync,
+} from "../../../open-sse/services/compression/strategySelector.ts";
+import type { CompressionConfig } from "../../../open-sse/services/compression/types.ts";
+
+const body = {
+  model: "gpt-test",
+  messages: [
+    { role: "system", content: "Answer accurately." },
+    {
+      role: "user",
+      content:
+        "Please basically actually simply carefully help with this very important task. ".repeat(
+          80
+        ),
+    },
+  ],
+};
+const config = {
+  enabled: true,
+  defaultMode: "stacked",
+  autoTriggerTokens: 1,
+  cacheMinutes: 0,
+  preserveSystemPrompt: true,
+  stackedPipeline: [{ engine: "rtk" }, { engine: "caveman" }],
+} as CompressionConfig;
+
+function comparable<T extends { stats: { durationMs?: number; timestamp: number } | null }>(
+  result: T
+) {
+  if (!result.stats) return result;
+  const {
+    durationMs: _duration,
+    timestamp: _timestamp,
+    engineBreakdown,
+    ...stats
+  } = result.stats as T["stats"] & {
+    engineBreakdown?: Array<Record<string, unknown>>;
+  };
+  const stableBreakdown = engineBreakdown?.map(({ durationMs: _stepDuration, ...step }) => step);
+  return {
+    ...result,
+    stats: {
+      ...stats,
+      ...(stableBreakdown ? { engineBreakdown: stableBreakdown } : {}),
+    },
+  };
+}
+
+after(() => closeCompressionWorkerPoolForTests());
+
+describe("compression worker eligibility", () => {
+  it("accepts only standard, rtk, and approved rtk+caveman stacks", () => {
+    assert.equal(isCompressionWorkerEligible(body, "standard", { config }), true);
+    assert.equal(isCompressionWorkerEligible(body, "rtk", { config }), true);
+    assert.equal(isCompressionWorkerEligible(body, "stacked", { config }), true);
+    for (const mode of ["off", "lite", "aggressive", "ultra", "omniglyph"] as const) {
+      assert.equal(isCompressionWorkerEligible(body, mode, { config }), false);
+    }
+    for (const engine of ["llmlingua", "omniglyph", "ccr", "session-dedup", "ultra"]) {
+      assert.equal(
+        isCompressionWorkerEligible(body, "stacked", {
+          config: { ...config, stackedPipeline: [{ engine }] } as CompressionConfig,
+        }),
+        false
+      );
+    }
+  });
+
+  it("rejects functions, symbols, classes, special objects, cycles, and non-finite numbers", () => {
+    for (const value of [
+      () => undefined,
+      Symbol("x"),
+      new Date(),
+      new Map(),
+      new Set(),
+      /x/,
+      NaN,
+      Infinity,
+    ]) {
+      assert.equal(isStrictlySerializable(value), false);
+    }
+    const cyclic: Record<string, unknown> = {};
+    cyclic.self = cyclic;
+    assert.equal(isStrictlySerializable(cyclic), false);
+  });
+});
+
+describe("compression worker execution", () => {
+  it("matches the synchronous body and stats except timing fields", async () => {
+    const sync = applyCompression(body, "stacked", { config });
+    const async = await applyCompressionAsync(body, "stacked", { config });
+    assert.deepEqual(comparable(async), comparable(sync));
+  });
+
+  it("preserves Responses bodies and hard-budget results", async () => {
+    const responsesBody = {
+      model: "gpt-test",
+      input: [{ role: "user", content: [{ type: "input_text", text: "word ".repeat(600) }] }],
+    };
+    const hardBudgetConfig = { ...config, targetTokens: 100 };
+    const sync = applyCompression(responsesBody, "stacked", { config: hardBudgetConfig });
+    const async = await applyCompressionAsync(responsesBody, "stacked", {
+      config: hardBudgetConfig,
+    });
+    assert.deepEqual(comparable(async), comparable(sync));
+  });
+
+  it("relays per-engine progress from the worker", async () => {
+    const steps: string[] = [];
+    await applyCompressionAsync(body, "stacked", {
+      config,
+      onEngineStep: (step) => steps.push(step.engine),
+    });
+    assert.deepEqual(steps, ["rtk", "caveman"]);
+  });
+
+  it("fails open without inline compression when a job times out", async () => {
+    const pool = new CompressionWorkerPool({ size: 1, timeoutMs: 1, idleMs: 100 });
+    try {
+      const result = await pool.run(body, "stacked", { config });
+      assert.deepEqual(result, { body, compressed: false, stats: null });
+    } finally {
+      await pool.close();
+    }
+  });
+
+  it("keeps the parent event loop responsive while two workers overlap", async () => {
+    const largeBody = {
+      messages: Array.from({ length: 400 }, (_, index) => ({
+        role: "user",
+        content: `message ${index} ` + "basically actually simply ".repeat(400),
+      })),
+    };
+    let ticked = false;
+    const tick = new Promise<void>((resolve) =>
+      setTimeout(() => {
+        ticked = true;
+        resolve();
+      }, 0)
+    );
+    const jobs = Promise.all([
+      applyCompressionAsync(largeBody, "standard", { config }),
+      applyCompressionAsync(largeBody, "standard", { config }),
+    ]);
+    await tick;
+    assert.equal(ticked, true);
+    await jobs;
+  });
+});