From 0dcf809159d4d7960d91beca1a24e6944d8b811a Mon Sep 17 00:00:00 2001 From: Gunther Schulz Date: Fri, 14 Aug 2026 14:05:13 +0200 Subject: [PATCH 1/2] Row 31: coalesce the duplicate sidecar send into one upstream call, gated off CC issues one sidecar request twice, 6-25 ms apart on 47/47 measured pairs, with distinct upstream request-ids and two completed usage-log records -- both answered, both charged. Dropping the second is unavailable, since two client requests are in flight and each is owed a response, so the only safe shape is one upstream call serving both callers. Four conditions, all required, and the mid-session duplicate class -- where the second send is a legitimate retry -- fails on nMsg alone, which is the discriminator the row asked for. Fan-out sits at the RESPONSE WRITER, not at the upstream reader: the extension pass and the telemetry record run once, so both callers receive byte-identical post-pipeline output. Tee-ing the raw upstream would hand the follower unmutated bytes while the leader got the pipeline's, and fidelity outranks cache here. TWO ARMS WERE NOT DISCRIMINATING AND THE MUTATION PROOF IS WHAT SHOWED IT. Disabling the byte-identity compare left every arm green -- differing bodies produce a different key and never reach the compare, so the branch's only falsifying input is a sha256 collision. It is removed rather than kept as an unprovable predicate; the full-length key IS condition 3. Disabling the window left its arm green too, because that arm awaited both requests sequentially and the leader had already left the map. Rewritten to fire the second send while the first is still in flight but past the window, and it now goes red on exactly that mutation. Gated OFF (CACHE_FIX_COALESCE_SIDECAR). Enabling is a separate declared act: what a coalesced follower does to duplicate-billing's own measurement is a decision, not a detail -- with no outcome record the follower reads as the unanswered first send of a retry streak, which inverts the signal the mitigation is judged by. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_011q6Zo7yKHCGokK4fT8LTPF --- proxy/server.mjs | 197 +++++++++++++++++++-- test/duplicate-coalesce.test.mjs | 293 +++++++++++++++++++++++++++++++ 2 files changed, 475 insertions(+), 15 deletions(-) create mode 100644 test/duplicate-coalesce.test.mjs diff --git a/proxy/server.mjs b/proxy/server.mjs index 21f7e76a..93614866 100644 --- a/proxy/server.mjs +++ b/proxy/server.mjs @@ -52,6 +52,124 @@ function debugLog(...args) { try { appendFileSync(path, line); } catch {} } +// ── Duplicate sidecar coalescing — threat-matrix row 31 ────────────────── +// +// CC issues one sidecar request TWICE, 6-25 ms apart on 47/47 measured +// pairs, with distinct upstream request-ids and two completed usage-log +// records: both sends are answered and both are charged (48,203 input-side +// tokens corpus-wide). Dropping the second is unavailable — two client +// requests are in flight and each is owed a response — so the only safe +// shape is ONE upstream call serving both callers. +// +// Four conditions, all required. The mid-session duplicate class, where a +// second send is a legitimate retry and suppressing it would leave a real +// request unanswered, fails on `nMsg` alone — that is the discriminator +// row 31 asked for: +// +// 1. exactly one message 3. byte-identical FORWARDED bodies +// 2. no tools[] 4. < 50 ms, first still in flight +// +// Why substituting one answer for the other is fidelity-safe, which is the +// objection that kept this parked: a request carrying no conversation +// history and no tools produces output that can enter no cached prefix, and +// CC issued the second send before it could have observed the first, so it +// already treats the two as interchangeable. +// +// Condition 3 is checked on the bytes we ACTUALLY send, after every +// extension has run — identical forwarded bodies is what makes the two +// upstream calls the same call — and it IS the map key: a full-length +// sha256 of those bytes, so a hit already means byte-identical. +const COALESCE_WINDOW_MS = 50; + +/** key (sha256 of forwarded bytes) -> in-flight leader. */ +const inFlightSidecars = new Map(); + +// Conditions 1 and 2 — the half that is a property of ONE request. 3 and 4 +// belong to a PAIR and are checked at the map hit. Exported for the bites: +// a predicate whose arms are only reachable through a live socket cannot be +// shown red on the case it was built for. +export function coalesceCandidate(parsed) { + if (!parsed || typeof parsed !== "object") return false; + if (!Array.isArray(parsed.messages) || parsed.messages.length !== 1) return false; + if (Array.isArray(parsed.tools) && parsed.tools.length > 0) return false; + return true; +} + +// One writable face over N client responses. `streamResponse` touches only +// write / once("drain") / end, and the non-streaming branch only +// writeHead / end, so this is the whole surface either needs. +// +// A follower may attach mid-stream, so the leader keeps every chunk it has +// written and replays it on attach — the alternative (buffering the whole +// response before writing any of it) would convert the leader's stream into +// a single delivery, which is a behaviour change to the streaming path for +// every coalesced request. +export function createFanOut(leaderRes) { + const sinks = [leaderRes]; + const replay = []; + let head = null; + let ended = false; + + const live = () => sinks.filter((r) => !r.writableEnded && !r.destroyed); + + return { + get sinkCount() { return live().length; }, + get writableEnded() { return live().length === 0; }, + writeHead(status, headers) { + head = { status, headers }; + for (const r of live()) r.writeHead(status, headers); + }, + // Returns false when the follower arrived too late to join — the caller + // has already been served in full and owes it nothing further. + attach(res) { + if (head) { + res.writeHead(head.status, head.headers); + for (const chunk of replay) res.write(chunk); + } + if (ended) { + res.end(); + return false; + } + sinks.push(res); + return true; + }, + write(chunk) { + replay.push(chunk); + let ok = true; + for (const r of live()) { + if (!r.write(chunk)) ok = false; + } + return ok; + }, + once(event, cb) { + if (event !== "drain") return; + const pending = live().filter((r) => r.writableNeedDrain); + if (pending.length === 0) { + setImmediate(cb); + return; + } + // The slowest attached client governs, and a client that goes away + // mid-drain must not hold the others: `close` counts as drained, or a + // follower hanging up would stall the leader's stream forever. + let left = pending.length; + const fire = () => { if (--left === 0) cb(); }; + for (const r of pending) { + r.once("drain", fire); + r.once("close", fire); + } + }, + end(data) { + if (data !== undefined) replay.push(data); + ended = true; + for (const r of live()) r.end(data); + }, + destroy(err) { + ended = true; + for (const r of live()) r.destroy(err); + }, + }; +} + function collectBody(req) { return new Promise((resolve, reject) => { const chunks = []; @@ -146,8 +264,13 @@ async function handleMessages(clientReq, clientRes) { // Bootstrap (handleBootstrap) doesn't install this because its response is // a single non-SSE JSON payload — aborting on clientReq close prematurely // would race the response write on fast-failure paths (e.g. ECONNREFUSED). + // `sink` is this request's own response until a coalescing leader adopts + // it; the close handler asks the SINK whether anyone is still listening, + // so a leader whose own client hangs up keeps serving its followers + // instead of aborting the upstream call they are waiting on. + let sink = clientRes; clientReq.on("close", () => { - if (!clientRes.writableEnded) abortController.abort(); + if (!sink.writableEnded) abortController.abort(); }); const pre = await preForward(clientReq, clientRes, abortController, extSnapshot, "messages"); @@ -160,6 +283,44 @@ async function handleMessages(clientReq, clientRes) { } const { parsed, forwardBody, headers, meta } = pre; + // Row 31. Gated OFF by default: the mechanism ships with its bites, and + // enabling it is a separate, declared act (ship-proxy-change step 4b). + const coalesceKey = process.env.CACHE_FIX_COALESCE_SIDECAR === "1" && coalesceCandidate(parsed) + ? createHash("sha256").update(forwardBody).digest("hex") + : null; + + if (coalesceKey) { + const leader = inFlightSidecars.get(coalesceKey); + // Condition 3 IS the key: a full-length sha256 of the forwarded bytes, + // so a map hit already means byte-identical. A second `Buffer.equals` + // beside it was written here first and removed after the mutation proof + // — disabling it left every arm green, because differing bodies produce + // a different key and never reach the compare. Its only falsifying input + // is a sha256 collision, which makes it an unprovable predicate wearing + // a check's clothes. Condition 4, the window, is what remains here. + if (leader && Date.now() - leader.at < COALESCE_WINDOW_MS) { + debugLog("[PROXY] coalescing duplicate sidecar into in-flight request", + "key:", coalesceKey.slice(0, 16), "ageMs:", Date.now() - leader.at); + if (leader.fanOut.attach(clientRes)) await leader.done; + return; + } + let settle; + const entry = { + at: Date.now(), + fanOut: createFanOut(clientRes), + done: new Promise((r) => { settle = r; }), + }; + entry.settle = settle; + inFlightSidecars.set(coalesceKey, entry); + sink = entry.fanOut; + // The map entry outlives neither the request nor an early throw: every + // exit below runs through this. + clientRes.on("close", () => { + if (inFlightSidecars.get(coalesceKey) === entry) inFlightSidecars.delete(coalesceKey); + entry.settle(); + }); + } + const requestedModel = parsed?.model || null; let upstreamRes, responseHeaders, statusCode, upstreamConnectionId; @@ -177,8 +338,8 @@ async function handleMessages(clientReq, clientRes) { } catch (err) { debugLog("[PROXY] forwardRequest error:", err.message); if (abortController.signal.aborted) return; - clientRes.writeHead(502, { "content-type": "application/json" }); - clientRes.end(JSON.stringify({ error: "upstream_error", message: err.message })); + sink.writeHead(502, { "content-type": "application/json" }); + sink.end(JSON.stringify({ error: "upstream_error", message: err.message })); return; } @@ -214,35 +375,41 @@ async function handleMessages(clientReq, clientRes) { if (responseBody) { const resCtx = { status: statusCode, headers: responseHeaders, body: responseBody, meta }; await runOnResponse(resCtx, extSnapshot); - clientRes.writeHead(statusCode, resCtx.headers); - clientRes.end(JSON.stringify(resCtx.body)); + sink.writeHead(statusCode, resCtx.headers); + sink.end(JSON.stringify(resCtx.body)); } else { - clientRes.writeHead(statusCode, responseHeaders); - clientRes.end(rawResponse); + sink.writeHead(statusCode, responseHeaders); + sink.end(rawResponse); } } else { - clientRes.writeHead(statusCode, responseHeaders); - clientRes.end(rawResponse); + sink.writeHead(statusCode, responseHeaders); + sink.end(rawResponse); } return; } - clientRes.writeHead(statusCode, responseHeaders); + sink.writeHead(statusCode, responseHeaders); const telemetry = createTelemetryRecord(); telemetry.requestedModel = requestedModel; upstreamRes.on("error", (err) => { - if (!clientRes.writableEnded) { - clientRes.destroy(err); + if (!sink.writableEnded) { + sink.destroy(err); } }); try { - await streamResponse(upstreamRes, clientRes, telemetry, extSnapshot, meta, responseHeaders); + // Fan-out sits at the RESPONSE WRITER, never at the upstream reader: the + // extension pass and the telemetry record run exactly once, so both + // callers receive byte-identical post-pipeline output. Tee-ing the raw + // upstream instead would hand the follower unmutated bytes while the + // leader got the pipeline's — a fidelity split, and fidelity outranks + // cache here. + await streamResponse(upstreamRes, sink, telemetry, extSnapshot, meta, responseHeaders); } catch (err) { - if (!clientRes.writableEnded) { - clientRes.destroy(err); + if (!sink.writableEnded) { + sink.destroy(err); } } } diff --git a/test/duplicate-coalesce.test.mjs b/test/duplicate-coalesce.test.mjs new file mode 100644 index 00000000..ba48b2a1 --- /dev/null +++ b/test/duplicate-coalesce.test.mjs @@ -0,0 +1,293 @@ +// Threat-matrix row 31: CC issues one sidecar request TWICE, 6-25 ms apart, +// with distinct upstream request-ids and two completed usage-log records — +// both answered, both charged. The mitigation coalesces the pair into ONE +// upstream call serving both callers. +// +// These bites exercise the predicate AT THE WIRE, through a real proxy +// instance against a real (local) upstream that counts what it received, +// because the defect is a count of upstream calls and nothing below that +// altitude can observe it. The arms MUST DIFFER: an assertion that only +// showed "one call" for the coalescing case would pass equally against a +// build that coalesced everything, which is the over-reach this predicate +// exists to prevent — so the mid-session arm asserting TWO calls is the +// discriminating half, not decoration. + +import { tmpDir } from "../tools/tmpdir.mjs"; +import { describe, it, before, after } from "node:test"; +import assert from "node:assert/strict"; +import http from "node:http"; +import { rm } from "node:fs/promises"; +import { startProxy, coalesceCandidate, createFanOut } from "../proxy/server.mjs"; + +function clientRequest(port, body) { + return new Promise((resolve, reject) => { + const data = JSON.stringify(body); + const req = http.request( + { + hostname: "127.0.0.1", + port, + path: "/v1/messages", + method: "POST", + headers: { "content-type": "application/json" }, + }, + (res) => { + const chunks = []; + res.on("data", (c) => chunks.push(c)); + res.on("end", () => resolve({ status: res.statusCode, body: Buffer.concat(chunks).toString() })); + }, + ); + req.on("error", reject); + req.end(data); + }); +} + +// Holds the response open long enough that a duplicate arriving inside the +// 50 ms window finds the first still IN FLIGHT — condition 4. Without the +// hold the first call would complete before the second arrived and the +// coalescing arm would pass for the wrong reason. +function slowSseUpstream(counter, holdMs = 120) { + return http.createServer((req, res) => { + const chunks = []; + req.on("data", (c) => chunks.push(c)); + req.on("end", () => { + counter.calls += 1; + counter.bodies.push(Buffer.concat(chunks).toString()); + res.writeHead(200, { "content-type": "text/event-stream" }); + res.write('data: {"type":"message_start","message":{"model":"claude-haiku-4-5","usage":{}}}\n\n'); + setTimeout(() => { + res.write('data: {"type":"message_stop"}\n\n'); + res.write("data: [DONE]\n\n"); + res.end(); + }, holdMs); + }); + }); +} + +const SIDECAR = { + model: "claude-haiku-4-5", + max_tokens: 32000, + stream: true, + messages: [{ role: "user", content: [{ type: "text", text: "x".repeat(337) }] }], +}; + +const MID_SESSION = { + ...SIDECAR, + messages: [ + { role: "user", content: [{ type: "text", text: "first" }] }, + { role: "assistant", content: [{ type: "text", text: "reply" }] }, + { role: "user", content: [{ type: "text", text: "second" }] }, + ], +}; + +describe("row 31 — the structural half of the predicate (conditions 1 and 2)", () => { + it("accepts a single-message request carrying no tools", () => { + assert.equal(coalesceCandidate(SIDECAR), true); + }); + + it("REJECTS a mid-session request — nMsg alone is the discriminator the row asked for", () => { + assert.equal(coalesceCandidate(MID_SESSION), false); + }); + + it("REJECTS a single-message request that carries tools", () => { + assert.equal(coalesceCandidate({ ...SIDECAR, tools: [{ name: "Bash" }] }), false); + }); + + it("treats an EMPTY tools array as no tools — the measured request carried 0", () => { + assert.equal(coalesceCandidate({ ...SIDECAR, tools: [] }), true); + }); + + it("rejects a body with no messages array at all", () => { + assert.equal(coalesceCandidate({ model: "x" }), false); + assert.equal(coalesceCandidate(null), false); + }); +}); + +describe("row 31 — the fan-out writable serves every attached caller", () => { + function fakeRes() { + return { + writableEnded: false, destroyed: false, writableNeedDrain: false, + head: null, chunks: [], ended: false, + writeHead(status, headers) { this.head = { status, headers }; }, + write(c) { this.chunks.push(String(c)); return true; }, + end(c) { if (c !== undefined) this.chunks.push(String(c)); this.ended = true; this.writableEnded = true; }, + destroy() { this.destroyed = true; }, + once() {}, + }; + } + + it("a follower attaching MID-STREAM receives the chunks already written", () => { + const leader = fakeRes(); + const fan = createFanOut(leader); + fan.writeHead(200, { "content-type": "text/event-stream" }); + fan.write("data: one\n\n"); + + const follower = fakeRes(); + assert.equal(fan.attach(follower), true); + + fan.write("data: two\n\n"); + fan.end(); + + assert.deepEqual(leader.chunks, ["data: one\n\n", "data: two\n\n"]); + assert.deepEqual(follower.chunks, ["data: one\n\n", "data: two\n\n"], + "the follower must receive the whole response, not only what came after it attached"); + assert.deepEqual(follower.head, { status: 200, headers: { "content-type": "text/event-stream" } }); + assert.equal(follower.ended, true); + }); + + it("a follower attaching AFTER the response ended is served in full and reports not-joined", () => { + const leader = fakeRes(); + const fan = createFanOut(leader); + fan.writeHead(200, {}); + fan.write("data: one\n\n"); + fan.end(); + + const late = fakeRes(); + assert.equal(fan.attach(late), false, "a late follower must not be added to the live set"); + assert.deepEqual(late.chunks, ["data: one\n\n"]); + assert.equal(late.ended, true, "it is still owed a complete response"); + }); + + it("a leader whose own client hung up keeps writing to its followers", () => { + const leader = fakeRes(); + const fan = createFanOut(leader); + fan.writeHead(200, {}); + const follower = fakeRes(); + fan.attach(follower); + + leader.destroyed = true; // the leader's client goes away mid-stream + fan.write("data: after\n\n"); + + assert.equal(fan.writableEnded, false, "someone is still listening"); + assert.deepEqual(follower.chunks, ["data: after\n\n"]); + assert.deepEqual(leader.chunks, [], "nothing is written to a dead socket"); + }); +}); + +describe("row 31 at the wire — the upstream call COUNT is the defect", () => { + let handle, upstream, counter, extDir; + + before(async () => { + extDir = await tmpDir("coalesce-ext-"); + counter = { calls: 0, bodies: [] }; + upstream = slowSseUpstream(counter); + await new Promise((r) => upstream.listen(0, "127.0.0.1", r)); + process.env.CACHE_FIX_PROXY_UPSTREAM = `http://127.0.0.1:${upstream.address().port}`; + process.env.CACHE_FIX_COALESCE_SIDECAR = "1"; + handle = await startProxy({ port: 0, watch: false, extensionsDir: extDir }); + }); + + after(async () => { + await handle.close(); + await new Promise((r) => upstream.close(r)); + delete process.env.CACHE_FIX_PROXY_UPSTREAM; + delete process.env.CACHE_FIX_COALESCE_SIDECAR; + await rm(extDir, { recursive: true, force: true }); + }); + + it("all four conditions: ONE upstream call, BOTH callers answered", async () => { + counter.calls = 0; + const a = clientRequest(handle.port, SIDECAR); + await new Promise((r) => setTimeout(r, 15)); // inside the 50 ms window + const b = clientRequest(handle.port, SIDECAR); + const [ra, rb] = await Promise.all([a, b]); + + assert.equal(counter.calls, 1, "the duplicate must not reach upstream"); + assert.equal(ra.status, 200); + assert.equal(rb.status, 200); + assert.equal(ra.body, rb.body, "both callers receive byte-identical output"); + assert.ok(ra.body.includes("message_stop"), "and it is the COMPLETE response, not a truncated replay"); + }); + + it("mid-session pair (nMsg > 1): TWO upstream calls, unchanged", async () => { + counter.calls = 0; + const a = clientRequest(handle.port, MID_SESSION); + await new Promise((r) => setTimeout(r, 15)); + const b = clientRequest(handle.port, MID_SESSION); + await Promise.all([a, b]); + + assert.equal(counter.calls, 2, + "a mid-session duplicate is a legitimate retry — suppressing it would leave a real request unanswered"); + }); + + it("three of four conditions (tools present): TWO upstream calls", async () => { + counter.calls = 0; + const withTools = { ...SIDECAR, tools: [{ name: "Bash", input_schema: {} }] }; + const a = clientRequest(handle.port, withTools); + await new Promise((r) => setTimeout(r, 15)); + const b = clientRequest(handle.port, withTools); + await Promise.all([a, b]); + + assert.equal(counter.calls, 2, "failing any one condition must not coalesce"); + }); + + it("three of four conditions (still in flight, but PAST the 50 ms window): TWO upstream calls", async () => { + // The second send must arrive while the first is STILL IN FLIGHT (the + // upstream holds 120 ms) but outside the window, or this arm proves + // nothing about condition 4. The first version awaited both requests + // sequentially, so the leader had already left the map and the window + // check was never reached — disabling the window left it green, which + // is how the gap was found. + counter.calls = 0; + const a = clientRequest(handle.port, SIDECAR); + await new Promise((r) => setTimeout(r, 80)); + const b = clientRequest(handle.port, SIDECAR); + await Promise.all([a, b]); + + assert.equal(counter.calls, 2, "past the window the pair is not a duplicate send"); + }); + + it("a sequential repeat (first already completed) is not coalesced", async () => { + counter.calls = 0; + await clientRequest(handle.port, SIDECAR); + await clientRequest(handle.port, SIDECAR); + + assert.equal(counter.calls, 2, "the leader must not outlive its own request"); + }); + + it("differing bodies inside the window: TWO upstream calls", async () => { + counter.calls = 0; + const a = clientRequest(handle.port, SIDECAR); + await new Promise((r) => setTimeout(r, 15)); + const b = clientRequest(handle.port, { ...SIDECAR, max_tokens: 16000 }); + await Promise.all([a, b]); + + // What this establishes, stated precisely because the first version of + // this arm claimed more: differing forwarded bytes produce a different + // KEY, so the pair never meets. It does not exercise a separate + // byte-compare, and the mutation proof is what showed that — disabling + // one left this arm green. + assert.equal(counter.calls, 2, "differing forwarded bytes never share a coalescing key"); + }); +}); + +describe("row 31 — the gate is OFF by default", () => { + let handle, upstream, counter, extDir; + + before(async () => { + extDir = await tmpDir("coalesce-off-ext-"); + counter = { calls: 0, bodies: [] }; + upstream = slowSseUpstream(counter); + await new Promise((r) => upstream.listen(0, "127.0.0.1", r)); + process.env.CACHE_FIX_PROXY_UPSTREAM = `http://127.0.0.1:${upstream.address().port}`; + delete process.env.CACHE_FIX_COALESCE_SIDECAR; + handle = await startProxy({ port: 0, watch: false, extensionsDir: extDir }); + }); + + after(async () => { + await handle.close(); + await new Promise((r) => upstream.close(r)); + delete process.env.CACHE_FIX_PROXY_UPSTREAM; + await rm(extDir, { recursive: true, force: true }); + }); + + it("without the gate the duplicate still reaches upstream — the pre-fix behaviour, pinned", async () => { + counter.calls = 0; + const a = clientRequest(handle.port, SIDECAR); + await new Promise((r) => setTimeout(r, 15)); + const b = clientRequest(handle.port, SIDECAR); + await Promise.all([a, b]); + + assert.equal(counter.calls, 2, + "this is the RED baseline: the same input under the shipped-but-disabled build double-bills"); + }); +}); From f4ca4d188e457fc835df5e2a6b4dd5dfcf1951bd Mon Sep 17 00:00:00 2001 From: Gunther Schulz Date: Fri, 14 Aug 2026 20:00:52 +0200 Subject: [PATCH 2/2] tools: the run-root temp helper the coalescing test needs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One self-contained helper, carried because the test in the previous commit imports it: every producer gets one run root per process, removed on exit, on throw, and on SIGINT/SIGTERM/SIGHUP. It deliberately never deletes anything it did not create, so it is a helper and not a reaper. Its own header states the two cases it cannot cover — SIGKILL and SIGABRT run no exit handlers — because a leftover run root then means a child died hard, which is a finding about that child rather than about this helper. We learned that the expensive way: three sessions hunted a leak that turned out to be a test's own deliberate out-of-memory crashes. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01RMiYvNxKq6G9gfJMzArm4q --- tools/tmpdir.mjs | 201 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 201 insertions(+) create mode 100644 tools/tmpdir.mjs diff --git a/tools/tmpdir.mjs b/tools/tmpdir.mjs new file mode 100644 index 00000000..12f55032 --- /dev/null +++ b/tools/tmpdir.mjs @@ -0,0 +1,201 @@ +// The one place this repo creates temporary directories. +// +// DEFINITION (what "correct" means here, so the checks below are not written +// from the implementation): a process that uses this helper leaves NOTHING +// under the OS temp root once it has exited. Not "usually", and not "when the +// happy path is taken" — the guarantee holds for a run that throws, a run that +// calls process.exit(), and a run that is terminated by SIGINT/SIGTERM/SIGHUP. +// The cases it cannot hold for are the ones that run no code at all: SIGKILL, +// and a V8 heap-limit failure, which calls abort() (SIGABRT, exit 134) without +// running a single exit handler. That residue is what gate-live's leftover +// signal exists to report, and the second case is not hypothetical — it is +// exactly what the `tmpLeftovers` guard was reporting for two days while this +// header named only SIGKILL and sent every reader looking for a kill that never +// happened (measured 2026-08-14: two aborted children per full suite run, +// manufactured by test/gate-live-rowpins.test.mjs's deliberate 8 MB-cap crash; +// that file now gives them a private TMPDIR, so the residue lands inside the +// suite's own scratch). A leftover root is therefore evidence that some child +// DIED HARD — which is a finding about that child, not about this module. +// +// WHY THIS EXISTS. Measured 2026-08-08: /tmp here is a 31 GB tmpfs and it +// reached 100% with 31,108 top-level directories — 7,024 `fixture-verd*`, +// ~8,000 `bt-*`, plus `census-*`, `harvest-*`, `verdict-*`, `ledger-*`, +// `mitigation-output-*` and more. Every one was an `mkdtemp` whose creator +// never removed it. The ENOSPC then broke unrelated tooling machine-wide while +// the test suite stayed GREEN — the silent-failure class — and it produced five +// consecutive runs of ONE commit returning 0, 3, 95, 525 and 528 failures, +// which read as a broken build and was first misdiagnosed as concurrency +// (docs/dev-loop.md, "A failure count that swings by hundreds"). +// +// WHY A PER-RUN PARENT rather than per-call cleanup. Per-call cleanup is what +// the leaking sites already tried: several tools do `rm(scratch)` on the happy +// path and skip it on every throw, and `fixture-verdict-identity.mjs` had a +// `finally` that restored env vars and forgot the directory. One parent per +// process, removed once at exit, makes the guarantee independent of how many +// call sites there are and of which of them remembered — a call site can only +// opt IN to the leak now, by not using this module. `test/no-raw-mkdtemp.test.mjs` +// is the writer-side guard that keeps that from happening quietly. +// +// NOT A REAPER, deliberately. This module never deletes anything it did not +// create in this process. A stale directory from someone else's run may belong +// to a run that is still going (a long replay legitimately outlives an hour), +// and a helper that swept the temp root on startup would be a destructive +// sweep racing every concurrent lane on this machine. Reporting is gate-live's +// job (`staleRunRoots` below); deleting stays a human decision. +import { mkdtempSync, rmSync, readdirSync, statSync } from "node:fs"; +import { mkdtemp } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +// Every directory this repo creates under the temp root is a child of a parent +// carrying this prefix, which is what makes the leftover scan below a closed +// question rather than a list of prefixes someone has to remember to extend. +// +// The name carries the creating PID (`cache-fix-run--XXXXXX`) so the scan +// can tell a run that DIED from one that is merely slow. Without it the age +// threshold alone would fire on gate-live's own long replay children — a check +// firing on legitimate work, which trains its reader to ignore red. +export const RUN_ROOT_PREFIX = "cache-fix-run-"; +const RUN_ROOT_RE = /^cache-fix-run-(\d+)-/; + +let runRoot = null; +let removed = false; + +// Cleanup must not throw: it runs from an `exit` handler, where the only thing +// a throw can accomplish is masking the real result the process was about to +// report. The third answer belongs to the caller, not to teardown. +function removeRunRoot() { + if (!runRoot || removed) return; + removed = true; + try { + rmSync(runRoot, { recursive: true, force: true }); + } catch { + // Nothing useful to do at exit time. + } +} + +function ensureRunRoot() { + if (runRoot) return runRoot; + runRoot = mkdtempSync(join(tmpdir(), `${RUN_ROOT_PREFIX}${process.pid}-`)); + removed = false; + + // `exit` covers the ordinary paths AND the crash paths: node runs exit + // handlers after an uncaught exception and after an unhandled rejection, so + // "the run threw" needs no separate arm. It must be synchronous — hence + // rmSync — because async work at exit time never runs. + process.on("exit", removeRunRoot); + + // Signals are the arm `exit` does NOT cover, and they are not hypothetical: + // gate-live and harvest run under systemd timers, and systemd stops a unit + // with SIGTERM. `once` removes the listener before invoking it, so the + // explicit exit below is the only termination path — we re-exit with the + // conventional 128+signal code rather than re-raising, because that is + // predictable regardless of what else has attached a listener. + const SIGNAL_EXIT = { SIGINT: 130, SIGTERM: 143, SIGHUP: 129 }; + for (const [sig, code] of Object.entries(SIGNAL_EXIT)) { + process.once(sig, () => { + removeRunRoot(); + process.exit(code); + }); + } + return runRoot; +} + +/** + * Async temp directory, inside this process's run root. + * Replaces `await mkdtemp(join(tmpdir(), prefix))`. + * @param {string} prefix e.g. "cache-fix-replay-" + * @returns {Promise} absolute path to a fresh directory + */ +export async function tmpDir(prefix) { + return mkdtemp(join(ensureRunRoot(), prefix)); +} + +/** + * Sync temp directory, inside this process's run root. + * Replaces `mkdtempSync(join(tmpdir(), prefix))`. + * @param {string} prefix e.g. "bt-" + * @returns {string} absolute path to a fresh directory + */ +export function tmpDirSync(prefix) { + return mkdtempSync(join(ensureRunRoot(), prefix)); +} + +/** + * This process's run root, or null if nothing has been created yet. Exported + * for the tests that assert containment; callers have no reason to want it. + */ +export function currentRunRoot() { + return runRoot; +} + +/** + * Remove this process's run root now, rather than at exit. For a long-lived + * process that wants its scratch back mid-run; idempotent, and the exit + * handler stays registered so a later tmpDir() call is still covered. + */ +export function cleanupRunRoot() { + removeRunRoot(); + runRoot = null; +} + +// `kill(pid, 0)` probes existence without signalling. EPERM means the process +// exists and belongs to someone else, which still counts as alive. +function defaultIsAlive(pid) { + try { + process.kill(pid, 0); + return true; + } catch (err) { + return err.code === "EPERM"; + } +} + +/** + * Run roots left behind by OTHER runs — the SIGKILL residue, and the tell that + * a call site has stopped using this module. Reads only; deletes nothing. + * + * @param {object} [opts] + * @param {number} [opts.olderThanMs] age threshold, default 1 hour. A run root + * younger than this may well belong to a run that is still going, so it is + * not evidence of anything. + * @param {string} [opts.root] temp root to scan, default os.tmpdir(). + * @param {number} [opts.now] clock injection point, so the test does not have + * to sleep an hour to exercise the threshold. + * @param {(pid: number) => boolean} [opts.isAlive] liveness probe, injectable + * for the same reason. + * @returns {{count: number, dirs: string[], scanned: boolean, reason: string|null}} + * `scanned: false` with a reason is the third answer — the temp root could + * not be read, which is neither clean nor dirty and must not be reported as + * a count of zero. + */ +export function staleRunRoots({ + olderThanMs = 60 * 60 * 1000, + root = tmpdir(), + now = Date.now(), + isAlive = defaultIsAlive, +} = {}) { + let entries; + try { + entries = readdirSync(root, { withFileTypes: true }); + } catch (err) { + return { count: 0, dirs: [], scanned: false, reason: `cannot read ${root}: ${err.message}` }; + } + const dirs = []; + for (const e of entries) { + if (!e.isDirectory() || !e.name.startsWith(RUN_ROOT_PREFIX)) continue; + const full = join(root, e.name); + if (full === runRoot) continue; // our own, still in use + // A run that is still going owns its directory however old it is: a sweep + // over a large corpus legitimately outlives the threshold. + const pid = RUN_ROOT_RE.exec(e.name)?.[1]; + if (pid && isAlive(Number(pid))) continue; + try { + if (now - statSync(full).mtimeMs >= olderThanMs) dirs.push(full); + } catch { + // Vanished between readdir and stat — a concurrent run cleaning up after + // itself, which is the healthy case and not a finding. + } + } + dirs.sort(); + return { count: dirs.length, dirs, scanned: true, reason: null }; +}