diff --git a/.env.example b/.env.example index a2156902a6..850727353e 100644 --- a/.env.example +++ b/.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. diff --git a/changelog.d/features/11023-compression-worker-pool.md b/changelog.d/features/11023-compression-worker-pool.md new file mode 100644 index 0000000000..4d9c1b9d60 --- /dev/null +++ b/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. diff --git a/docs/reference/ENVIRONMENT.md b/docs/reference/ENVIRONMENT.md index e9ef0df9c2..faeb76090d 100644 --- a/docs/reference/ENVIRONMENT.md +++ b/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. | diff --git a/open-sse/services/compression/compressionWorker.ts b/open-sse/services/compression/compressionWorker.ts new file mode 100644 index 0000000000..7ba32e8a18 --- /dev/null +++ b/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); + } +}); diff --git a/open-sse/services/compression/compressionWorkerPool.ts b/open-sse/services/compression/compressionWorkerPool.ts new file mode 100644 index 0000000000..c3f0f839c1 --- /dev/null +++ b/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): CompressionResult { + return { body, compressed: false, stats: null }; +} +interface PendingJob extends CompressionWorkerJob { + originalBody: Record; + 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(); + 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, + mode: CompressionWorkerJob["mode"], + options?: CompressionWorkerOptions, + onEngineStep?: (step: StackedCompressionStep) => void + ): Promise { + return new Promise((resolve) => { + this.queue.push({ + id: this.nextId++, + body, + mode, + options, + originalBody: body, + resolve, + onEngineStep, + }); + this.dispatch(); + }); + } + async close(): Promise { + 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 { + 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, + mode: CompressionWorkerJob["mode"], + options?: CompressionWorkerOptions, + onEngineStep?: (step: StackedCompressionStep) => void +): Promise { + pool ??= new CompressionWorkerPool(); + return pool.run(body, mode, options, onEngineStep); +} +export async function closeCompressionWorkerPoolForTests(): Promise { + const active = pool; + pool = null; + await active?.close(); +} diff --git a/open-sse/services/compression/compressionWorkerProtocol.ts b/open-sse/services/compression/compressionWorkerProtocol.ts new file mode 100644 index 0000000000..3e281e820a --- /dev/null +++ b/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; + 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 { + const prototype = Object.getPrototypeOf(value); + return prototype === Object.prototype || prototype === null; +} +export function isStrictlySerializable(value: unknown, seen = new Set()): 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, + 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 } : {}) }); +} diff --git a/open-sse/services/compression/strategySelector.ts b/open-sse/services/compression/strategySelector.ts index 2c9624730a..b22fee8344 100644 --- a/open-sse/services/compression/strategySelector.ts +++ b/open-sse/services/compression/strategySelector.ts @@ -519,6 +519,28 @@ async function runCompressionAsync( cachingContext?: CachingDetectionContext; } ): Promise { + 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 diff --git a/scripts/build/colocate-standalone.mjs b/scripts/build/colocate-standalone.mjs index f8dca14a51..5fe8e08b48 100644 --- a/scripts/build/colocate-standalone.mjs +++ b/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( diff --git a/scripts/build/pack-artifact-policy.ts b/scripts/build/pack-artifact-policy.ts index f5edcf994c..5f2c7e6b47 100644 --- a/scripts/build/pack-artifact-policy.ts +++ b/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", diff --git a/scripts/build/prepare-electron-standalone.mjs b/scripts/build/prepare-electron-standalone.mjs index b04ffa2812..b929a942fb 100644 --- a/scripts/build/prepare-electron-standalone.mjs +++ b/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) { diff --git a/scripts/build/prepublish.ts b/scripts/build/prepublish.ts index d29ec32560..207f88b6bc 100644 --- a/scripts/build/prepublish.ts +++ b/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"); diff --git a/tests/unit/build/colocate-standalone-esm-scope.test.ts b/tests/unit/build/colocate-standalone-esm-scope.test.ts index 548d72f6f1..d31772d11a 100644 --- a/tests/unit/build/colocate-standalone-esm-scope.test.ts +++ b/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 }); + } +}); diff --git a/tests/unit/compression/compression-worker.test.ts b/tests/unit/compression/compression-worker.test.ts new file mode 100644 index 0000000000..0ca4cbd453 --- /dev/null +++ b/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( + result: T +) { + if (!result.stats) return result; + const { + durationMs: _duration, + timestamp: _timestamp, + engineBreakdown, + ...stats + } = result.stats as T["stats"] & { + engineBreakdown?: Array>; + }; + 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 = {}; + 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((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; + }); +});