[PR] fix: release input-only request data before streaming

src/plugin.ts

@@ -411,10 +411,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
@@ -535,17 +555,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) {

src/server.ts

@@ -4559,20 +4559,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);
     });
 }
 

src/server/handle.ts

@@ -41,7 +41,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, pluginCompressionProxyHeader, pluginConversationHeader, pluginHeadersMatchModel, pluginReportedContextWindow, pluginReportedMaxOutput, pluginReportedModel, pluginRequestAgentHeader, pluginRuntimeInfoFor, pluginRuntimeInfoForConversation, publicForkInputCheck, recordChainVerdict, recordPluginSession, rememberPluginMessages, resolveConversation, runtimeConversationId, takePendingPluginRegister } from "../plugin.js";
+import { consumePluginRegisterFor, pluginAgentHeader, pluginCompressionProxyHeader, 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";
@@ -525,6 +525,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
@@ -551,6 +554,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 = (() => {
@@ -2433,10 +2437,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).
@@ -2838,7 +2842,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
@@ -2853,21 +2857,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 {

src/server/relay.ts

@@ -267,12 +267,14 @@ 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();
     const reply = prepared && !prepared.sidePassthrough ? createHandoffReply(prepared.session, () => !res.destroyed) : undefined;
     // 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);
@@ -282,6 +284,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);
@@ -1470,6 +1473,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