diff --git a/src/plugin.ts b/src/plugin.ts index 24458800d0..bade83b053 100644 --- a/src/plugin.ts +++ b/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; + 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; - 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; + 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 b716992248..beae7efa0b 100644 --- a/src/server.ts +++ b/src/server.ts @@ -4559,20 +4559,48 @@ export function readBody(req: http.IncomingMessage): Promise { 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 8095189a58..54e5a00ca6 100644 --- a/src/server/handle.ts +++ b/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 => { // #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 { diff --git a/src/server/relay.ts b/src/server/relay.ts index 1b7642d5f3..d82fa365c3 100644 --- a/src/server/relay.ts +++ b/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, + releaseRequestInput?: () => void, ): Promise { 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 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 = { 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((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((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() { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((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(); + const completed = deferred(); + const trace: string[] = []; + let releaseCount = 0; + let forwardCompleted = false; + let upstreamCalls = 0; + let streamController: ReadableStreamDefaultController | 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({ 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 | undefined; + try { + if (scenario === "success" || scenario === "refold") { + await Promise.race([ + released.promise, + completed.promise.then(() => assert.fail("forward finished before input release")), + new Promise((_, 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(); + } + }); +}