From 9c8c72c7810669eed18ac9ad0e6d18617f75b19c Mon Sep 17 00:00:00 2001
From: z0d0dOr <128966342+z0d0dOr@users.noreply.github.com>
Date: Sun, 25 Oct 2026 09:52:00 -0700
Subject: [PATCH 1/7] fix: release input-only request data before streaming
 (#4753)

---
 src/plugin.ts                        |  45 +++++++--
 src/server.ts                        |  46 +++++++--
 src/server/handle.ts                 |  33 +++++--
 src/server/relay.ts                  |   8 ++
 tests/plugin-wire-metadata.test.ts   | 119 ++++++++++++++++++++++
 tests/proxy-readbody.test.ts         |  56 +++++++++++
 tests/request-input-lifetime.test.ts | 142 +++++++++++++++++++++++++++
 7 files changed, 423 insertions(+), 26 deletions(-)
 create mode 100644 tests/plugin-wire-metadata.test.ts
 create mode 100644 tests/request-input-lifetime.test.ts

diff --git a/src/plugin.ts b/src/plugin.ts
index 55024362d1..79f6b344e8 100644
--- a/src/plugin.ts
+++ b/src/plugin.ts
@@ -391,10 +391,30 @@ export function recordPluginSession(conversationId: string, sessionId: string):
     scheduleSaveConversations();
 }
 
+interface PluginWireMetadata {
+    wireHash: string;
+    textComparable: boolean;
+}
+
+/** Capture the raw request's snapshot evidence before prepare mutates its
+ *  parsed view. Only these scalars need to survive forwarding. */
+export function preparePluginWireMetadata(rawWire: Buffer, parsedWire: unknown): PluginWireMetadata {
+    let textComparable = false;
+    try {
+        if (parsedWire !== null && typeof parsedWire === "object") {
+            const wire = parsedWire as Record<string, unknown>;
+            textComparable = comparableHistory(wire.messages ?? wire.input ?? wire.contents);
+        }
+    } catch {
+        textComparable = false;
+    }
+    return { wireHash: createHash("sha256").update(rawWire).digest("hex"), textComparable };
+}
+
 /** Keep the last prepare()'s view for a plugin session so tool-API execution
  *  sees the exact refs the model was shown (mirrors the wire-mode loop, which
  *  runs executeProxyTool against prepared.processedMessages). */
-export function rememberPluginMessages(sessionId: string, processed: CoreMessage[], original: CoreMessage[], nudge?: NudgeDecision, rawWire?: Buffer): void {
+export function rememberPluginMessages(sessionId: string, processed: CoreMessage[], original: CoreMessage[], nudge?: NudgeDecision, rawWire?: Buffer | PluginWireMetadata): void {
     // #1307: auxiliary requests (auto-review / classifier prompts) bound to the
     // same session key can carry a normal output budget and any message count,
     // so both the ≤200 heuristic and size-based guards are proxies that a new
@@ -515,17 +535,22 @@ export function rememberPluginMessages(sessionId: string, processed: CoreMessage
             snapshotRefsEvaluatedStore.set(session, session.contentStore);
         }
         if (rawWire !== undefined) {
-            // The comparable verdict is a property of the wire body: hash it
-            // (native speed) and only re-parse + walk when the body actually
-            // changed, so a retry with an identical body stays cheap.
-            const wireHash = createHash("sha256").update(rawWire).digest("hex");
+            // The handler captures this evidence before forwarding so it need
+            // not retain the raw body. Buffer callers keep the hash-gated parse.
+            const wireHash = Buffer.isBuffer(rawWire)
+                ? createHash("sha256").update(rawWire).digest("hex")
+                : rawWire.wireHash;
             if (session.metadata.publicSnapshotWireHash !== wireHash) {
                 let comparable: boolean;
-                try {
-                    const wire = JSON.parse(rawWire.toString("utf8")) as Record<string, unknown>;
-                    comparable = comparableHistory(wire.messages ?? wire.input ?? wire.contents);
-                } catch {
-                    comparable = false;
+                if (Buffer.isBuffer(rawWire)) {
+                    try {
+                        const wire = JSON.parse(rawWire.toString("utf8")) as Record<string, unknown>;
+                        comparable = comparableHistory(wire.messages ?? wire.input ?? wire.contents);
+                    } catch {
+                        comparable = false;
+                    }
+                } else {
+                    comparable = rawWire.textComparable;
                 }
                 session.metadata.publicSnapshotWireHash = wireHash;
                 if (session.metadata.publicSnapshotTextComparable !== comparable) {
diff --git a/src/server.ts b/src/server.ts
index 2dd74daee8..6da0a0f161 100644
--- a/src/server.ts
+++ b/src/server.ts
@@ -4170,20 +4170,48 @@ export function readBody(req: http.IncomingMessage): Promise<Buffer> {
     return new Promise((resolve, reject) => {
         const chunks: Buffer[] = [];
         let size = 0;
-        let aborted = false;
-        req.on("data", (c: Buffer) => {
-            if (aborted) return;
+        let settled = false;
+        const cleanup = () => {
+            settled = true;
+            chunks.length = 0;
+            req.off("data", onData);
+            req.off("end", onEnd);
+            req.off("error", onError);
+            req.off("close", onClose);
+        };
+        const onError = (error: unknown) => {
+            if (settled) return;
+            cleanup();
+            reject(error);
+        };
+        const onData = (c: Buffer) => {
+            if (settled) return;
             size += c.length;
-            if (size > knobLimitsReceiveBytes()) {
-                aborted = true;
+            const limit = knobLimitsReceiveBytes();
+            if (size > limit) {
+                onError(new BodyTooLargeError(limit));
                 req.destroy();
-                reject(new BodyTooLargeError(knobLimitsReceiveBytes()));
                 return;
             }
             chunks.push(c);
-        });
-        req.on("end", () => { if (!aborted) resolve(Buffer.concat(chunks)); });
-        req.on("error", (e) => { if (!aborted) reject(e); });
+        };
+        const onEnd = () => {
+            if (settled) return;
+            try {
+                const body = Buffer.concat(chunks);
+                cleanup();
+                resolve(body);
+            } catch (error) {
+                onError(error);
+            }
+        };
+        // IncomingMessage also closes after a normal end, whose handler has
+        // already settled and removed this premature-close fallback.
+        const onClose = () => onError(new Error("request closed before body completed"));
+        req.on("data", onData);
+        req.on("end", onEnd);
+        req.on("error", onError);
+        req.on("close", onClose);
     });
 }
 
diff --git a/src/server/handle.ts b/src/server/handle.ts
index 1ec676fcd4..8d3737b7d3 100644
--- a/src/server/handle.ts
+++ b/src/server/handle.ts
@@ -40,7 +40,7 @@ import { maybeAdoptForkBlocks, maybeAdoptResume, maybeAdoptDerivedFork } from ".
 import { executeThreadsReforge } from "./threads-lane.js";
 import { resetThreadTailTracking } from "./thread-tracking.js";
 import { resolveClaudeTranscriptLineage } from "../claude-transcript-lineage.js";
-import { consumePluginRegisterFor, pluginAgentHeader, pluginConversationHeader, pluginHeadersMatchModel, pluginReportedContextWindow, pluginReportedMaxOutput, pluginReportedModel, pluginRequestAgentHeader, pluginRuntimeInfoFor, pluginRuntimeInfoForConversation, publicForkInputCheck, recordChainVerdict, recordPluginSession, rememberPluginMessages, resolveConversation, runtimeConversationId, takePendingPluginRegister } from "../plugin.js";
+import { consumePluginRegisterFor, pluginAgentHeader, pluginConversationHeader, pluginHeadersMatchModel, pluginReportedContextWindow, pluginReportedMaxOutput, pluginReportedModel, pluginRequestAgentHeader, pluginRuntimeInfoFor, pluginRuntimeInfoForConversation, preparePluginWireMetadata, publicForkInputCheck, recordChainVerdict, recordPluginSession, rememberPluginMessages, resolveConversation, runtimeConversationId, takePendingPluginRegister } from "../plugin.js";
 import { scheduleAffinityPersist } from "../affinity-persist.js";
 import { evaluateChain, extractChainCarriers } from "../chain-checkpoint.js";
 import { BILI_PASSTHROUGH_HEADER, BILI_PLUGIN_BYPASS_HEADER, isLoopbackAddress, normalizeUpstreamOrigin, provenCompletionReserve, reserveOutputHeadroom, resolveOutputHeadroomCap, shouldReserveOutputHeadroom, type WireProtocol } from "../util.js";
@@ -519,6 +519,9 @@ async function handleRequest(
             parsed = null;
         }
     }
+    // Capture the raw-wire verdict before unwrapping or preparing can mutate
+    // parsed. Only these scalars are needed for the post-forward bookkeeping.
+    const rawWireMetadata = protocol ? preparePluginWireMetadata(bodyBuffer, parsed) : undefined;
     // #3129: CloudCode/Antigravity v1internal dialect — unwrap the outer envelope
     // so the pipeline sees a standard GoogleRequest, and decide identity. The
     // inner request.sessionId is a constant FNV-1a hash, so identity comes from
@@ -545,6 +548,7 @@ async function handleRequest(
     //  index builds (each one re-converts the full history wire→core). The
     //  predicate over-approximates by design; see bodyMightCarryImages.
     const bodyMayCarryImages = protocol !== null && bodyText !== undefined ? bodyMightCarryImages(bodyText, protocol) : false;
+    bodyText = undefined;
     // #903: inbound message count for the per-request cost line — messages
     // (anthropic/openai) or input (responses); null when the body has neither.
     const inboundMsgs: number | null = (() => {
@@ -2381,10 +2385,10 @@ async function handleRequest(
             // #2155: self-heal evaluates once per HTTP request — the refold
             // re-runs prepare below but must not double-count a round.
             let selfHealRoundDone = false;
-            const runPreparedPipeline = async (
+            let runPreparedPipeline: ((
                 respondFailFast: boolean,
                 overflowWindow?: number,
-            ): Promise<{ body: string | Buffer; prepared: Prepared | null } | null> => {
+            ) => Promise<{ body: string | Buffer; prepared: Prepared | null } | null>) | undefined = async (respondFailFast, overflowWindow) => {
                 const runPrepare = async (): Promise<Prepared> => {
                     // #1820: this prepare consumes one unit of post-rebuild anchor
                     // validity (the last one deletes it — see session.ts).
@@ -2756,7 +2760,7 @@ async function handleRequest(
             if (guardScheme !== undefined && !resignArmable && !guardResign.enabled) {
                 clearSignedRefusal(guardScheme);
             }
-            const pendingForward = await withSessionLock(session, () => runPreparedPipeline(true));
+            const pendingForward = await withSessionLock(session, () => runPreparedPipeline!(true));
             if (pendingForward) {
                 forwarded = true;
                 // #1195: wire clients (omp/pi on the plain proxy path) treat an
@@ -2771,21 +2775,36 @@ async function handleRequest(
                 // original 400 through verbatim.
                 const overflowRefold = pendingForward.prepared
                     ? async (realWindow: number | undefined): Promise<{ body: string | Buffer; prepared: Prepared | null } | null> => {
-                          const next = await withSessionLock(session, () => runPreparedPipeline(false, realWindow));
+                          const next = await withSessionLock(session, () => runPreparedPipeline!(false, realWindow));
                           if (!next || String(next.body) === String(pendingForward.body)) return null;
                           pendingForward.body = next.body;
                           pendingForward.prepared = next.prepared;
                           return next;
                       }
                     : undefined;
-                await forward(req, res, opts, pendingForward.body, pendingForward.prepared, core, reqConfig, log, route, instanceId, affinity, overflowRefold, v1Meta);
+                const releaseRequestInput = () => {
+                    // Refolding is finished before streaming starts. Drop the
+                    // input-only bindings even if forward's wrapper retains the
+                    // callback; accepted outbound bodies still support retries.
+                    bodyBuffer = Buffer.alloc(0);
+                    parsed = null;
+                    affinityMessagesCache = null;
+                    runPreparedPipeline = undefined;
+                };
+                try {
+                    await forward(req, res, opts, pendingForward.body, pendingForward.prepared, core, reqConfig, log, route, instanceId, affinity, overflowRefold, v1Meta, releaseRequestInput);
+                } finally {
+                    // Transport failures and locally forged replies can return
+                    // before the upstream response boundary.
+                    releaseRequestInput();
+                }
                 // Remember for ALL modes (not just plugin): wire clients (dsh,
                 // hermes, unplug'd pi) read the same panel via /__bili/plugin/status
                 // and need the nudge/breakdown sections too; locked so a racing
                 // plugin tool call sees a consistent window.
                 if (pendingForward.prepared) {
                     const preparedToRemember = pendingForward.prepared;
-                    await withSessionLock(session, () => rememberPluginMessages(sessionId, preparedToRemember.processedMessages, preparedToRemember.originalMessages, preparedToRemember.nudge, bodyBuffer));
+                    await withSessionLock(session, () => rememberPluginMessages(sessionId, preparedToRemember.processedMessages, preparedToRemember.originalMessages, preparedToRemember.nudge, rawWireMetadata));
                 }
             }
         } finally {
diff --git a/src/server/relay.ts b/src/server/relay.ts
index e085989315..ab4e74f6d6 100644
--- a/src/server/relay.ts
+++ b/src/server/relay.ts
@@ -265,11 +265,13 @@ async function forwardCaptured(
     affinity?: string,
     overflowRefold?: (realWindow: number | undefined) => Promise<{ body: string | Buffer; prepared: Prepared | null } | null>,
     v1Meta?: Record<string, unknown>,
+    releaseRequestInput?: () => void,
 ): Promise<void> {
     const forwardStartedAt = Date.now();
     // E2: a codex native-compaction request intercepted in prepare() carries a
     // forged success response — serve it without contacting upstream.
     if (prepared?.codexForge) {
+        releaseRequestInput?.();
         log("info", `[${prepared.session.id}] codex compact served locally (${prepared.codexForge.kind}); upstream not contacted`);
         if (!res.headersSent) res.writeHead(200, { "content-type": prepared.codexForge.contentType });
         res.end(prepared.codexForge.body);
@@ -279,6 +281,7 @@ async function forwardCaptured(
     // zero-egress contract as codexForge; the audit line was written in
     // prepare (this line is the egress-proof for the serve itself).
     if (prepared?.autoModeForge) {
+        releaseRequestInput?.();
         log("info", `[${prepared.session.id}] auto-mode classifier served locally (${prepared.autoModeForge.variant}); upstream not contacted`);
         if (!res.headersSent) res.writeHead(200, { "content-type": prepared.autoModeForge.contentType });
         res.end(prepared.autoModeForge.body);
@@ -1444,6 +1447,11 @@ async function forwardCaptured(
             }
         }
     }
+    // No further overflow refold can run. Release its inbound-only state before
+    // consuming a potentially long response; wireBody still owns retry/echo data.
+    releaseRequestInput?.();
+    releaseRequestInput = undefined;
+    overflowRefold = undefined;
     const { response: upstream, clearTimer: clearUpstreamTimer } = upstreamResult;
     // [#1343] delivery decided: the attached full text rode THIS request's body, so settle
     // its lifecycle here — delivered on a 2xx, dropped-and-logged otherwise. The ack is
diff --git a/tests/plugin-wire-metadata.test.ts b/tests/plugin-wire-metadata.test.ts
new file mode 100644
index 0000000000..f1f3906cfe
--- /dev/null
+++ b/tests/plugin-wire-metadata.test.ts
@@ -0,0 +1,119 @@
+import assert from "node:assert/strict";
+import { createHash } from "node:crypto";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import test from "node:test";
+import type { CoreMessage } from "acp-kernel";
+import { preparePluginWireMetadata, rememberPluginMessages } from "../src/plugin.ts";
+import { SessionStore, _setStoreForTest } from "../src/persist.ts";
+import { _resetSessionsForTest, getSession } from "../src/session.ts";
+
+const messages: CoreMessage[] = [
+    { id: "raw-a", role: "user", contentType: "text", text: "first" },
+    { id: "raw-b", role: "assistant", contentType: "text", text: "second" },
+    { id: "raw-c", role: "user", contentType: "text", text: "third" },
+];
+
+test.beforeEach(() => {
+    _resetSessionsForTest();
+    _setStoreForTest(new SessionStore({ dir: join(tmpdir(), "bili-wire-metadata-disabled"), enabled: false }));
+});
+test.afterEach(() => _resetSessionsForTest());
+
+function pluginSession(id: string) {
+    const session = getSession(id);
+    session.metadata.pluginAgent = "codex";
+    return session;
+}
+
+test("precomputed wire metadata matches raw-buffer snapshot evidence across wire shapes", () => {
+    const cases: { wire: unknown; comparable: boolean; raw?: string }[] = [
+        { wire: { messages: [{ role: "user", content: "中😀" }] }, comparable: true },
+        { wire: { input: [{ type: "message", content: [{ type: "input_text", text: "text" }] }] }, comparable: true },
+        { wire: { contents: [{ role: "user", parts: [{ text: "text" }] }] }, comparable: true },
+        { wire: { messages: null, input: "fallback" }, comparable: true },
+        { wire: { messages: [{ content: [{ type: "image", source: {} }] }], input: "ignored" }, comparable: false },
+        { wire: { messages: [{ content: [{ type: "tool_use", input: { type: "image" } }] }] }, comparable: true },
+        { wire: { messages: [{ content: [{ type: "thinking", thinking: "signed" }] }] }, comparable: false },
+        { wire: { messages: [{ content: "text", audio: {} }] }, comparable: false },
+        { wire: null, comparable: false },
+        { wire: 42, comparable: false },
+        { wire: "text", comparable: false },
+        { wire: undefined, raw: "{invalid", comparable: false },
+    ];
+    for (const [i, { wire, comparable, raw }] of cases.entries()) {
+        const body = Buffer.from(raw ?? JSON.stringify(wire, null, 2));
+        const evidence = preparePluginWireMetadata(body, wire);
+        assert.deepEqual(evidence, {
+            wireHash: createHash("sha256").update(body).digest("hex"),
+            textComparable: comparable,
+        });
+        const legacy = pluginSession(`raw-${i}`);
+        const prepared = pluginSession(`prepared-${i}`);
+        rememberPluginMessages(legacy.id, messages, messages, undefined, body);
+        rememberPluginMessages(prepared.id, messages, messages, undefined, evidence);
+        assert.equal(prepared.metadata.publicSnapshotWireHash, legacy.metadata.publicSnapshotWireHash);
+        assert.equal(prepared.metadata.publicSnapshotTextComparable, legacy.metadata.publicSnapshotTextComparable);
+        assert.equal(prepared.revisionEpoch, legacy.revisionEpoch);
+    }
+});
+
+test("preparing wire metadata neither reparses nor commits, and survives input mutation", (t) => {
+    const session = pluginSession("deferred");
+    const wire: Record<string, unknown> = { messages: [{ role: "user", content: "original" }] };
+    const body = Buffer.from(JSON.stringify(wire));
+    const expectedHash = createHash("sha256").update(body).digest("hex");
+    const parse = t.mock.method(JSON, "parse", () => { throw new Error("must reuse the parsed request"); });
+    const evidence = preparePluginWireMetadata(body, wire);
+    assert.equal(parse.mock.callCount(), 0);
+    parse.mock.restore();
+    assert.equal(session.metadata.publicSnapshotWireHash, undefined);
+    assert.equal(session.metadata.publicSnapshotTextComparable, undefined);
+    assert.equal(session.pluginSnapshot, undefined);
+    assert.equal(session.revisionEpoch, undefined);
+
+    body.fill(0);
+    wire.messages = [{ content: [{ type: "image", source: {} }] }];
+    rememberPluginMessages(session.id, messages, messages, undefined, evidence);
+    assert.equal(session.metadata.publicSnapshotWireHash, expectedHash);
+    assert.equal(session.metadata.publicSnapshotTextComparable, true);
+});
+
+test("wire evidence preserves hash-gated bookkeeping and rejected-view guards", () => {
+    const session = pluginSession("bookkeeping");
+    const wire = { messages: [{ role: "user", content: "text" }] };
+    const evidence = preparePluginWireMetadata(Buffer.from(JSON.stringify(wire)), wire);
+    rememberPluginMessages(session.id, messages, messages, undefined, evidence);
+    const revision = session.revisionEpoch;
+    const snapshot = session.pluginSnapshot;
+    rememberPluginMessages(session.id, messages, messages, undefined, evidence);
+    assert.equal(session.revisionEpoch, revision, "identical retries do not mark dirty");
+
+    const reformatted = preparePluginWireMetadata(Buffer.from(JSON.stringify(wire, null, 2)), wire);
+    assert.notEqual(reformatted.wireHash, evidence.wireHash, "hash follows exact request bytes");
+    rememberPluginMessages(session.id, messages, messages, undefined, reformatted);
+    assert.equal(session.metadata.publicSnapshotWireHash, reformatted.wireHash);
+    assert.equal(session.revisionEpoch, revision, "a new hash with the same verdict does not mark dirty");
+
+    const opaque = { input: [{ type: "input_image", image_url: "data:image/png;base64,AA==" }] };
+    const opaqueEvidence = preparePluginWireMetadata(Buffer.from(JSON.stringify(opaque)), opaque);
+    rememberPluginMessages(session.id, messages, messages, undefined, opaqueEvidence);
+    assert.equal(session.metadata.publicSnapshotTextComparable, false);
+    assert.equal(session.revisionEpoch, (revision ?? 0) + 1, "changed comparability marks dirty");
+    assert.equal(session.pluginSnapshot, snapshot, "wire evidence does not replace the message snapshot");
+    rememberPluginMessages(session.id, messages, messages, undefined, { ...opaqueEvidence, textComparable: true });
+    assert.equal(session.metadata.publicSnapshotTextComparable, false, "the saved hash gates verdict changes");
+
+    const side: CoreMessage[] = [{ id: "unrelated", role: "user", contentType: "text", text: "side request" }];
+    rememberPluginMessages(session.id, side, side, undefined, evidence);
+    assert.equal(session.metadata.publicSnapshotWireHash, opaqueEvidence.wireHash);
+    assert.equal(session.metadata.publicSnapshotTextComparable, false);
+    assert.equal(session.revisionEpoch, (revision ?? 0) + 1);
+
+    const nonPlugin = getSession("non-plugin");
+    rememberPluginMessages(nonPlugin.id, messages, messages, undefined, evidence);
+    assert.equal(nonPlugin.metadata.publicSnapshotWireHash, undefined);
+    const empty = pluginSession("empty");
+    rememberPluginMessages(empty.id, [], [], undefined, evidence);
+    assert.equal(empty.metadata.publicSnapshotWireHash, undefined);
+});
diff --git a/tests/proxy-readbody.test.ts b/tests/proxy-readbody.test.ts
index 2349ea2b55..6903af5108 100644
--- a/tests/proxy-readbody.test.ts
+++ b/tests/proxy-readbody.test.ts
@@ -2,6 +2,8 @@ import { test } from "node:test";
 import assert from "node:assert/strict";
 import { Readable } from "node:stream";
 import { EventEmitter } from "node:events";
+import { IncomingMessage } from "node:http";
+import { Socket } from "node:net";
 import { BodyTooLargeError, readBody } from "../src/server.ts";
 import { MAX_REQUEST_BYTES } from "../src/fetch-util.ts";
 
@@ -19,6 +21,60 @@ test("readBody: returns buffer for a normal request", async () => {
     assert.equal(out.toString("utf8"), "hello");
 });
 
+for (const outcome of ["success", "error"] as const) {
+    test(`readBody: releases its request listeners after ${outcome}`, async () => {
+        const req = new EventEmitter();
+        const externalListener = () => {};
+        const events = ["data", "end", "error", "close"] as const;
+        for (const event of events) req.on(event, externalListener);
+        const pending = readBody(req as never);
+        req.emit("data", Buffer.from("hello"));
+        if (outcome === "success") {
+            req.emit("end");
+            assert.equal((await pending).toString("utf8"), "hello");
+        } else {
+            const error = new Error("read failure");
+            const rejection = assert.rejects(pending, (actual: unknown) => actual === error);
+            req.emit("error", error);
+            await rejection;
+        }
+        for (const event of events) {
+            assert.deepEqual(req.listeners(event), [externalListener], `${event}: readBody must remove only its own listener`);
+        }
+    });
+}
+
+for (const withError of [false, true]) {
+    test(`readBody: real IncomingMessage cancellation ${withError ? "with" : "without"} an error settles and cleans up`, { timeout: 2000 }, async () => {
+        const req = new IncomingMessage(new Socket());
+        const error = withError ? new Error("client disconnected") : undefined;
+        const closed = new Promise<void>((resolve) => req.once("close", resolve));
+        const pending = readBody(req);
+        const rejection = assert.rejects(pending, (actual: unknown) => withError ? actual === error : actual instanceof Error);
+        req.push(Buffer.from("hello"));
+        setImmediate(() => req.destroy(error));
+        await rejection;
+        await closed;
+        for (const event of ["data", "end", "error", "close"]) assert.equal(req.listenerCount(event), 0, event);
+    });
+}
+
+test("readBody: a real IncomingMessage over the limit keeps BodyTooLargeError through destroy", { timeout: 2000 }, async (t) => {
+    const previousLimit = process.env.BILI_RECEIVE_BODY_BYTES;
+    process.env.BILI_RECEIVE_BODY_BYTES = "8";
+    t.after(() => { process.env.BILI_RECEIVE_BODY_BYTES = previousLimit; });
+    const req = new IncomingMessage(new Socket());
+    const closed = new Promise<void>((resolve) => req.once("close", resolve));
+    const pending = readBody(req);
+    const rejection = assert.rejects(pending, (error: unknown) => error instanceof BodyTooLargeError && error.limit === 8);
+    req.push(Buffer.from("1234"));
+    req.push(Buffer.from("56789"));
+    await rejection;
+    await closed;
+    assert.equal(req.destroyed, true);
+    for (const event of ["data", "end", "error", "close"]) assert.equal(req.listenerCount(event), 0, event);
+});
+
 test("readBody: rejects with BodyTooLargeError AND destroys socket when body exceeds MAX_REQUEST_BYTES", async () => {
     const req = new EventEmitter();
     let destroyed = false;
diff --git a/tests/request-input-lifetime.test.ts b/tests/request-input-lifetime.test.ts
new file mode 100644
index 0000000000..bdc8f8711c
--- /dev/null
+++ b/tests/request-input-lifetime.test.ts
@@ -0,0 +1,142 @@
+import assert from "node:assert/strict";
+import http from "node:http";
+import { once } from "node:events";
+import { randomUUID } from "node:crypto";
+import test from "node:test";
+import { createCore, defaultConfig, type CoreMessage } from "acp-kernel";
+import type { ProxyOptions } from "../src/config.ts";
+import type { Prepared } from "../src/server.ts";
+import { forward } from "../src/server/relay.ts";
+import { getSession } from "../src/session.ts";
+import { SessionStore, _setStoreForTest } from "../src/persist.ts";
+import { _liveUpstreamTimersForTest, _resetFetchUtilForTest } from "../src/fetch-util.ts";
+
+process.env.NODE_ENV = "test";
+_setStoreForTest(new SessionStore({ enabled: false }));
+
+const OVERFLOW = JSON.stringify({ error: { type: "invalid_request_error", message: "prompt is too long: 130000 tokens > 128000 maximum" } });
+const RETRY_REJECTION = JSON.stringify({ error: { type: "invalid_request_error", message: "refold retry rejected" } });
+function deferred<T>() {
+    let resolve!: (value: T) => void;
+    let reject!: (error: unknown) => void;
+    const promise = new Promise<T>((yes, no) => { resolve = yes; reject = no; });
+    return { promise, resolve, reject };
+}
+
+for (const scenario of ["success", "refold", "declined", "throwing", "rejected"] as const) {
+    test(`request input release boundary: ${scenario}`, { timeout: 10_000 }, async t => {
+        const url = "http://request-lifetime.test/responses";
+        const session = getSession(`input-lifetime-${randomUUID()}`);
+        const core = createCore();
+        const config = defaultConfig(200_000);
+        const messages: CoreMessage[] = [{ id: "m1", role: "user", contentType: "text", text: "Continue." }];
+        const body = JSON.stringify({ model: "test-model", stream: true, input: [{ role: "user", content: "Continue." }] });
+        const prepared: Prepared = {
+            body, session, processedMessages: messages, originalMessages: messages,
+            protocol: "responses", stream: true, compressInjected: false,
+        };
+        const opts: ProxyOptions = {
+            host: "127.0.0.1", port: 0, upstream: url, routes: {},
+            proxy: "", proxyMode: "direct", proxySource: "direct",
+            modelContextLimit: 200_000, kernelConfig: config,
+            compress: { injectTool: false, injectNudge: false },
+            promptCache: { routing: "auto" }, compat: { roles: {} },
+            streamErrorShape: "protocol", sessionHeader: "x-acp-session",
+            log: false, debug: false, passthrough: false, passthroughSource: null,
+            autoUpdate: false, autoRestartOnUpdate: false, updateTag: "latest",
+            advisoryCheck: false, releaseNotesCheck: false, mitm: { enabled: false, domains: [] },
+        };
+        const released = deferred<void>();
+        const completed = deferred<void>();
+        const trace: string[] = [];
+        let releaseCount = 0;
+        let forwardCompleted = false;
+        let upstreamCalls = 0;
+        let streamController: ReadableStreamDefaultController<Uint8Array> | undefined;
+        const nativeFetch = globalThis.fetch;
+        t.mock.method(globalThis, "fetch", async (_input: unknown, init?: RequestInit) => {
+            upstreamCalls++;
+            trace.push(`fetch:${upstreamCalls}`);
+            assert.equal(releaseCount, 0, "inbound reconstruction must remain available until the retry is accepted");
+            if (scenario !== "success" && upstreamCalls === 1) {
+                return new Response(OVERFLOW, { status: 400, headers: { "content-type": "application/json" } });
+            }
+            assert.equal(init?.body, upstreamCalls === 1 ? body : `${body} `, "the accepted request keeps its outgoing bytes");
+            if (scenario === "rejected") {
+                return new Response(RETRY_REJECTION, { status: 422, headers: { "content-type": "application/json" } });
+            }
+            return new Response(new ReadableStream<Uint8Array>({ start(controller) { streamController = controller; } }), {
+                headers: { "content-type": "text/event-stream" },
+            });
+        });
+        const server = http.createServer((req, res) => {
+            req.resume();
+            req.url = url;
+            void forward(req, res, opts, body, prepared, core, config, () => {}, undefined, "input-lifetime",
+                undefined,
+                async () => {
+                    trace.push("refold");
+                    assert.equal(releaseCount, 0, "the refold callback still owns original input");
+                    if (scenario === "declined") return null;
+                    if (scenario === "throwing") throw new Error("cannot refold");
+                    return { body: `${body} `, prepared: { ...prepared, body: `${body} ` } };
+                },
+                undefined,
+                () => { releaseCount++; trace.push("release"); released.resolve(); },
+            ).then(() => { forwardCompleted = true; completed.resolve(); }, error => { res.destroy(); completed.reject(error); });
+        });
+        server.listen(0, "127.0.0.1");
+        await once(server, "listening");
+        const address = server.address();
+        assert.ok(address && typeof address === "object");
+        const client = nativeFetch(`http://127.0.0.1:${address.port}`, { method: "POST", body });
+        // Attach rejection handlers immediately; cleanup can abort either side.
+        void client.catch(() => {});
+        void completed.promise.catch(() => {});
+        let deadline: ReturnType<typeof setTimeout> | undefined;
+        try {
+            if (scenario === "success" || scenario === "refold") {
+                await Promise.race([
+                    released.promise,
+                    completed.promise.then(() => assert.fail("forward finished before input release")),
+                    new Promise<never>((_, reject) => { deadline = setTimeout(() => reject(new Error("input retained while accepted response stream is pending")), 5000); }),
+                ]);
+                assert.equal(forwardCompleted, false);
+                assert.equal(releaseCount, 1);
+                assert.ok(streamController, "upstream stream remains pending at the release boundary");
+                streamController.enqueue(new TextEncoder().encode(`event: response.completed\ndata: ${JSON.stringify({ type: "response.completed", response: { id: "r1", status: "completed", output: [{ type: "message", role: "assistant", content: [{ type: "output_text", text: "done" }] }] } })}\n\n`));
+                streamController.close();
+                streamController = undefined;
+            }
+            const response = await client;
+            const responseText = await response.text();
+            await completed.promise;
+            assert.equal(releaseCount, 1, "release runs exactly once, including error responses");
+            if (scenario === "declined" || scenario === "throwing") {
+                assert.equal(response.status, 400);
+                assert.equal(responseText, OVERFLOW);
+            } else if (scenario === "rejected") {
+                assert.equal(response.status, 422);
+                assert.equal(responseText, RETRY_REJECTION);
+            } else {
+                assert.equal(response.status, 200);
+                assert.ok(responseText.includes("done"));
+            }
+            const expectsRetry = scenario === "refold" || scenario === "rejected";
+            assert.equal(upstreamCalls, expectsRetry ? 2 : 1);
+            assert.deepEqual(trace, scenario === "success" ? ["fetch:1", "release"]
+                : expectsRetry ? ["fetch:1", "refold", "fetch:2", "release"]
+                : ["fetch:1", "refold", "release"]);
+        } finally {
+            if (deadline) clearTimeout(deadline);
+            streamController?.close();
+            const closed = once(server, "close");
+            server.close();
+            server.closeAllConnections();
+            await closed;
+            await completed.promise.catch(() => {});
+            assert.equal(_liveUpstreamTimersForTest(), 0);
+            _resetFetchUtilForTest();
+        }
+    });
+}

From 5ebf7f752b31cbbb52d0ebd289241e426ab2495c Mon Sep 17 00:00:00 2001
From: z0d0dOr <128966342+z0d0dOr@users.noreply.github.com>
Date: Sun, 25 Oct 2026 12:09:26 -0700
Subject: [PATCH 2/7] fix: preserve blog metadata and launcher checks on
 Windows

---
 src/web/blogfeed.ts                    |  2 ++
 tests/launcher-opencode-inline.test.ts | 20 +++++++++++---------
 tests/web-blogfeed.test.ts             | 17 ++++++++++++++++-
 3 files changed, 29 insertions(+), 10 deletions(-)

diff --git a/src/web/blogfeed.ts b/src/web/blogfeed.ts
index 6efe1560a7..0d67c61d8f 100644
--- a/src/web/blogfeed.ts
+++ b/src/web/blogfeed.ts
@@ -76,6 +76,8 @@ export function _setBlogRootForTest(root: string | null): void {
 
 /** Mirror of the website tools' frontmatter parser (key: value between --- fences). */
 function parseFrontmatter(text: string): { fm: Record<string, string>; body: string } {
+    // Windows checkouts use CRLF; parse and render the same Markdown on every platform.
+    text = text.replace(/\r\n/g, "\n");
     const fm: Record<string, string> = {};
     let body = text;
     if (text.startsWith("---\n")) {
diff --git a/tests/launcher-opencode-inline.test.ts b/tests/launcher-opencode-inline.test.ts
index 7ff04b2ace..fd5452a332 100644
--- a/tests/launcher-opencode-inline.test.ts
+++ b/tests/launcher-opencode-inline.test.ts
@@ -28,6 +28,7 @@ interface SpawnRecord {
     command: string;
     args: string[];
     env: NodeJS.ProcessEnv | undefined;
+    windowsVerbatimArguments?: boolean;
 }
 
 // `bili opencode-inline` (#4669): the launcher must start NO proxy, hand the
@@ -37,16 +38,12 @@ interface SpawnRecord {
 // clone (plugin injection only; 2.x directory wrapper when the host is 2.x).
 test("runLaunch opencode-inline: no proxy spawned, BILI_OPENCODE_INLINE=1, attach/proxy env stripped, plugin rides OPENCODE_CONFIG", async () => {
     const home = fs.mkdtempSync(path.join(os.tmpdir(), "bili-ocinline-"));
-    const prevHome = process.env.HOME;
-    const prevUserProfile = process.env.USERPROFILE;
     const prevXdg = process.env.XDG_CONFIG_HOME;
     const prevOcConfig = process.env.OPENCODE_CONFIG;
     const prevClientBin = process.env.BILI_CLIENT_BIN;
     const prevBcp = process.env.BILLION_CONTEXT_PROXY;
     const prevHttpsProxy = process.env.HTTPS_PROXY;
     const prevMitmHosts = process.env.BILI_MITM_HOSTS;
-    process.env.HOME = home;
-    if (prevUserProfile !== undefined) process.env.USERPROFILE = home;
     // Isolated config root: no user opencode.json, no ambient OPENCODE_CONFIG.
     process.env.XDG_CONFIG_HOME = path.join(home, ".config");
     delete process.env.OPENCODE_CONFIG;
@@ -75,7 +72,7 @@ test("runLaunch opencode-inline: no proxy spawned, BILI_OPENCODE_INLINE=1, attac
 
     const spawns: SpawnRecord[] = [];
     const spawnImpl: SpawnFn = (cmd, args, options) => {
-        spawns.push({ command: cmd, args: [...args], env: options?.env });
+        spawns.push({ command: cmd, args: [...args], env: options?.env, windowsVerbatimArguments: options?.windowsVerbatimArguments });
         const child = makeFakeChild(0);
         const orig = child.on!.bind(child);
         (child as { on: SpawnChild["on"] }).on = (event, listener) => {
@@ -109,8 +106,16 @@ test("runLaunch opencode-inline: no proxy spawned, BILI_OPENCODE_INLINE=1, attac
             { fetchImpl, fetchHealthInfo, spawnImpl, sleep: () => Promise.resolve() },
         );
         assert.equal(spawns.length, 1, "exactly one child (opencode itself) — no proxy spawn");
-        assert.equal(spawns[0].command, fakeOc);
         const env = spawns[0].env!;
+        if (process.platform === "win32") {
+            assert.equal(spawns[0].command, env.COMSPEC?.trim() ? env.COMSPEC : "cmd.exe");
+            const quotedOc = /\s/.test(fakeOc) ? `"${fakeOc}"` : fakeOc;
+            assert.deepEqual(spawns[0].args, ["/d", "/s", "/c", `"${quotedOc}"`]);
+            assert.equal(spawns[0].windowsVerbatimArguments, true);
+        } else {
+            assert.equal(spawns[0].command, fakeOc);
+            assert.deepEqual(spawns[0].args, []);
+        }
         assert.equal(env.BILI_OPENCODE_INLINE, "1");
         assert.equal(env.BILLION_CONTEXT_PROXY, undefined, "inherited BILLION_CONTEXT_PROXY must be stripped (attach would win)");
         assert.equal(env.HTTPS_PROXY, undefined, "inherited HTTPS_PROXY must be stripped (no MITM lane)");
@@ -134,9 +139,6 @@ test("runLaunch opencode-inline: no proxy spawned, BILI_OPENCODE_INLINE=1, attac
         assert.equal(clone.compaction?.auto, false, "host auto-compaction stays off (ACP owns compression)");
     } finally {
         process.exit = prevExit;
-        process.env.HOME = prevHome;
-        if (prevUserProfile === undefined) delete process.env.USERPROFILE;
-        else process.env.USERPROFILE = prevUserProfile;
         if (prevXdg === undefined) delete process.env.XDG_CONFIG_HOME;
         else process.env.XDG_CONFIG_HOME = prevXdg;
         if (prevOcConfig === undefined) delete process.env.OPENCODE_CONFIG;
diff --git a/tests/web-blogfeed.test.ts b/tests/web-blogfeed.test.ts
index 0d1d820907..5280ba9dd3 100644
--- a/tests/web-blogfeed.test.ts
+++ b/tests/web-blogfeed.test.ts
@@ -9,7 +9,7 @@
  */
 import { test, beforeEach, afterEach } from "node:test";
 import assert from "node:assert/strict";
-import { mkdtempSync, rmSync, writeFileSync, mkdirSync } from "node:fs";
+import { mkdtempSync, rmSync, readFileSync, writeFileSync, mkdirSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import type { ServerResponse } from "node:http";
@@ -93,6 +93,21 @@ test("rendering is escape-first: script tags survive only escaped; tables become
     assert.ok(alpha.html.includes("<td>two</td>"));
 });
 
+test("CRLF markdown preserves bilingual frontmatter and rendered content", () => {
+    const expected = getBlogFeed();
+    for (const relative of [
+        "website/blog/posts/2026-10-08_demo.md",
+        "website/blog/posts/2026-10-08_demo.zh.md",
+        "paper/a.md",
+        "paper/b.md",
+    ]) {
+        const file = join(pkgRoot, relative);
+        writeFileSync(file, readFileSync(file, "utf8").replace(/\n/g, "\r\n"));
+    }
+    _setBlogRootForTest(pkgRoot);
+    assert.deepEqual(getBlogFeed(), expected);
+});
+
 test("image refs: shipped files reroute to the panel asset prefix; traversal/unshipped refs drop silently", () => {
     const data = getBlogFeed();
     const demo = data.items.find((i) => i.id === "demo")!;

From 1fcb64203e32842be9e23c35c8e8802c40656e76 Mon Sep 17 00:00:00 2001
From: z0d0dOr <128966342+z0d0dOr@users.noreply.github.com>
Date: Sun, 25 Oct 2026 12:35:44 -0700
Subject: [PATCH 3/7] test: verify fork restart under legacy and full storage

---
 tests/plugin-fork.test.ts | 136 ++++++++++++++++++++------------------
 1 file changed, 71 insertions(+), 65 deletions(-)

diff --git a/tests/plugin-fork.test.ts b/tests/plugin-fork.test.ts
index d15bc89fe9..caf2c00688 100644
--- a/tests/plugin-fork.test.ts
+++ b/tests/plugin-fork.test.ts
@@ -114,7 +114,7 @@ async function harness(persist = false, seedParent = true, compressExtra: Record
         assert.equal(r.status, 200);
         await r.text();
     }
-    return { request, messages, forwarded, dir, store, upstreamUrl, origin, close: async () => { store.cancelAll(); await closeServers([proxy, upstream], "plugin-fork harness"); } };
+    return { request, messages, forwarded, dir, store, upstreamUrl, origin, close: async () => { store.closeQuietly(); await closeServers([proxy, upstream], "plugin-fork harness"); } };
 }
 
 test("plain proxy panel snapshots do not change raw-history persistence", { timeout: 60_000 }, async () => {
@@ -742,71 +742,77 @@ for (const corruption of ["missing-ref", "missing-payload", "orphan-payload", "w
     });
 }
 
-test("HTTP fork revision and originals survive a new proxy process", { timeout: 60_000 }, async () => {
-    const h = await harness(true);
-    let child: ReturnType<typeof spawn> | undefined;
-    try {
-        await compress(h);
-        const parent = resolveConversation("parent").session!;
-        parent.metadata.effectiveCcr = { ...DEFAULT_CCR_CONFIG, enabled: true };
-        parent.contentStore = storeOriginal(createContentStore(), { ref: "m00001", rawId: parent.pluginSnapshot![0].id, text: "CCR original retained independently after restart", kind: "original", tokens: 10, head: "CCR original" });
-        parent.contentStoreDirty = true;
-        assert(h.store.flushSync(parent));
-        // #4316 PR3/3: cold load runs the one-time ledger alignment — the loaded
-        // state now carries refLedgerAligned. Mirror it in the expectation.
-        const persistedState = JSON.parse(JSON.stringify(parent.state));
-        persistedState.refLedgerAligned = true;
-        assert.deepEqual(JSON.parse(JSON.stringify(h.store.loadSync("parent")!.state)), persistedState);
-        const snapshot = (await h.request("/__bili/plugin/snapshot?conversationId=parent")).body;
-        const req = forkRequest(snapshot);
-        const fork = await h.request("/__bili/plugin/fork", req);
-        assert.equal(fork.status, 201);
-        const childSnapshot = (await h.request("/__bili/plugin/snapshot?conversationId=child")).body;
-        const code = `import {startServer} from './src/server.ts'; import {SessionStore,_setStoreForTest} from './src/persist.ts'; import {defaultConfig} from 'acp-kernel'; import {_setForTest} from './src/registry.ts'; _setForTest({}); _setStoreForTest(new SessionStore({dir:${JSON.stringify(h.dir + "/sessions")},enabled:true})); const s=await startServer({port:0,host:'127.0.0.1',upstream:${JSON.stringify(h.upstreamUrl)},modelContextLimit:400000,routes:{},kernelConfig:defaultConfig(400000),compress:{injectTool:true,injectNudge:true},promptCache:{routing:'auto'},sessionHeader:'x-acp-session',log:false,debug:false,passthrough:false,autoUpdate:false,mitm:{enabled:false,domains:[]}}); const ready=()=>process.stdout.write(JSON.stringify({port:s.address().port})+'\\n'); if(s.listening)ready();else s.once('listening',ready);`;
-        child = spawn(process.execPath, ["--import", "tsx", "--input-type=module", "--eval", code], { cwd: fileURLToPath(new URL("../", import.meta.url)), env: { ...process.env, NODE_ENV: "test" }, stdio: ["ignore", "pipe", "pipe"] });
-        let stderr = "";
-        child.stderr!.on("data", (b) => { stderr += b.toString(); });
-        const port = await new Promise<number>((resolve, reject) => {
-            let stdout = "";
-            const timer = setTimeout(() => reject(new Error("restart timeout: " + stderr + stdout)), 15000);
-            child!.stdout!.on("data", (b) => {
-                stdout += b.toString();
-                const port = stdout.match(/\{"port":(\d+)\}/);
-                if (port) { clearTimeout(timer); resolve(Number(port[1])); }
+for (const storageMode of ["legacy", "full"] as const) {
+    test(`HTTP fork revision and originals survive a new proxy process (${storageMode})`, { timeout: 60_000 }, async () => {
+        const previousMode = process.env.BILI_STORAGE_UNIFIED;
+        process.env.BILI_STORAGE_UNIFIED = storageMode;
+        const h = await harness(true);
+        let child: ReturnType<typeof spawn> | undefined;
+        try {
+            await compress(h);
+            const parent = resolveConversation("parent").session!;
+            parent.metadata.effectiveCcr = { ...DEFAULT_CCR_CONFIG, enabled: true };
+            parent.contentStore = storeOriginal(createContentStore(), { ref: "m00001", rawId: parent.pluginSnapshot![0].id, text: "CCR original retained independently after restart", kind: "original", tokens: 10, head: "CCR original" });
+            parent.contentStoreDirty = true;
+            assert(h.store.flushSync(parent));
+            // #4316 PR3/3: only a unified cold load runs the one-time ledger
+            // alignment; the CI legacy baseline retains the original state.
+            const persistedState = JSON.parse(JSON.stringify(parent.state));
+            if (storageMode === "full") persistedState.refLedgerAligned = true;
+            assert.deepEqual(JSON.parse(JSON.stringify(h.store.loadSync("parent")!.state)), persistedState);
+            const snapshot = (await h.request("/__bili/plugin/snapshot?conversationId=parent")).body;
+            const req = forkRequest(snapshot);
+            const fork = await h.request("/__bili/plugin/fork", req);
+            assert.equal(fork.status, 201);
+            const childSnapshot = (await h.request("/__bili/plugin/snapshot?conversationId=child")).body;
+            const code = `import {startServer} from './src/server.ts'; import {SessionStore,_setStoreForTest} from './src/persist.ts'; import {defaultConfig} from 'acp-kernel'; import {_setForTest} from './src/registry.ts'; _setForTest({}); _setStoreForTest(new SessionStore({dir:${JSON.stringify(h.dir + "/sessions")},enabled:true})); const s=await startServer({port:0,host:'127.0.0.1',upstream:${JSON.stringify(h.upstreamUrl)},modelContextLimit:400000,routes:{},kernelConfig:defaultConfig(400000),compress:{injectTool:true,injectNudge:true},promptCache:{routing:'auto'},sessionHeader:'x-acp-session',log:false,debug:false,passthrough:false,autoUpdate:false,mitm:{enabled:false,domains:[]}}); const ready=()=>process.stdout.write(JSON.stringify({port:s.address().port})+'\\n'); if(s.listening)ready();else s.once('listening',ready);`;
+            child = spawn(process.execPath, ["--import", "tsx", "--input-type=module", "--eval", code], { cwd: fileURLToPath(new URL("../", import.meta.url)), env: { ...process.env, NODE_ENV: "test" }, stdio: ["ignore", "pipe", "pipe"] });
+            let stderr = "";
+            child.stderr!.on("data", (b) => { stderr += b.toString(); });
+            const port = await new Promise<number>((resolve, reject) => {
+                let stdout = "";
+                const timer = setTimeout(() => reject(new Error("restart timeout: " + stderr + stdout)), 15000);
+                child!.stdout!.on("data", (b) => {
+                    stdout += b.toString();
+                    const port = stdout.match(/\{"port":(\d+)\}/);
+                    if (port) { clearTimeout(timer); resolve(Number(port[1])); }
+                });
+                child!.once("exit", (code) => { clearTimeout(timer); reject(new Error(`restart exited ${code}: ${stderr}`)); });
+                child!.once("error", (error) => { clearTimeout(timer); reject(error); });
             });
-            child!.once("exit", (code) => { clearTimeout(timer); reject(new Error(`restart exited ${code}: ${stderr}`)); });
-            child!.once("error", (error) => { clearTimeout(timer); reject(error); });
-        });
-        const call = async <Path extends string>(path: Path, body?: unknown) => {
-            const r = await fetch(`http://127.0.0.1:${port}` + path, body === undefined ? {} : { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body) });
-            return { status: r.status, body: await responseBody(r, path) };
-        };
-        const restored = await call("/__bili/plugin/snapshot?conversationId=child");
-        const restoredStatus = await call("/__bili/plugin/status?conversationId=parent&fallback=latest");
-        assert.equal(restoredStatus.status, 200);
-        assert.equal(restoredStatus.body.sessionId, "parent");
-        assert.equal(restoredStatus.body.fallback, undefined);
-        assert.equal(restoredStatus.body.sessionRevision, snapshot.parentRevision);
-        assert.equal(restoredStatus.body.contextTokensSource, "estimate");
-        assert.equal(restored.status, 200);
-        assert.equal(restored.body.parentRevision, childSnapshot.parentRevision);
-        assert.deepEqual(restored.body.orderedMessages, childSnapshot.orderedMessages);
-        assert.equal((await call("/__bili/plugin/snapshot?conversationId=parent")).body.parentRevision, snapshot.parentRevision);
-        const replay = await call("/__bili/plugin/fork", req);
-        assert.equal(replay.status, 200);
-        assert.equal(replay.body.replayed, true);
-        const retrieved = await call("/__bili/plugin/tool", { conversationId: "child", tool: "acp_retrieve", args: { ref: "m00001" } });
-        assert.equal(retrieved.status, 200);
-        assert.match(retrieved.body.result, /CCR original retained independently after restart/);
-        const original = await call("/__bili/plugin/tool", { conversationId: "child", tool: "decompress", args: { blockId: "b1", full: true } });
-        assert.match(original.body.result, /first original/);
-        assert.match(original.body.result, /second original/);
-        assert(!original.body.result.includes("tail original"));
-    } finally {
-        if (child?.pid && child.exitCode === null && child.signalCode === null) { child.kill("SIGTERM"); await once(child, "exit"); }
-        await h.close();
-    }
-});
+            const call = async <Path extends string>(path: Path, body?: unknown) => {
+                const r = await fetch(`http://127.0.0.1:${port}` + path, body === undefined ? {} : { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body) });
+                return { status: r.status, body: await responseBody(r, path) };
+            };
+            const restored = await call("/__bili/plugin/snapshot?conversationId=child");
+            const restoredStatus = await call("/__bili/plugin/status?conversationId=parent&fallback=latest");
+            assert.equal(restoredStatus.status, 200);
+            assert.equal(restoredStatus.body.sessionId, "parent");
+            assert.equal(restoredStatus.body.fallback, undefined);
+            assert.equal(restoredStatus.body.sessionRevision, snapshot.parentRevision);
+            assert.equal(restoredStatus.body.contextTokensSource, "estimate");
+            assert.equal(restored.status, 200);
+            assert.equal(restored.body.parentRevision, childSnapshot.parentRevision);
+            assert.deepEqual(restored.body.orderedMessages, childSnapshot.orderedMessages);
+            assert.equal((await call("/__bili/plugin/snapshot?conversationId=parent")).body.parentRevision, snapshot.parentRevision);
+            const replay = await call("/__bili/plugin/fork", req);
+            assert.equal(replay.status, 200);
+            assert.equal(replay.body.replayed, true);
+            const retrieved = await call("/__bili/plugin/tool", { conversationId: "child", tool: "acp_retrieve", args: { ref: "m00001" } });
+            assert.equal(retrieved.status, 200);
+            assert.match(retrieved.body.result, /CCR original retained independently after restart/);
+            const original = await call("/__bili/plugin/tool", { conversationId: "child", tool: "decompress", args: { blockId: "b1", full: true } });
+            assert.match(original.body.result, /first original/);
+            assert.match(original.body.result, /second original/);
+            assert(!original.body.result.includes("tail original"));
+        } finally {
+            if (child?.pid && child.exitCode === null && child.signalCode === null) { child.kill("SIGTERM"); await once(child, "exit"); }
+            await h.close();
+            if (previousMode === undefined) delete process.env.BILI_STORAGE_UNIFIED;
+            else process.env.BILI_STORAGE_UNIFIED = previousMode;
+        }
+    });
+}
 
 test("HTTP status separates real usage from a manual-compression estimate and changing generation", { timeout: 60_000 }, async () => {
     const h = await harness();

From 8364915ca0d49f00869e3f62e350bc208f04472f Mon Sep 17 00:00:00 2001
From: z0d0dOr <128966342+z0d0dOr@users.noreply.github.com>
Date: Sun, 25 Oct 2026 12:36:40 -0700
Subject: [PATCH 4/7] test(storage): account for persisted engine lane in
 render goldens

The engineLane=server metadata added by f590da94 explains the entire drift: removing that one field reproduces the old metadata digest. Regeneration changes only metadata and the enclosing digest; all wire bodies, state, originals and statistics remain identical.
---
 tests/golden/storage-render/anthropic.json | 4 ++--
 tests/golden/storage-render/chat.json      | 4 ++--
 tests/golden/storage-render/google.json    | 4 ++--
 tests/golden/storage-render/responses.json | 4 ++--
 4 files changed, 8 insertions(+), 8 deletions(-)

diff --git a/tests/golden/storage-render/anthropic.json b/tests/golden/storage-render/anthropic.json
index 5afa14b2c4..13811dcab0 100644
--- a/tests/golden/storage-render/anthropic.json
+++ b/tests/golden/storage-render/anthropic.json
@@ -80,12 +80,12 @@
  "foldAt": 9,
  "editedAt": 12,
  "summaryFrom": 9,
- "stateDigest": "6727f42bd91879a7aa370e1ce0c78963bcba0df92e50390d4a8011f8f0da334d",
+ "stateDigest": "0a241435d456ba994c92b0072d2173bf2a6d00ee73223458fb4302e170efb1e3",
  "sectionDigests": {
   "state": "f93dc1c876c672d4",
   "blockContents": "dc7d13d250c94bb4",
   "stats": "c6f1d434975d047d",
-  "metadata": "f1ec4aab3aff777a"
+  "metadata": "16cad648ccf41950"
  },
  "stateKeyDigests": {
   "blocks": "3a454b9a16e4f0b9",
diff --git a/tests/golden/storage-render/chat.json b/tests/golden/storage-render/chat.json
index 46764f97db..01f69b1036 100644
--- a/tests/golden/storage-render/chat.json
+++ b/tests/golden/storage-render/chat.json
@@ -80,12 +80,12 @@
  "foldAt": 9,
  "editedAt": 12,
  "summaryFrom": 9,
- "stateDigest": "e149c301f382438ee063a38f70dd794207b88e94bdf3c14c5191da0eeaa99bc9",
+ "stateDigest": "9f1b8258a916fcf63af14625def9ed0236eb8ec9c542f0027206985598d2a5b8",
  "sectionDigests": {
   "state": "8390f21336885863",
   "blockContents": "94a58f96d7c5cfc3",
   "stats": "24d705b0e39bd13c",
-  "metadata": "55733fb81ccf7d69"
+  "metadata": "2198f47054c8acae"
  },
  "stateKeyDigests": {
   "blocks": "d9770feee91dafc6",
diff --git a/tests/golden/storage-render/google.json b/tests/golden/storage-render/google.json
index ba999cadec..fe9f9c95dd 100644
--- a/tests/golden/storage-render/google.json
+++ b/tests/golden/storage-render/google.json
@@ -80,12 +80,12 @@
  "foldAt": 9,
  "editedAt": 12,
  "summaryFrom": 9,
- "stateDigest": "80e83935e5fa7b052a3cb0f1d641bd81cb6ee59f91f3c1f2f31c204de29368a5",
+ "stateDigest": "c4979d4f58cfa49bc8be8c43b8e1f1b0861b3b9d44cdc415ee09e4ca21d9aa5a",
  "sectionDigests": {
   "state": "2e1def3a3234afdc",
   "blockContents": "473264d9df8ba338",
   "stats": "900481685c848462",
-  "metadata": "88cd02a467b4457e"
+  "metadata": "939dc0c09b6cbcbf"
  },
  "stateKeyDigests": {
   "blocks": "0cfd0b9c0fc53073",
diff --git a/tests/golden/storage-render/responses.json b/tests/golden/storage-render/responses.json
index 7004d90435..0402e7d43c 100644
--- a/tests/golden/storage-render/responses.json
+++ b/tests/golden/storage-render/responses.json
@@ -80,12 +80,12 @@
  "foldAt": 9,
  "editedAt": 12,
  "summaryFrom": 9,
- "stateDigest": "08526d3552fca7ac7552cd16a221893a652fbe740c9b8dccac6db57b3d6302ee",
+ "stateDigest": "b3bddfa75ce3f8b94424be9a0164c1b9a183faa9daf705e6175cabdd671c95ff",
  "sectionDigests": {
   "state": "db313e8eba2d65fc",
   "blockContents": "b6701268a0bdcd8a",
   "stats": "327d3967b8371aa8",
-  "metadata": "b5c39d84db67f424"
+  "metadata": "a9eecd3816738158"
  },
  "stateKeyDigests": {
   "blocks": "34b0dc17f8c24595",

From 04c7979d5deb1fe62a585a9f39c8a63b78cbdc83 Mon Sep 17 00:00:00 2001
From: z0d0dOr <128966342+z0d0dOr@users.noreply.github.com>
Date: Sun, 25 Oct 2026 12:44:01 -0700
Subject: [PATCH 5/7] test(storage): account for one-time ledger ref migration
 on restart

Compare unified reads with an independent envelope reader plus the documented migration. Persist alignment before rollback/corrupt-index checks, select legacy explicitly, and assert the recovered raw ID under its ledger ref. Keep checkpoint writes deterministic so sampled ingest cadence cannot change the seed between reader legs. All 28 targeted storage tests pass; no product migration or digest assertion is disabled.
---
 tests/golden-storage-drive.ts         | 25 ++++++++++++---
 tests/storage-engine-proofs.test.ts   | 45 ++++++++++++++++-----------
 tests/storage-orphan-recovery.test.ts |  6 +++-
 3 files changed, 52 insertions(+), 24 deletions(-)

diff --git a/tests/golden-storage-drive.ts b/tests/golden-storage-drive.ts
index a6b14205f8..41a8b2166d 100644
--- a/tests/golden-storage-drive.ts
+++ b/tests/golden-storage-drive.ts
@@ -17,10 +17,10 @@ import os from "node:os";
 import path from "node:path";
 import { once } from "node:events";
 import { canonicalize, sha256 } from "./golden-canonical.ts";
-import { defaultConfig } from "acp-kernel";
+import { defaultConfig, restampRefsToLedger } from "acp-kernel";
 import { startServer } from "../src/server.ts";
 import { SessionStore, _setStoreForTest } from "../src/persist.ts";
-import { flushAllSessions, _resetSessionsForTest } from "../src/session.ts";
+import { flushAllSessions, getSession, _resetSessionsForTest } from "../src/session.ts";
 import type { ProxyOptions } from "../src/config.ts";
 
 /**
@@ -462,6 +462,8 @@ export async function driveWire(wire: Wire, sessionId = `golden-${wire}`, opts:
     dumpStateTo?: string;
     storageUnified?: "shadow" | "full";
     reloadAfterEdit?: "keepFile" | "dropFile";
+    // Independent legacy-read oracle for #4316's intentional cold-load migration.
+    alignLegacyRefsOnReload?: boolean;
 } = {}): Promise<GoldenRecord> {
     __goldenGuard();
     fs.mkdirSync(SCRATCH_ROOT, { recursive: true });
@@ -503,7 +505,12 @@ export async function driveWire(wire: Wire, sessionId = `golden-${wire}`, opts:
         upstream.listen(0, "127.0.0.1");
         await listen(upstream);
         const upstreamPort = (upstream.address() as { port: number }).port;
-        store = new SessionStore({ dir: sessionsDir, debounceMs: 1, enabled: true });
+        // Save at the explicit restart/final checkpoints below. A 1ms timer
+        // makes ingest/verification counts depend on machine speed, so the
+        // three reader legs can migrate refs from different ledger snapshots.
+        // The debounce exceeds the suite's four-minute deadline; it is never
+        // awaited. Delta/sampled-write behavior has storage-watermark tests.
+        store = new SessionStore({ dir: sessionsDir, debounceMs: 300_000, enabled: true });
         _setStoreForTest(store);
         setRegistryForTest({});
         proxy = await startServer(goldenProxyOptions(upstreamPort, 200_000));
@@ -558,8 +565,16 @@ export async function driveWire(wire: Wire, sessionId = `golden-${wire}`, opts:
                     if (fs.statSync(fp).isFile() && (f as string).endsWith(".json") && !fs.existsSync(fp + ".tmp")) fs.rmSync(fp, { force: true });
                 }
             }
-            store = new SessionStore({ dir: sessionsDir, debounceMs: 1, enabled: true });
+            store = new SessionStore({ dir: sessionsDir, debounceMs: 300_000, enabled: true });
             _setStoreForTest(store);
+            if (opts.alignLegacyRefsOnReload) {
+                const loaded = getSession(sessionId);
+                const seqByRaw = store.unifiedSeqByRaw(sessionId);
+                if (!seqByRaw?.size) throw new Error("legacy alignment oracle requires a populated shadow ledger");
+                const outcome = restampRefsToLedger(loaded.state, seqByRaw);
+                if (outcome.superseded.size > 0) loaded.restampSuperseded = outcome.superseded;
+                store.scheduleSave(loaded);
+            }
         }
         // Post-edit growth: 2 trailing turns. Also load-bearing for the
         // STATE digest: the persist layer does not reliably write the final
@@ -660,7 +675,7 @@ export async function driveWire(wire: Wire, sessionId = `golden-${wire}`, opts:
         // orphaned inode and REOPEN the deleted path, resurrecting the tree
         // mid-rmrf (the #1646/#1910 ENOTEMPTY/EBUSY race class). The detach
         // + target-clear below are synchronous, so no interleaving window.
-        // The 1ms-debounced store writer races the same way (recreating
+        // A pending store writer races the same way (recreating
         // sessions/<provider>/ after the final flush).
         try { store?.cancelAll(); } catch { /* no store on early-exit paths */ }
         void closeLogger();
diff --git a/tests/storage-engine-proofs.test.ts b/tests/storage-engine-proofs.test.ts
index 118ca0bde7..782ab1973a 100644
--- a/tests/storage-engine-proofs.test.ts
+++ b/tests/storage-engine-proofs.test.ts
@@ -5,21 +5,19 @@
  * STATE-level equality between the legacy files and the unified index. This
  * suite pins the seams that audit found unproven:
  *
- *   WIRE PARITY      — BILI_STORAGE_UNIFIED=full must be invisible on the
- *                      wire: per-turn outbound digests, fold landmarks and
- *                      the final state digest equal the COMMITTED storage
- *                      render goldens, both with a mid-conversation restart
- *                      and with the legacy file DELETED at the restart (the
- *                      index alone carries the session — the migration's
- *                      end-state promise).
+ *   WIRE PARITY      — full reads match envelope reads plus the documented
+ *                      #4316 ledger-ref migration. All post-restart bodies,
+ *                      landmarks and final digests agree, even when the
+ *                      legacy file is DELETED at restart. Pre-migration
+ *                      bodies also match the committed render goldens.
  *   STALE REJECT     — a multi-instance stale write that the guarded build
  *                      rejects must never reach index.db; the converged disk
  *                      record is what gets ingested.
  *   IMPORT GUARD     — a hand-aged legacy file (older savedAt than the
  *                      index watermark) must not re-import over newer index
  *                      state on load.
- *   ROLLBACK         — unset the env after running full → the legacy file
- *                      loads with the exact digest.
+ *   ROLLBACK         — explicitly select legacy after persisting the full
+ *                      load's migration → the file loads the exact digest.
  *   SELF-DISABLE     — 3 consecutive sidecar failures disable it; reads flip
  *                      back to legacy; the legacy file stays readable.
  *   CORRUPT INDEX    — a corrupt index.db degrades the store to legacy-only
@@ -141,11 +139,11 @@ function unifiedRows(store: UnifiedStore, id: string): ReturnType<typeof loadLeg
 }
 
 // ---------------------------------------------------------------------------
-// WIRE PARITY: full mode + mid-conversation restart == committed goldens.
+// WIRE PARITY: full restart == envelope restart + ledger-ref migration.
 // ---------------------------------------------------------------------------
 
 for (const wire of WIRES) {
-    test(`storage engine: full-mode restart is invisible on the ${wire} wire (== legacy restart, wire == committed goldens)`, { timeout: 240_000 }, async () => {
+    test(`storage engine: full-mode ${wire} restart matches legacy reads plus ledger alignment`, { timeout: 240_000 }, async () => {
         const golden = JSON.parse(fs.readFileSync(path.join(RENDER_ROOT, `${wire}.json`), "utf8")) as {
             turns: Array<{ i: number; sha: string }>;
             foldAt: number; editedAt: number; summaryFrom: number;
@@ -157,13 +155,20 @@ for (const wire of WIRES) {
         // slots (cacheLedger line bs/bl/be pairing, #2131) live in memory, so
         // exactly the first post-restart ledger line loses its bs fields —
         // identically in both legs. Everything else must be byte-equal.
-        const legacy = await driveWire(wire as Wire, `golden-${wire}`, { reloadAfterEdit: "keepFile" });
+        // #4316 deliberately renumbers legacy refs at the first unified cold
+        // load. Keep the independent envelope reader as the oracle, applying
+        // only that documented migration against its shadow ledger. Every
+        // post-restart byte still has to match; the old pre-migration golden
+        // remains the oracle for all turns before the restart.
+        const legacy = await driveWire(wire as Wire, `golden-${wire}`, {
+            storageUnified: "shadow", reloadAfterEdit: "keepFile", alignLegacyRefsOnReload: true,
+        });
         for (const leg of ["keepFile", "dropFile"] as const) {
             const record = await driveWire(wire as Wire, `golden-${wire}`, { storageUnified: "full", reloadAfterEdit: leg });
             assert.equal(record.turns.length, legacy.turns.length, `${wire}/${leg}: body count drifted vs legacy restart`);
             for (let i = 0; i < legacy.turns.length; i++) {
                 assert.equal(record.turns[i]!.sha, legacy.turns[i]!.sha, `${wire}/${leg}: outbound body #${i} digest drift vs legacy restart`);
-                assert.equal(record.turns[i]!.sha, golden.turns[i]!.sha, `${wire}/${leg}: outbound body #${i} digest drift vs committed goldens (restart must not change the wire)`);
+                if (i <= record.editedAt) assert.equal(record.turns[i]!.sha, golden.turns[i]!.sha, `${wire}/${leg}: pre-migration body #${i} digest drift vs committed goldens`);
             }
             assert.equal(record.foldAt, legacy.foldAt, `${wire}/${leg}: fold landmark drifted`);
             assert.equal(record.editedAt, legacy.editedAt, `${wire}/${leg}: edit landmark drifted`);
@@ -274,10 +279,10 @@ test("storage engine: load-side import guard skips a hand-aged legacy file (inde
 });
 
 // ---------------------------------------------------------------------------
-// ROLLBACK: full → unset env → legacy file authoritative, digest exact.
+// ROLLBACK: full → explicit legacy → legacy file authoritative, digest exact.
 // ---------------------------------------------------------------------------
 
-test("storage engine: rollback contract — unset the env and the legacy read path serves the identical state", async () => {
+test("storage engine: rollback contract — explicit legacy mode serves the identical persisted state", async () => {
     const dir = tmpdir("bili-engproof-rollback-");
     const prev = process.env.BILI_STORAGE_UNIFIED;
     try {
@@ -295,11 +300,13 @@ test("storage engine: rollback contract — unset the env and the legacy read pa
         // tail normalization, identically under both paths.)
         const storeFull = new SessionStore({ dir, debounceMs: 1, enabled: true });
         const viaFull = storeFull.loadSync("corpus-folded", { protocol: "anthropic", upstreamOrigin: "http://127.0.0.1:8199" });
-        storeFull.closeQuietly();
         assert.ok(viaFull, "full-mode reload returned null");
+        // Persist the one-time cold-load alignment before exercising rollback.
+        await storeFull.writeNow(viaFull);
+        storeFull.closeQuietly();
         const digestFull = viewDigest(toLegacyLike(viaFull));
 
-        delete process.env.BILI_STORAGE_UNIFIED;
+        process.env.BILI_STORAGE_UNIFIED = "legacy";
         const store2 = new SessionStore({ dir, debounceMs: 1, enabled: true });
         const legacy = store2.loadSync("corpus-folded", { protocol: "anthropic", upstreamOrigin: "http://127.0.0.1:8199" });
         store2.closeQuietly();
@@ -366,8 +373,10 @@ test("storage engine: corrupt index.db degrades to legacy-only instead of bricki
         // fallback must reproduce exactly from the legacy file.
         const clean = new SessionStore({ dir, debounceMs: 1, enabled: true });
         const viaFull = clean.loadSync("corpus-folded", { protocol: "anthropic", upstreamOrigin: "http://127.0.0.1:8199" });
-        clean.closeQuietly();
         assert.ok(viaFull, "full-mode read returned null");
+        // Both replicas must contain the migration before corrupting the index.
+        await clean.writeNow(viaFull);
+        clean.closeQuietly();
 
         // #4338: everything past this point (corrupt the header → degrade
         // construction → fallback read → rename guards → write → close) runs
diff --git a/tests/storage-orphan-recovery.test.ts b/tests/storage-orphan-recovery.test.ts
index c63f3f63fb..cb39fcd6a1 100644
--- a/tests/storage-orphan-recovery.test.ts
+++ b/tests/storage-orphan-recovery.test.ts
@@ -69,7 +69,11 @@ test("cold load recovers the pre-fix orphan freeze before processing a request",
     assert.equal(recovered.stats.requests, 876);
     assert.equal(recovered.state.nextBlockId, 4);
     assert.deepEqual(recovered.state.blocks.map(b => b.blockId), ["b3"]);
-    assert.equal(recovered.state.messageRefs.byRef.m00003, "raw-new");
+    // Recovery is followed by #4316's one-time ledger alignment: raw-new is
+    // the first stored message, so its durable ref is seq 0 + 1.
+    assert.equal(recovered.state.messageRefs.byRaw["raw-new"], "m00001");
+    assert.equal(recovered.state.messageRefs.byRef.m00001, "raw-new");
+    assert.equal(recovered.restampSuperseded?.get("m00003"), "raw-new");
     assert.equal(recovered.lastMessages?.at(-1)?.text, "new turn");
     assert.equal(recovered.metadata.recoveryMarker, "fresh");
     assert.equal(recovered.blockContents.get("b1")?.full.text, "retained original");

From e60c0191620c5da709dbae0e068b554b2b695343 Mon Sep 17 00:00:00 2001
From: z0d0dOr <128966342+z0d0dOr@users.noreply.github.com>
Date: Wed, 28 Oct 2026 03:47:58 -0700
Subject: [PATCH 6/7] test: fix Windows graceful shutdown and fixture teardown

Windows SIGTERM terminates the child instead of running its graceful
shutdown handler. Trigger the existing loopback shutdown route on Windows
so the draining-marker regression exercises the same lifecycle as POSIX.

Wait for child close before deleting its database, destroy sockets owned
by the silent TCP fixture, and stop health polling when the child exits.
Keep both regression scenarios and their behavioral assertions intact.

Validation: the isolated CI-preloaded serial test file passes 6/6 and exits
naturally in 11.58 seconds; npm run typecheck and npm run build pass.
---
 tests/issue4982-draining-twin-ladder.test.ts | 50 +++++++++++++++-----
 1 file changed, 37 insertions(+), 13 deletions(-)

diff --git a/tests/issue4982-draining-twin-ladder.test.ts b/tests/issue4982-draining-twin-ladder.test.ts
index 71ea29a279..d0eb94783e 100644
--- a/tests/issue4982-draining-twin-ladder.test.ts
+++ b/tests/issue4982-draining-twin-ladder.test.ts
@@ -228,21 +228,26 @@ async function waitForStatus(port: number, timeoutMs: number): Promise<boolean>
 
 async function killChild(child: ChildProcess): Promise<void> {
     if (child.exitCode !== null || child.signalCode !== null) return;
-    child.kill("SIGTERM");
-    await Promise.race([
-        new Promise<void>((resolve) => child.once("exit", () => resolve())),
-        new Promise<void>((resolve) => setTimeout(resolve, 10_000)),
-    ]);
-    if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL");
+    await new Promise<void>((resolve) => {
+        const timer = setTimeout(() => child.kill("SIGKILL"), 10_000);
+        child.once("close", () => {
+            clearTimeout(timer);
+            resolve();
+        });
+        child.kill("SIGTERM");
+    });
 }
 
 /** A TCP listener that accepts connections and then says nothing — invisible
  *  to /__bili/health probes (they time out as "no evidence") but still owns
  *  the socket, so binders get EADDRINUSE. Deterministically forces the LATE
  *  EADDRINUSE decision path instead of the pre-bind foreign-owner ladder. */
-function silentTcpListener(port: number): Promise<net.Server> {
+function silentTcpListener(port: number): Promise<{ close(): Promise<void> }> {
     return new Promise((resolve, reject) => {
+        const sockets = new Set<net.Socket>();
         const server = net.createServer((sock) => {
+            sockets.add(sock);
+            sock.once("close", () => sockets.delete(sock));
             // accept and hang — never answer HTTP; drain the probe's bytes so peer
             // EOF is actually delivered (a paused socket buffers them and never sees
             // EOF) and release the connection, or a dead prober pins server.close()
@@ -251,7 +256,12 @@ function silentTcpListener(port: number): Promise<net.Server> {
             sock.on("end", () => sock.destroy());
         });
         server.once("error", reject);
-        server.listen(port, "127.0.0.1", () => resolve(server));
+        server.listen(port, "127.0.0.1", () => resolve({
+            close: () => new Promise<void>((done) => {
+                server.close(() => done());
+                for (const socket of sockets) socket.destroy();
+            }),
+        }));
     });
 }
 
@@ -319,6 +329,9 @@ function spawnBili(sandbox: string, port: number): { child: ChildProcess; out: {
 async function waitUntilLadderedHealthy(stateHome: string, child: ChildProcess, label: string): Promise<number> {
     let port = 0;
     await waitUntil(async () => {
+        if (child.exitCode !== null || child.signalCode !== null) {
+            throw new Error(`${label} exited before becoming healthy`);
+        }
         const m = readMarkers(stateHome).find((e) => e.pid === child.pid);
         if (!m || typeof m.port !== "number") return false;
         port = m.port as number;
@@ -335,7 +348,7 @@ test("#4982 e2e: draining same-lane predecessor + squatter → newcomer ladders
     let p1Ref: ReturnType<typeof spawnBili> | null = null;
     let p2Ref: ReturnType<typeof spawnBili> | null = null;
     let heldRef: net.Socket | null = null;
-    let squatterRef: net.Server | null = null;
+    let squatterRef: Awaited<ReturnType<typeof silentTcpListener>> | null = null;
     try {
         // P1: the serving proxy that will enter graceful shutdown.
         const p1 = spawnBili(sandbox, port);
@@ -353,7 +366,18 @@ test("#4982 e2e: draining same-lane predecessor + squatter → newcomer ladders
         });
         heldRef = held;
 
-        p1.child.kill("SIGTERM");
+        if (process.platform === "win32") {
+            // Windows SIGTERM terminates immediately instead of invoking the
+            // handler. The loopback admin route enters the same graceful path.
+            const response = await fetch(`http://127.0.0.1:${port}/__bili/shutdown`, {
+                method: "POST",
+                signal: AbortSignal.timeout(2_000),
+            });
+            assert.equal(response.status, 202, "P1 must accept the graceful shutdown request");
+            await response.arrayBuffer();
+        } else {
+            p1.child.kill("SIGTERM");
+        }
         const p1MarkerFile = path.join(stateHome, "billion-context", "instances", `${p1Marker.instanceId}.json`);
         await waitUntil(async () => {
             let stamped = false;
@@ -395,7 +419,7 @@ test("#4982 e2e: draining same-lane predecessor + squatter → newcomer ladders
         heldRef?.destroy();
         if (p1Ref) await killChild(p1Ref.child);
         const squatter = squatterRef;
-        if (squatter) await new Promise<void>((resolve) => squatter.close(() => resolve()));
+        if (squatter) await squatter.close();
         fs.rmSync(sandbox, { recursive: true, force: true });
     }
 });
@@ -406,7 +430,7 @@ test("#4982 e2e: unverified registry twin claim (unstamped marker) → ladder, n
     const port = await freePort();
     await assertPortDead(port, { label: "4982 unverified-twin port" });
     let p2Ref: ReturnType<typeof spawnBili> | null = null;
-    let squatterRef: net.Server | null = null;
+    let squatterRef: Awaited<ReturnType<typeof silentTcpListener>> | null = null;
     try {
         // A registry marker claiming `port` with a LIVE pid and full identity
         // match — but NO portReleasedAt stamp (pre-fix build, or a failed
@@ -456,7 +480,7 @@ test("#4982 e2e: unverified registry twin claim (unstamped marker) → ladder, n
     } finally {
         if (p2Ref) await killChild(p2Ref.child);
         const squatter = squatterRef;
-        if (squatter) await new Promise<void>((resolve) => squatter.close(() => resolve()));
+        if (squatter) await squatter.close();
         fs.rmSync(sandbox, { recursive: true, force: true });
     }
 });

From 9b99e4d52926a07f198d05470ebdae6e106b7213 Mon Sep 17 00:00:00 2001
From: z0d0dOr <128966342+z0d0dOr@users.noreply.github.com>
Date: Wed, 28 Oct 2026 06:11:26 -0700
Subject: [PATCH 7/7] test: handle reset health probes in silent TCP fixture

Windows CI can reset an aborted probe connection instead of ending it.
The silent TCP fixture did not handle errors on accepted sockets, so
ECONNRESET became an uncaught test exception after the probe finished.

Destroy reset sockets and add a real resetAndDestroy regression against
the same fixture. Keep the existing port-owner and ladder assertions.

Validation: the new regression reproduces ECONNRESET before the fix;
the isolated CI-preloaded serial test file then passes 7/7 and exits
naturally. Pure tsc and git diff --check pass.
---
 tests/issue4982-draining-twin-ladder.test.ts | 22 ++++++++++++++++++++
 1 file changed, 22 insertions(+)

diff --git a/tests/issue4982-draining-twin-ladder.test.ts b/tests/issue4982-draining-twin-ladder.test.ts
index d0eb94783e..5c8e178cce 100644
--- a/tests/issue4982-draining-twin-ladder.test.ts
+++ b/tests/issue4982-draining-twin-ladder.test.ts
@@ -254,6 +254,9 @@ function silentTcpListener(port: number): Promise<{ close(): Promise<void> }> {
             // with CLOSE-WAIT sockets
             sock.resume();
             sock.on("end", () => sock.destroy());
+            // An aborted health probe may reset instead of ending the connection.
+            // Socket errors belong to this silent fixture, not the enclosing test.
+            sock.on("error", () => sock.destroy());
         });
         server.once("error", reject);
         server.listen(port, "127.0.0.1", () => resolve({
@@ -265,6 +268,25 @@ function silentTcpListener(port: number): Promise<{ close(): Promise<void> }> {
     });
 }
 
+test("silent TCP fixture survives an aborted health probe", { timeout: 5_000 }, async () => {
+    const port = await freePort();
+    const fixture = await silentTcpListener(port);
+    const client = net.connect({ host: "127.0.0.1", port });
+    try {
+        await new Promise<void>((resolve, reject) => {
+            client.once("connect", resolve);
+            client.once("error", reject);
+        });
+        client.write("GET /__bili/health HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n");
+        await new Promise((resolve) => setTimeout(resolve, 25));
+        client.resetAndDestroy();
+        await new Promise((resolve) => setTimeout(resolve, 25));
+    } finally {
+        client.destroy();
+        await fixture.close();
+    }
+});
+
 function tcpRefused(port: number): Promise<boolean> {
     return new Promise((resolve) => {
         const s = net.connect({ port, host: "127.0.0.1" });