Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/fixes/7847-combo-attempt-body-cow.md
Original file line number Diff line number Diff line change
@@ -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
27 changes: 14 additions & 13 deletions open-sse/services/combo.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1804,12 +1804,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<string, unknown>) } as typeof body;

// Proactive Context Compression for fallbacks (Zero-Latency optimization)
if (
Expand Down Expand Up @@ -2924,7 +2921,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");
Expand Down Expand Up @@ -3182,9 +3183,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<string, unknown>) } as typeof body;
{
const bodyRecord = body as Record<string, unknown>;
const bodyRecord = attemptBody as Record<string, unknown>;
const currentMaxTokens = toPositiveInteger(bodyRecord.max_tokens);
const bufferedMaxTokens = resolveReasoningBufferedMaxTokens(
modelStr,
Expand All @@ -3196,10 +3199,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}`
Expand Down
11 changes: 3 additions & 8 deletions stryker.conf.json
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -406,11 +405,7 @@
".worktrees",
".stryker-tmp"
],
"reporters": [
"progress",
"html",
"json"
],
"reporters": ["progress", "html", "json"],
"htmlReporter": {
"fileName": "reports/mutation/mutation.html"
},
Expand Down
243 changes: 243 additions & 0 deletions tests/unit/combo-attempt-body-isolation-7847.test.ts
Original file line number Diff line number Diff line change
@@ -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<T>(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<string, unknown>, 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<string, unknown>) => {
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<string, unknown>[] = [];

await handleComboChat({
body,
combo: comboOf("priority", "cow-shallow"),
handleSingleModel: async (b: Record<string, unknown>) => {
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)}`
);
});
Loading