From c06cd830229bd5433dd3ff7649df2a8af6edced6 Mon Sep 17 00:00:00 2001 From: MumuTW <42820974+MumuTW@users.noreply.github.com> Date: Sun, 26 Jul 2026 14:52:35 +0800 Subject: [PATCH] fix(sse): shallow per-target copy for the combo attempt body (#7847) (#8553) * fix(sse): shallow per-target copy for the combo attempt body (#7847) combo.ts deep-cloned the request body for every target. On a 3.05 MiB agent request that is 9.53 MiB at 3 targets, and it scales linearly: 3 targets 9.53 MiB (3.12x wire) -> ~0.001 MiB 5 targets 15.89 MiB (5.19x wire) -> ~0.000 MiB 10 targets 31.78 MiB (10.39x wire) -> ~0.001 MiB The isolation it bought only ever needed to contain TOP-LEVEL SCALAR writes. The full mutation surface on this path is two assignments: combo.ts bodyRecord.max_tokens = ... (reasoning buffer) chatCore.ts body.model = model (Background Task Redirection T41) Nothing mutates the nested payload; applyCompression and injectUniversalHandoffBody both return new objects (verified empirically -- neither touches its input, and both tolerate a frozen one). So a fresh top-level object per target gives identical isolation while sharing the expensive messages/tools arrays. Also fixes a REAL cross-target leak in handleRoundRobinCombo. It already used a shallow copy, but took it only when the reasoning buffer actually changed max_tokens -- every other attempt shared the caller's object outright. The new test reproduces it on the unmodified code: target 2 received model "mutated-by-openai/gpt-4o-mini". In production that is a Background Task Redirection on one round-robin target rewriting body.model for the next. The copy is now unconditional. The invariant is pinned by tests rather than by a comment listing mutation sites, so the clone strategy can change again without anyone re-deriving them by hand: - a target's in-place write must not leak into the next (priority / fill-first / round-robin; the stub reproduces chatCore's body.model write) - the caller's body is never mutated - the per-target copy stays shallow (targets share one messages array) - freeze probe: combo's own body handling performs no in-place writes * test(sse): register the combo attempt-body isolation test in stryker tap.testFiles The new test resets the circuit breaker in beforeEach, so it counts as a covering test for src/shared/utils/circuitBreaker.ts. Without registering it, check:mutation-test-coverage --strict reported a 4th drift entry that was not there on the pristine tip -- new drift introduced by this PR. Registered in sorted position; the gate is back to the 3 pre-existing entries (accountFallback.ts, error.ts, comboPredicates.ts) that #8538 addresses. --- .../fixes/7847-combo-attempt-body-cow.md | 1 + open-sse/services/combo.ts | 27 +- stryker.conf.json | 11 +- .../combo-attempt-body-isolation-7847.test.ts | 243 ++++++++++++++++++ 4 files changed, 261 insertions(+), 21 deletions(-) create mode 100644 changelog.d/fixes/7847-combo-attempt-body-cow.md create mode 100644 tests/unit/combo-attempt-body-isolation-7847.test.ts diff --git a/changelog.d/fixes/7847-combo-attempt-body-cow.md b/changelog.d/fixes/7847-combo-attempt-body-cow.md new file mode 100644 index 0000000000..afa3539cd0 --- /dev/null +++ b/changelog.d/fixes/7847-combo-attempt-body-cow.md @@ -0,0 +1 @@ +- fix(sse): copy the combo attempt body shallowly instead of deep-cloning it per target (#7847) — the deep clone cost 9.53 MiB at 3 targets and scaled linearly with the target count (31.78 MiB at 10) on a 3.05 MiB agent request, while the isolation it provided only ever needed to contain top-level scalar writes. Also fixes a real cross-target leak in round-robin, which copied the body only when the reasoning buffer changed `max_tokens` and otherwise shared the caller's object, so a Background Task Redirection on one target rewrote `body.model` for the next diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 7cc1712681..23ad08cb34 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -1718,12 +1718,9 @@ export async function handleComboChat({ // QA P0 diagnostics: capture the attempt order (provider/model ids only). comboAttemptOrder.push({ provider: provider ?? "unknown", model: modelStr }); - // Deep clone the body to ensure context preservation and prevent mutations - // from affecting other targets in the combo. structuredClone avoids the - // full intermediate JSON string that JSON.parse(JSON.stringify(...)) builds - // (a second multi-hundred-KB allocation per target on large agent payloads), - // halving the per-target transient heap on the hot path (#5152). - let attemptBody = structuredClone(body); + // Copy-on-write, not a deep clone (#7847 — 9.53 MiB at 3 targets). Writes here are + // top-level scalars. Invariant: tests/unit/combo-attempt-body-isolation-7847.test.ts. + let attemptBody = { ...(body as Record) } as typeof body; // Proactive Context Compression for fallbacks (Zero-Latency optimization) if ( @@ -2838,7 +2835,11 @@ async function handleRoundRobinCombo({ // BEFORE availability is known; if every compat-kept target then turns out to be // runtime-unavailable, we must reconsider these before returning 503, instead of // permanently dropping a compat-rejected-but-healthy provider. - const compatRejectedTargets = computeCompatRejectedTargets(evalRankedTargets, filteredTargets, body); + const compatRejectedTargets = computeCompatRejectedTargets( + evalRankedTargets, + filteredTargets, + body + ); let modelCount = filteredTargets.length; if (modelCount === 0) { return comboModelNotFoundResponse("Round-robin combo has no executable targets"); @@ -3096,9 +3097,11 @@ async function handleRoundRobinCombo({ // Issue #3587: Reasoning models can spend the whole output budget on // reasoning. Apply any safe buffer to a per-attempt copy so round-robin // retries never compound across models. - let attemptBody = body; + // #7847: UNCONDITIONAL — copying only when the buffer changed max_tokens left every + // other attempt sharing the caller's object, leaking chatCore's `body.model` forward. + let attemptBody = { ...(body as Record) } as typeof body; { - const bodyRecord = body as Record; + const bodyRecord = attemptBody as Record; const currentMaxTokens = toPositiveInteger(bodyRecord.max_tokens); const bufferedMaxTokens = resolveReasoningBufferedMaxTokens( modelStr, @@ -3110,10 +3113,8 @@ async function handleRoundRobinCombo({ bufferedMaxTokens !== null && bufferedMaxTokens !== currentMaxTokens ) { - attemptBody = { - ...bodyRecord, - max_tokens: bufferedMaxTokens, - } as typeof body; + // Safe to write in place: bodyRecord is the per-attempt copy above, not the caller's. + bodyRecord.max_tokens = bufferedMaxTokens; log.info( "COMBO-RR", `Reasoning model ${modelStr}: adjusted max_tokens ${currentMaxTokens} -> ${bufferedMaxTokens}` diff --git a/stryker.conf.json b/stryker.conf.json index 342fd18fa5..1ba51f2537 100644 --- a/stryker.conf.json +++ b/stryker.conf.json @@ -39,9 +39,7 @@ "incremental": true, "incrementalFile": "reports/mutation/stryker-incremental.json", "testRunner": "tap", - "plugins": [ - "@stryker-mutator/tap-runner" - ], + "plugins": ["@stryker-mutator/tap-runner"], "tap": { "testFiles": [ "tests/unit/7993-noauth-proxy-routing.test.ts", @@ -136,6 +134,7 @@ "tests/unit/collect-metrics-module-coverage.test.ts", "tests/unit/combo-499-abort.test.ts", "tests/unit/combo-account-allowlist-3266.test.ts", + "tests/unit/combo-attempt-body-isolation-7847.test.ts", "tests/unit/combo-auto-candidate-expansion.test.ts", "tests/unit/combo-breaker-429.test.ts", "tests/unit/combo-cache-invalidation.test.ts", @@ -406,11 +405,7 @@ ".worktrees", ".stryker-tmp" ], - "reporters": [ - "progress", - "html", - "json" - ], + "reporters": ["progress", "html", "json"], "htmlReporter": { "fileName": "reports/mutation/mutation.html" }, diff --git a/tests/unit/combo-attempt-body-isolation-7847.test.ts b/tests/unit/combo-attempt-body-isolation-7847.test.ts new file mode 100644 index 0000000000..6238baf8ff --- /dev/null +++ b/tests/unit/combo-attempt-body-isolation-7847.test.ts @@ -0,0 +1,243 @@ +// Target isolation for the combo attempt body (#7847). +// +// combo.ts deep-clones the request body per target (`attemptBody = structuredClone(body)`) so one +// target's mutations cannot reach the next. On a 3.05 MiB agent request that costs 9.53 MiB at +// only 3 targets — 3x the wire size — and it scales with the target count. +// +// A shallow per-target copy would give the same isolation for ~nothing, because every known +// in-place mutation on this path is a TOP-LEVEL SCALAR: +// combo.ts bodyRecord.max_tokens = ... (reasoning buffer) +// chatCore.ts body.model = ... (Background Task Redirection T41) +// +// These tests lock the isolation invariant itself, so the clone strategy can be changed +// underneath them without anyone having to trust a grep for mutation sites. +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-combo-cow-")); +const ORIGINAL_DATA_DIR = process.env.DATA_DIR; +process.env.DATA_DIR = TEST_DATA_DIR; + +const { handleComboChat } = await import("../../open-sse/services/combo.ts"); +const core = await import("../../src/lib/db/core.ts"); +const { resetAllComboMetrics } = await import("../../open-sse/services/comboMetrics.ts"); +const { resetAllCircuitBreakers } = await import("../../src/shared/utils/circuitBreaker.ts"); +const { resetAll: resetAllSemaphores } = + await import("../../open-sse/services/rateLimitSemaphore.ts"); +const { _resetAllDecks } = await import("../../src/shared/utils/shuffleDeck.ts"); +const { clearSessions } = await import("../../open-sse/services/sessionManager.ts"); + +function createLog() { + const entries: unknown[] = []; + const push = (level: string) => (tag: unknown, msg: unknown) => entries.push({ level, tag, msg }); + return { + info: push("info"), + warn: push("warn"), + error: push("error"), + debug: push("debug"), + entries, + }; +} + +const okResponse = () => + new Response(JSON.stringify({ choices: [{ message: { content: "ok" } }] }), { + status: 200, + headers: { "content-type": "application/json" }, + }); +const errResponse = (status: number) => + new Response(JSON.stringify({ error: { message: `Error ${status}` } }), { + status, + headers: { "content-type": "application/json" }, + }); + +/** A body with nested structure, so a shallow copy is visibly different from a deep clone. */ +function agentBody() { + return { + model: "openai/gpt-4o-mini", + max_tokens: 100, + messages: [ + { role: "user", content: "first" }, + { role: "assistant", content: "second" }, + ], + tools: [{ type: "function", function: { name: "t", parameters: { type: "object" } } }], + }; +} + +function deepFreeze(value: T): T { + if (value && typeof value === "object") { + Object.getOwnPropertyNames(value).forEach((k) => deepFreeze((value as never)[k])); + Object.freeze(value); + } + return value; +} + +const MODELS = ["openai/gpt-4o-mini", "claude/sonnet", "gemini/flash"]; + +function comboOf(strategy: string, name: string) { + return { + name, + strategy, + models: MODELS, + config: { maxRetries: 0, retryDelayMs: 0, fallbackDelayMs: 0 }, + }; +} + +test.beforeEach(() => { + resetAllComboMetrics(); + resetAllCircuitBreakers(); + resetAllSemaphores(); + _resetAllDecks(); + clearSessions(); +}); + +test.after(() => { + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); + if (ORIGINAL_DATA_DIR === undefined) delete process.env.DATA_DIR; + else process.env.DATA_DIR = ORIGINAL_DATA_DIR; +}); + +// ── The invariant: one target's writes must never reach the next ────────────── +// The stub reproduces chatCore.ts's `body.model = model` (Background Task Redirection), +// which is the real downstream mutation this isolation exists to contain. +for (const strategy of ["priority", "fill-first", "round-robin"]) { + test(`${strategy}: a target's in-place write must not leak into the next target`, async () => { + const seen: { model: string; maxTokens: number }[] = []; + + await handleComboChat({ + body: agentBody(), + combo: comboOf(strategy, `cow-isolation-${strategy}`), + handleSingleModel: async (received: Record, modelStr: string) => { + seen.push({ + model: received.model as string, + maxTokens: received.max_tokens as number, + }); + // Simulate chatCore.ts:693 (`body.model = model`) and the reasoning buffer write. + received.model = `mutated-by-${modelStr}`; + received.max_tokens = 999; + return seen.length < MODELS.length ? errResponse(503) : okResponse(); + }, + isModelAvailable: async () => true, + log: createLog(), + settings: null, + allCombos: null, + }); + + assert.ok(seen.length >= 2, `expected at least 2 targets to run, got ${seen.length}`); + for (const [i, s] of seen.entries()) { + assert.equal( + s.model, + "openai/gpt-4o-mini", + `target ${i} received model "${s.model}" — a previous target's write leaked through` + ); + assert.equal( + s.maxTokens, + 100, + `target ${i} received max_tokens ${s.maxTokens} — a previous target's write leaked through` + ); + } + }); +} + +// ── The caller's body is an input, not scratch space ────────────────────────── +test("the caller's body object is never mutated by the combo loop", async () => { + const body = agentBody(); + const before = JSON.stringify(body); + + await handleComboChat({ + body, + combo: comboOf("priority", "cow-caller-body"), + handleSingleModel: async (received: Record) => { + received.model = "mutated"; + received.max_tokens = 1; + return okResponse(); + }, + isModelAvailable: async () => true, + log: createLog(), + settings: null, + allCombos: null, + }); + + assert.equal(JSON.stringify(body), before, "handleComboChat must treat `body` as read-only"); +}); + +// ── The copy must stay shallow — that is the whole point ───────────────────── +test("the per-target copy shares the nested payload instead of deep-cloning it", async () => { + const body = agentBody(); + const received: Record[] = []; + + await handleComboChat({ + body, + combo: comboOf("priority", "cow-shallow"), + handleSingleModel: async (b: Record) => { + received.push(b); + return received.length < 2 ? errResponse(503) : okResponse(); + }, + isModelAvailable: async () => true, + log: createLog(), + settings: null, + allCombos: null, + }); + + assert.ok(received.length >= 2, "need at least two targets to compare"); + for (const [i, b] of received.entries()) { + assert.notEqual( + b, + body, + `target ${i} received the caller's object — isolation requires a copy` + ); + } + assert.notEqual(received[0], received[1], "each target needs its own top-level object"); + + // The memory property (#7847): `messages` is the expensive part of an agent request, and + // duplicating it per target is what cost 9.53 MiB at 3 targets. Every target must therefore + // point at the SAME array. (handleComboChat rebuilds the messages container once during + // setup, before the per-target loop, so this is not necessarily the caller's own array — + // what matters is that it is not rebuilt per target.) + assert.equal( + received[0].messages, + received[1].messages, + "targets got different messages arrays — the per-target copy must stay shallow (#7847)" + ); + assert.equal(received[0].tools, body.tools, "tools must be shared, not copied"); + + // And nothing deep-clones: the message objects themselves are still the caller's. + const first = (body.messages as unknown[])[0]; + assert.equal( + (received[0].messages as unknown[])[0], + first, + "message objects were cloned — nothing on this path mutates them, so they must be shared" + ); +}); + +// ── Freeze probe: nothing on the combo-owned path may write in place ────────── +// Scope note: handleSingleModel is stubbed, so this covers combo.ts's own handling of the body +// (compression, handoff injection, the reasoning-buffer write) — NOT chatCore or the executors. +// The isolation tests above are what cover a mutating downstream. +test("freeze probe: combo's own body handling performs no in-place writes", async () => { + const frozen = deepFreeze(agentBody()); + let threw: unknown = null; + + try { + await handleComboChat({ + body: frozen, + combo: comboOf("priority", "cow-freeze-probe"), + handleSingleModel: async () => okResponse(), + isModelAvailable: async () => true, + log: createLog(), + settings: null, + allCombos: null, + }); + } catch (err) { + threw = err; + } + + assert.equal( + threw, + null, + `combo wrote to a frozen body: ${threw instanceof Error ? threw.message : String(threw)}` + ); +});