From 0d60e1b5782e8820e3a2fd620f453de6d09958bf Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 18 Sep 2026 23:46:51 +0900 Subject: [PATCH 1/3] fix(transport): match freeform streaming to its completion grammar and decode SOCKS5 content-codings The bridge decided whether a freeform argument buffer was a wrapper by comparing it with the compact literals {"input":" and {"code":", while the completion path reads the same text with JSON.parse. A wrapper written with spaces or newlines matched no prefix, so it streamed as raw JSON and then completed as the unwrapped body - the disagreement #5047 closed for compact wrappers, reached through a different spelling. The same applies to a Markdown-fenced body, which stripMarkdownCodeFence removes at completion for exec and apply_patch. socks5Fetch assembles the response from a raw socket, so new Response(body, headers) surfaced a coded body unchanged: .json() threw SyntaxError on the gzip magic number. The request now asks for identity unless the caller chose otherwise, gzip and deflate are decoded, and any other coding is refused by name. The keep-alive socket-close tests asserted that close propagation is observable within a fixed 50ms, which failed in the unsharded control lane and passed everywhere else; they now wait for the state the contract promises under a bounded deadline. --- src/bridge/sse.ts | 104 ++++++++++++++++------- src/lib/socks5-fetch.ts | 43 +++++++++- structure/transports/inventory.md | 7 ++ tests/adapters/bridge.test.ts | 55 ++++++++++++ tests/lib/socks5-fetch.test.ts | 136 ++++++++++++++++++++++++++++-- 5 files changed, 306 insertions(+), 39 deletions(-) diff --git a/src/bridge/sse.ts b/src/bridge/sse.ts index e38168cf7ba..a17247b7a07 100644 --- a/src/bridge/sse.ts +++ b/src/bridge/sse.ts @@ -157,40 +157,52 @@ export function bridgeToResponsesSSE( // (`response.custom_tool_call_input.delta` — codex-rs uses it for UI preview only; // the completed custom_tool_call item stays authoritative). Compact `{"input":"...` // buffers get their string value progressively unescaped; anything else streams raw. - const FREEFORM_WRAP_PREFIX = '{"input":"'; - /** `{"key":"` for every wrapper this tool name accepts. Keys are distinct, so order is free. */ - const freeformWrapPrefixes = (toolName: string): string[] => [ - FREEFORM_WRAP_PREFIX, - ...freeformFallbackKeys(toolName).map(key => `{"${key}":"`), - ]; + const JSON_WHITESPACE = new Set([" ", "\t", "\n", "\r"]); + type WrapperOpening = + | { state: "none" } + | { state: "prefix" } + | { state: "open"; valueStart: number }; /** - * The value to stream so far, or `null` to HOLD because nothing can be decided yet. - * - * `input` is decidable from its prefix: `unwrapFreeformToolInput` returns it whenever the - * key is present, whatever else the object carries, so its value can be unescaped - * progressively and never retracted. + * Where the string value of `{"":"` begins, tolerating the insignificant whitespace + * `JSON.parse` accepts. * - * A fallback key is not. It only unwraps when it is the SINGLE string field, and a second - * key can still arrive — so a value emitted early would have to be taken back. That is the - * rewind this holds instead: stream nothing until the object closes, then publish the one - * repaired body. The routed passthrough in `responses-custom-tool-repair.ts` already holds - * any object prefix for the same reason (#5047). + * The earlier form of this compared the buffer against the compact literal `{"key":"`, so a + * wrapper written with spaces or newlines matched no prefix at all, streamed as raw JSON + * deltas and then completed as the unwrapped body. That is the same delta/completion + * disagreement #5047 closed for compact wrappers, reached through a different spelling: + * `unwrapFreeformToolInput` reads the completed text with `JSON.parse`, which does not care + * how the object is laid out, so neither can the streaming side. */ - const freeformPartialInput = (args: string, toolName: string): string | null => { - const prefixes = freeformWrapPrefixes(toolName); - // Still an ambiguous prefix of some wrapper: which wrapper, if any, is not known yet. - if (prefixes.some(prefix => prefix.startsWith(args))) return null; - if (!args.startsWith(FREEFORM_WRAP_PREFIX)) { - if (!prefixes.some(prefix => args.startsWith(prefix))) return args; - // Committed to a fallback wrapper. Undecidable until the object is complete. - try { - JSON.parse(args); - } catch { - return null; + const wrapperOpening = (args: string, key: string): WrapperOpening => { + let index = 0; + for (const token of ["{", `"${key}"`, ":", '"']) { + while (index < args.length && JSON_WHITESPACE.has(args[index]!)) index++; + if (index >= args.length) return { state: "prefix" }; + for (const expected of token) { + if (index >= args.length) return { state: "prefix" }; + if (args[index] !== expected) return { state: "none" }; + index++; } - return unwrapFreeformToolInput(args, toolName); } - const body = args.slice(FREEFORM_WRAP_PREFIX.length); + return { state: "open", valueStart: index }; + }; + /** + * Whether a body could still grow into one complete outer Markdown fence. + * + * `stripMarkdownCodeFence` removes such a fence at completion for exactly the two tools that + * own the grammar, so a fenced body's streamed bytes and its completed input disagree unless + * the stream holds. A buffer that does not open with a fence can never acquire one, so + * ordinary bodies are unaffected; a buffer that does keeps its preview suppressed for the + * whole call, because a closing fence can still be followed by more text that withdraws it. + */ + const mayBecomeFencedBody = (text: string, toolName: string): boolean => { + if (toolName !== "exec" && toolName !== "apply_patch") return false; + const head = text.trimStart(); + if (head === "") return true; + return head.startsWith("```") || "```".startsWith(head); + }; + /** The decoded prefix of a JSON string body, stopping at the first byte it cannot resolve. */ + const decodeJsonStringPrefix = (body: string): string => { let out = ""; for (let i = 0; i < body.length; i++) { const c = body[i]; @@ -211,6 +223,40 @@ export function bridgeToResponsesSSE( } return out; }; + /** + * The value to stream so far, or `null` to HOLD because nothing can be decided yet. + * + * `input` is decidable from its prefix: `unwrapFreeformToolInput` returns it whenever the + * key is present, whatever else the object carries, so its value can be unescaped + * progressively and never retracted. + * + * A fallback key is not. It only unwraps when it is the SINGLE string field, and a second + * key can still arrive — so a value emitted early would have to be taken back. That is the + * rewind this holds instead: stream nothing until the object closes, then publish the one + * repaired body. The routed passthrough in `responses-custom-tool-repair.ts` already holds + * any object prefix for the same reason (#5047). + */ + const freeformPartialInput = (args: string, toolName: string): string | null => { + const openings = ["input", ...freeformFallbackKeys(toolName)] + .map(key => ({ key, opening: wrapperOpening(args, key) })); + const canonical = openings[0]!.opening; + if (canonical.state === "open") { + const decoded = decodeJsonStringPrefix(args.slice(canonical.valueStart)); + return mayBecomeFencedBody(decoded, toolName) ? null : decoded; + } + if (openings.some(entry => entry.opening.state === "open")) { + // Committed to a fallback wrapper. Undecidable until the object is complete. + try { + JSON.parse(args); + } catch { + return null; + } + return unwrapFreeformToolInput(args, toolName); + } + // Still an ambiguous prefix of some wrapper: which wrapper, if any, is not known yet. + if (openings.some(entry => entry.opening.state === "prefix")) return null; + return mayBecomeFencedBody(args, toolName) ? null : args; + }; // tool_search_call carries arguments as a JSON object ({query, limit}); parse the model's arg string. const parseArgsObj = (args: string): Record => { try { const o = JSON.parse(args); return o && typeof o === "object" ? o : {}; } catch { return {}; } diff --git a/src/lib/socks5-fetch.ts b/src/lib/socks5-fetch.ts index ab2e222e2b0..c97e68b9435 100644 --- a/src/lib/socks5-fetch.ts +++ b/src/lib/socks5-fetch.ts @@ -321,6 +321,11 @@ function requestHeaders(request: Request, target: URL): { text: string; chunked: const headers = new Headers(request.headers); if (!headers.has("host")) headers.set("host", target.host); if (!headers.has("connection")) headers.set("connection", "close"); + // This transport assembles the body itself, so a content-coding it did not ask for becomes + // its own problem to undo. Ask for none by default and leave an explicit caller choice + // alone: `fetchWithHeaderTimeout` selects identity for some streaming calls and preserves a + // caller's selection elsewhere, and silently overriding that would make the two disagree. + if (!headers.has("accept-encoding")) headers.set("accept-encoding", "identity"); const chunked = request.body !== null && !headers.has("content-length") && !headers.has("transfer-encoding"); @@ -504,6 +509,28 @@ function responseBody( }); } +/** + * Undo the content-coding this transport has to undo itself. + * + * `fetch` decodes content-codings below the Response constructor. This one assembles the body + * from a raw socket, so `new Response(body, { headers })` hands the compressed bytes straight + * to whatever reads them: `response.json()` throws SyntaxError on the gzip magic number and an + * SSE reader sees noise instead of frames. The request asks for `identity`, so a coded + * response means an upstream ignored that; gzip and deflate are undone here, and any other + * coding fails closed rather than surfacing bytes no caller can parse. + * + * No decompressed-size ceiling is imposed. The identity path has no total-size bound either — + * it cannot, because a long-lived SSE stream is legitimately unbounded — and a ceiling on only + * the coded path would fail responses that succeed uncompressed. + */ +function contentCodingFormat(headers: Headers): "gzip" | "deflate" | undefined { + const coding = (headers.get("content-encoding") ?? "").trim().toLowerCase(); + if (coding === "" || coding === "identity") return undefined; + if (coding === "gzip" || coding === "x-gzip") return "gzip"; + if (coding === "deflate") return "deflate"; + throw new Socks5FetchError("SOCKS5 upstream returned an unsupported content-encoding: " + coding); +} + export async function socks5Fetch( input: RequestInfo | URL, init: RequestInit | undefined, @@ -559,12 +586,24 @@ export async function socks5Fetch( while (responseHead.status >= 100 && responseHead.status < 200 && responseHead.status !== 101) { responseHead = parseResponseHead(await reader.readUntil(HEADER_END, MAX_RESPONSE_HEADER_BYTES, request.signal)); } + // Reject a coding this transport cannot undo before the body stream takes the socket, so + // the failure path is the ordinary one below rather than an orphaned reader. + const codingFormat = contentCodingFormat(responseHead.headers); const body = responseBody(reader, socket, request.signal, responseHead.headers, responseHead.status, request.method); + const responseHeaders = new Headers(responseHead.headers); + if (codingFormat !== undefined) { + responseHeaders.delete("content-encoding"); + // The declared length describes the coded bytes, not what the caller now reads. + responseHeaders.delete("content-length"); + } + const decodedBody = body !== null && codingFormat !== undefined + ? body.pipeThrough(new DecompressionStream(codingFormat)) + : body; request.signal.removeEventListener("abort", onAbort); - return new Response(body, { + return new Response(decodedBody, { status: responseHead.status, statusText: responseHead.statusText, - headers: responseHead.headers, + headers: responseHeaders, }); } catch (error) { request.signal.removeEventListener("abort", onAbort); diff --git a/structure/transports/inventory.md b/structure/transports/inventory.md index 4c19b262fe9..b63b0f23a25 100644 --- a/structure/transports/inventory.md +++ b/structure/transports/inventory.md @@ -240,4 +240,11 @@ fragmented framing, header limits and explicit-route snapshot preservation. Explicit `http2` / `h2` pins reject before network I/O: this HTTP/1.1 tunnel cannot honor them and must not silently downgrade the provider contract. +Content-coding is this transport's own obligation. `fetch` decodes a coded body below the +Response constructor; this tunnel assembles the body from a socket, so a response wrapped with +its upstream headers hands the coded bytes to whatever parses them. The request therefore asks +for `identity` unless the caller chose an `accept-encoding` itself, a `gzip` or `deflate` +response is decoded and stops advertising the coding and the coded length, and any other coding +is refused by name rather than surfaced as bytes no caller can read. + Dashboard Fast-row persistence and client refresh follow the [Fast selector rows setting contract](../gui-and-management-api.md#fast-selector-rows-setting). diff --git a/tests/adapters/bridge.test.ts b/tests/adapters/bridge.test.ts index 8703d52fb5a..d07ec028948 100644 --- a/tests/adapters/bridge.test.ts +++ b/tests/adapters/bridge.test.ts @@ -1803,6 +1803,61 @@ describe("fallback freeform wrappers stream one stable representation (#5047)", .toEqual({ concatenated: "line1\nline2", done: "line1\nline2", itemInput: "line1\nline2" }); }); + // #5047 matched the buffer against the compact literals `{"input":"` and `{"code":"`, so a + // wrapper carrying the insignificant whitespace `JSON.parse` accepts matched nothing at all: + // it streamed as raw JSON and then completed as the unwrapped body. The completion path reads + // the text with `JSON.parse`, which does not care how the object is laid out, so the two + // disagreed again through a different spelling of the same wrapper. + test("a wrapper written with whitespace streams the same value it completes with", async () => { + const spacings = [ + '{ "code": "const x = 1;" }', + '{"code" : "const x = 1;"}', + '{\n "code": "const x = 1;"\n}', + '{\t"input":\t"const x = 1;"}', + '{ "input" : "const x = 1;" }', + ]; + for (const wrapper of spacings) { + const view = inputView(await streamExec([wrapper])); + expect({ wrapper, ...view }) + .toEqual({ wrapper, concatenated: "const x = 1;", done: "const x = 1;", itemInput: "const x = 1;" }); + } + }); + + test("a spaced wrapper split at every byte boundary never leaks raw JSON", async () => { + const wrapper = '{ "code": "a\\nb" }'; + for (let cut = 1; cut < wrapper.length; cut++) { + const view = inputView(await streamExec([wrapper.slice(0, cut), wrapper.slice(cut)])); + expect({ cut, ...view }).toEqual({ cut, concatenated: "a\nb", done: "a\nb", itemInput: "a\nb" }); + } + }); + + // `stripMarkdownCodeFence` removes one complete outer fence at completion for `exec` and + // `apply_patch`. Streaming the fence bytes first and then completing with the stripped body is + // the same rewind the wrapper holds exist to avoid, so a fenced body streams no preview at + // all. A closing fence can still be followed by more text that withdraws it, which is why the + // hold lasts the whole call rather than releasing when the fence looks complete. + test("a fenced body streams no preview rather than bytes the completed item drops", async () => { + const fenced = ["```js", "const x = 1;", "```"].join("\n"); + for (const chunks of [[fenced], [fenced.slice(0, 4), fenced.slice(4)], Array.from(fenced)]) { + const view = inputView(await streamExec(chunks)); + expect(view.done).toBe("const x = 1;"); + expect(view.itemInput).toBe("const x = 1;"); + expect(view.concatenated).toBe(""); + } + + // The same body inside the canonical wrapper: the value is decidable, the fence is not. + const wrapped = inputView(await streamExec([JSON.stringify({ input: fenced })])); + expect(wrapped).toEqual({ concatenated: "", done: "const x = 1;", itemInput: "const x = 1;" }); + }); + + test("a body that only starts like a fence resumes streaming as soon as it cannot be one", async () => { + // One held character, then the buffer can no longer open a fence and streams normally. This + // is what keeps the fence hold from swallowing ordinary template-literal JavaScript. + const body = "`hello` + world"; + const view = inputView(await streamExec(Array.from(body))); + expect(view).toEqual({ concatenated: body, done: body, itemInput: body }); + }); + test("a stream that dies inside a held wrapper manufactures no tool call", async () => { // The held buffer is suppressed output, never content. An aborted turn must not turn it // into a completed call, and must not release it as raw JSON either. diff --git a/tests/lib/socks5-fetch.test.ts b/tests/lib/socks5-fetch.test.ts index ef58ed40171..6299e0cbb8f 100644 --- a/tests/lib/socks5-fetch.test.ts +++ b/tests/lib/socks5-fetch.test.ts @@ -52,6 +52,22 @@ async function close(server: TcpServer | ReturnType): P await new Promise(resolve => server.close(() => resolve())); } +/** + * Wait for a socket to be observably destroyed, under a bounded deadline. + * + * The fixed `Bun.sleep(50)` this replaces asserted that close propagation is observable within + * fifty milliseconds, which is a claim about machine load rather than about the transport. It + * failed in the unsharded macOS control lane, where the whole suite shares one process (#4997), + * while passing in every sharded lane. The contract is that `finish()` destroys the socket, not + * that it does so inside any particular window, so this waits for the state the contract + * promises and fails only when it never arrives. + */ +async function awaitDestroyed(socket: Socket | undefined, timeoutMs = 2_000): Promise { + const deadline = Date.now() + timeoutMs; + while (socket?.destroyed !== true && Date.now() < deadline) await Bun.sleep(5); + return socket?.destroyed === true; +} + function socksProxy(options: { username?: string; password?: string; @@ -167,8 +183,7 @@ describe("socks5Fetch", () => { `socks5://127.0.0.1:${proxyPort}`, ); expect(await response.text()).toBe("ok"); - await Bun.sleep(50); - expect(targetConnection?.destroyed).toBe(true); + expect(await awaitDestroyed(targetConnection)).toBe(true); } finally { targetConnection?.destroy(); await Promise.all([close(proxy), close(target)]); @@ -194,14 +209,121 @@ describe("socks5Fetch", () => { `socks5://127.0.0.1:${proxyPort}`, ); expect(await response.text()).toBe(""); - await Bun.sleep(50); - expect(targetConnection?.destroyed).toBe(true); + expect(await awaitDestroyed(targetConnection)).toBe(true); } finally { targetConnection?.destroy(); await Promise.all([close(proxy), close(target)]); } }); + // A content-coding is undone by `fetch` below the Response constructor. This transport builds + // the body from a raw socket, so before this was handled `new Response(body, { headers })` + // surfaced the compressed bytes unchanged: `.json()` threw SyntaxError on the gzip magic + // number and an SSE reader saw noise. These pin the decode, the headers that stop describing + // the coded bytes, and the refusal to hand over a coding this transport cannot undo. + describe("content-coding", () => { + function codedTarget(coding: string, payload: Uint8Array) { + let requestText = ""; + const server = createTcpServer(socket => { + socket.once("error", () => undefined); + let request = Buffer.alloc(0); + socket.on("data", chunk => { + request = Buffer.concat([request, chunk]); + requestText = request.toString("latin1"); + if (!requestText.includes("\r\n\r\n")) return; + const head = "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-encoding: " + + coding + + "\r\ncontent-length: " + payload.byteLength + "\r\n\r\n"; + socket.write(Buffer.concat([Buffer.from(head, "latin1"), Buffer.from(payload)])); + }); + }); + return { server, requestHead: () => requestText }; + } + + test("a gzip body is decoded and stops advertising a coding it no longer carries", async () => { + const body = JSON.stringify({ ok: true, note: "compressed" }); + const { server: target } = codedTarget("gzip", Bun.gzipSync(new TextEncoder().encode(body))); + const proxy = socksProxy(); + const [targetPort, proxyPort] = await Promise.all([listen(target), listen(proxy)]); + try { + const response = await socks5Fetch( + `http://provider.invalid:${targetPort}/gzip`, + undefined, + `socks5://127.0.0.1:${proxyPort}`, + ); + expect(response.headers.get("content-encoding")).toBeNull(); + // The declared length counted the coded bytes; keeping it would misdescribe the body. + expect(response.headers.get("content-length")).toBeNull(); + expect(await response.json()).toEqual({ ok: true, note: "compressed" }); + } finally { + await Promise.all([close(proxy), close(target)]); + } + }); + + test("a deflate body is decoded the same way", async () => { + const body = JSON.stringify({ ok: true }); + const { server: target } = codedTarget("deflate", Bun.deflateSync(new TextEncoder().encode(body))); + const proxy = socksProxy(); + const [targetPort, proxyPort] = await Promise.all([listen(target), listen(proxy)]); + try { + const response = await socks5Fetch( + `http://provider.invalid:${targetPort}/deflate`, + undefined, + `socks5://127.0.0.1:${proxyPort}`, + ); + expect(await response.json()).toEqual({ ok: true }); + } finally { + await Promise.all([close(proxy), close(target)]); + } + }); + + test("a coding this transport cannot undo fails closed instead of surfacing coded bytes", async () => { + // Brotli is not a format `DecompressionStream` implements. Returning the bytes anyway is + // the behavior being removed: the caller would get a SyntaxError from its own parser with + // nothing naming the cause. + const { server: target } = codedTarget("br", new TextEncoder().encode("not really brotli")); + const proxy = socksProxy(); + const [targetPort, proxyPort] = await Promise.all([listen(target), listen(proxy)]); + try { + await expect(socks5Fetch( + `http://provider.invalid:${targetPort}/brotli`, + undefined, + `socks5://127.0.0.1:${proxyPort}`, + )).rejects.toThrow(/unsupported content-encoding: br/); + } finally { + await Promise.all([close(proxy), close(target)]); + } + }); + + test("the request asks for identity and keeps an explicit caller choice", async () => { + const payload = new TextEncoder().encode("{}"); + const bare = codedTarget("identity", payload); + const chosen = codedTarget("identity", payload); + const proxy = socksProxy(); + const [barePort, chosenPort, proxyPort] = await Promise.all([ + listen(bare.server), + listen(chosen.server), + listen(proxy), + ]); + try { + await socks5Fetch( + `http://provider.invalid:${barePort}/default`, + undefined, + `socks5://127.0.0.1:${proxyPort}`, + ); + expect(bare.requestHead().toLowerCase()).toContain("accept-encoding: identity"); + await socks5Fetch( + `http://provider.invalid:${chosenPort}/explicit`, + { headers: { "accept-encoding": "gzip" } }, + `socks5://127.0.0.1:${proxyPort}`, + ); + expect(chosen.requestHead().toLowerCase()).toContain("accept-encoding: gzip"); + } finally { + await Promise.all([close(proxy), close(bare.server), close(chosen.server)]); + } + }); + }); + test("rejects a request stuck in body backpressure when the socket errors", async () => { let heldSocket: Socket | undefined; const proxy = socksProxy({ holdAfterConnect: true, onHold: socket => { heldSocket = socket; } }); @@ -408,9 +530,8 @@ describe("socks5Fetch", () => { await Bun.sleep(50); await reader.cancel(); await Promise.race([pending.catch(() => undefined), Bun.sleep(250)]); - await Bun.sleep(50); if (!clientSocket) throw new Error("SOCKS5 client socket was not captured"); - expect(clientSocket.destroyed).toBe(true); + expect(await awaitDestroyed(clientSocket)).toBe(true); expect(clientSocket.listenerCount("data")).toBe(0); expect(clientSocket.listenerCount("error")).toBe(0); expect(clientSocket.listenerCount("close")).toBe(0); @@ -463,8 +584,7 @@ describe("socks5Fetch", () => { Bun.sleep(1_000).then(() => "timed out"), ]); expect(outcome).toBeInstanceOf(Error); - await Bun.sleep(50); - expect(clientSocket.destroyed).toBe(true); + expect(await awaitDestroyed(clientSocket)).toBe(true); expect(clientSocket.listenerCount("data")).toBe(0); expect(clientSocket.listenerCount("error")).toBe(0); expect(clientSocket.listenerCount("close")).toBe(0); From 3e848f9afb3160164b6b63ff0a937ca5c67771f7 Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 19 Sep 2026 00:03:28 +0900 Subject: [PATCH 2/3] fix(transport): correct the streaming JSON escape set, hold undecidable prefixes, and keep bodyless SOCKS5 responses out of the decoder Typecheck: DecompressionStream declares its writable side as WritableStream, which TypeScript measures as invariant, so the pair needed an explicit conversion to reach pipeThrough. Review findings: the streaming decoder mapped every unlisted escape to its own suffix, so a valid backslash-b streamed as the letter b; an escape JSON does not define now stops the preview because such a wrapper never parses and completion returns the raw text. A lone high surrogate is held until its pair arrives. A decorated apply_patch envelope is held because normalizeApplyPatchDelimiters rewrites it at completion. A bodyless response no longer has its content-encoding decoded or refused, and the deflate fixture now states zlib framing through node:zlib rather than relying on a runtime default. --- src/bridge/sse.ts | 74 ++++++++++++++++++------ src/lib/socks5-fetch.ts | 31 +++++++--- tests/adapters/bridge.test.ts | 48 ++++++++++++++++ tests/lib/socks5-fetch.test.ts | 100 ++++++++++++++++++++++++++++++++- 4 files changed, 228 insertions(+), 25 deletions(-) diff --git a/src/bridge/sse.ts b/src/bridge/sse.ts index a17247b7a07..cbb7111fb27 100644 --- a/src/bridge/sse.ts +++ b/src/bridge/sse.ts @@ -201,25 +201,54 @@ export function bridgeToResponsesSSE( if (head === "") return true; return head.startsWith("```") || "```".startsWith(head); }; - /** The decoded prefix of a JSON string body, stopping at the first byte it cannot resolve. */ - const decodeJsonStringPrefix = (body: string): string => { + /** The two-character escapes JSON defines, and nothing else. */ + const JSON_ESCAPES = new Map([ + ['"', '"'], ["\\", "\\"], ["/", "/"], + ["b", "\b"], ["f", "\f"], ["n", "\n"], ["r", "\r"], ["t", "\t"], + ]); + const LOW_SURROGATE_ESCAPE = /^\\u[dD][c-fC-F][0-9a-fA-F]{2}$/; + /** + * The decoded prefix of a JSON string body, stopping at the first byte it cannot resolve. + * + * Every stop is a hold rather than a guess, because `JSON.parse` decides the completed value + * and anything invented here would be retracted. Three of them are not obvious: + * + * An escape JSON does not define makes the whole wrapper unparseable no matter what arrives + * next, so completion falls back to the raw text. Returning `null` for that case stops the + * preview rather than continuing to decode a value the completed item will never carry. + * `\\b` and `\\f` are defined, and were previously decoded to the letters b and f. + * + * A lone high surrogate is not a character. Emitting one alone puts an unpaired code unit in a + * delta the client has to decode by itself, so the pair is emitted together or not at all. + */ + const decodeJsonStringPrefix = (body: string): string | null => { let out = ""; for (let i = 0; i < body.length; i++) { const c = body[i]; if (c === '"') break; // unescaped closing quote: value complete - if (c === "\\") { - const n = body[i + 1]; - if (n === undefined) break; // escape split across chunks: wait for more - i++; - if (n === "n") out += "\n"; - else if (n === "t") out += "\t"; - else if (n === "r") out += "\r"; - else if (n === "u") { - const hex = body.slice(i + 1, i + 5); - if (hex.length === 4 && /^[0-9a-fA-F]{4}$/.test(hex)) { out += String.fromCharCode(parseInt(hex, 16)); i += 4; } - else break; // incomplete \uXXXX: wait for more - } else out += n; // \" \\ \/ etc. - } else out += c; + if (c !== "\\") { out += c; continue; } + const n = body[i + 1]; + if (n === undefined) break; // escape split across chunks: wait for more + if (n === "u") { + const hex = body.slice(i + 2, i + 6); + if (hex.length < 4) break; // split across chunks: wait for more + if (!/^[0-9a-fA-F]{4}$/.test(hex)) return null; // never parses + const code = parseInt(hex, 16); + if (code >= 0xd800 && code <= 0xdbff) { + const low = body.slice(i + 6, i + 12); + if (!LOW_SURROGATE_ESCAPE.test(low)) break; + out += String.fromCharCode(code, parseInt(low.slice(2), 16)); + i += 11; + continue; + } + out += String.fromCharCode(code); + i += 5; + continue; + } + const escaped = JSON_ESCAPES.get(n); + if (escaped === undefined) return null; // never parses + out += escaped; + i += 1; } return out; }; @@ -230,6 +259,12 @@ export function bridgeToResponsesSSE( * key is present, whatever else the object carries, so its value can be unescaped * progressively and never retracted. * + * One case escapes that claim and is accepted rather than fixed: a duplicate `input` key. + * `JSON.parse` keeps the last one, so `{"input":"a","input":"b"}` streams a and completes with + * b. Closing it means holding every canonical wrapper until its object parses, which is the + * progressive streaming this path exists to provide. The completed item stays authoritative, + * and no model emits a duplicate key in practice. + * * A fallback key is not. It only unwraps when it is the SINGLE string field, and a second * key can still arrive — so a value emitted early would have to be taken back. That is the * rewind this holds instead: stream nothing until the object closes, then publish the one @@ -242,6 +277,7 @@ export function bridgeToResponsesSSE( const canonical = openings[0]!.opening; if (canonical.state === "open") { const decoded = decodeJsonStringPrefix(args.slice(canonical.valueStart)); + if (decoded === null) return null; return mayBecomeFencedBody(decoded, toolName) ? null : decoded; } if (openings.some(entry => entry.opening.state === "open")) { @@ -1178,7 +1214,13 @@ export function bridgeToResponsesSSE( const mayCompile = declaresCodeModeExec(options?.declaredToolNames) && !currentToolCall.namespace && currentToolCall.name === "exec"; - if (!(mayCompile && mayBecomePatchEnvelope(full)) && full.startsWith(emitted) && full.length > emitted.length) { + // `apply_patch` holds for a different reason with the same shape: + // `normalizeApplyPatchDelimiters` rewrites a decorated `*** Begin Patch ***` + // envelope at completion, so streaming the decorated markers would be + // replaced by the normalized ones. + const mayNormalize = ownsFreeformGrammar && currentToolCall.name === "apply_patch"; + if (!((mayCompile || mayNormalize) && mayBecomePatchEnvelope(full)) + && full.startsWith(emitted) && full.length > emitted.length) { emit("response.custom_tool_call_input.delta", { item_id: currentToolCall.itemId, output_index: currentToolCall.outputIndex, delta: full.slice(emitted.length), diff --git a/src/lib/socks5-fetch.ts b/src/lib/socks5-fetch.ts index c97e68b9435..725d0efb0e6 100644 --- a/src/lib/socks5-fetch.ts +++ b/src/lib/socks5-fetch.ts @@ -429,8 +429,7 @@ function responseBody( status: number, method: string, ): ReadableStream | null { - const bodyless = method === "HEAD" || status === 204 || status === 304 || (status >= 100 && status < 200); - if (bodyless) { + if (bodylessResponse(method, status)) { reader.dispose(); socket.setTimeout(0); socket.destroy(); @@ -509,6 +508,11 @@ function responseBody( }); } +/** Statuses and methods that carry no response body, whatever the headers say about one. */ +function bodylessResponse(method: string, status: number): boolean { + return method === "HEAD" || status === 204 || status === 304 || (status >= 100 && status < 200); +} + /** * Undo the content-coding this transport has to undo itself. * @@ -586,9 +590,14 @@ export async function socks5Fetch( while (responseHead.status >= 100 && responseHead.status < 200 && responseHead.status !== 101) { responseHead = parseResponseHead(await reader.readUntil(HEADER_END, MAX_RESPONSE_HEADER_BYTES, request.signal)); } - // Reject a coding this transport cannot undo before the body stream takes the socket, so - // the failure path is the ordinary one below rather than an orphaned reader. - const codingFormat = contentCodingFormat(responseHead.headers); + // A bodyless response has no coded bytes to undo, so its `content-encoding` describes the + // representation it would have sent and must neither be decoded nor refused — a HEAD whose + // peer advertises brotli is a correct answer, not an unreadable body. For everything else, + // reject a coding this transport cannot undo before the body stream takes the socket, so the + // failure path is the ordinary one below rather than an orphaned reader. + const codingFormat = bodylessResponse(request.method, responseHead.status) + ? undefined + : contentCodingFormat(responseHead.headers); const body = responseBody(reader, socket, request.signal, responseHead.headers, responseHead.status, request.method); const responseHeaders = new Headers(responseHead.headers); if (codingFormat !== undefined) { @@ -596,8 +605,16 @@ export async function socks5Fetch( // The declared length describes the coded bytes, not what the caller now reads. responseHeaders.delete("content-length"); } - const decodedBody = body !== null && codingFormat !== undefined - ? body.pipeThrough(new DecompressionStream(codingFormat)) + // `DecompressionStream` declares its writable side as `WritableStream`, and + // TypeScript measures `WritableStream` as invariant in its chunk type, so the pair is not + // assignable to `ReadableWritablePair` even though every chunk this + // body produces is a valid `BufferSource`. The conversion states that relationship and + // nothing else; it does not widen what is actually written. + const decompressor = codingFormat === undefined + ? undefined + : new DecompressionStream(codingFormat) as unknown as ReadableWritablePair; + const decodedBody = body !== null && decompressor !== undefined + ? body.pipeThrough(decompressor) : body; request.signal.removeEventListener("abort", onAbort); return new Response(decodedBody, { diff --git a/tests/adapters/bridge.test.ts b/tests/adapters/bridge.test.ts index d07ec028948..a5e657d1110 100644 --- a/tests/adapters/bridge.test.ts +++ b/tests/adapters/bridge.test.ts @@ -1858,6 +1858,54 @@ describe("fallback freeform wrappers stream one stable representation (#5047)", expect(view).toEqual({ concatenated: body, done: body, itemInput: body }); }); + test("every escape JSON defines decodes to the character JSON.parse produces", async () => { + // The decoder used to treat any escape it did not list as its own literal suffix, so a valid + // backspace streamed as the letter b while the completed item carried U+0008. The routed + // passthrough already pins this shape in responses-custom-tool-repair.test.ts. + const body = 'a"b\\c/d\be\ff\ng\rh\ti'; + const view = inputView(await streamExec([JSON.stringify({ input: body })])); + expect(view).toEqual({ concatenated: body, done: body, itemInput: body }); + }); + + test("an escape JSON does not define holds instead of inventing a value", async () => { + // A wrapper carrying an undefined escape does not parse, so the completed item is the raw + // text. Decoding the escape to its own suffix would stream a value nothing else ever carries. + const wrapper = '{"input":"a\\qb"}'; + const view = inputView(await streamExec([wrapper])); + expect(view).toEqual({ concatenated: "", done: wrapper, itemInput: wrapper }); + }); + + test("a surrogate pair split across chunks is emitted whole, never as a lone half", async () => { + const emoji = "\u{1F600}"; + const wrapper = '{"input":"x\\ud83d\\ude00y"}'; + for (let cut = 1; cut < wrapper.length; cut++) { + const deltas = (await streamExec([wrapper.slice(0, cut), wrapper.slice(cut)])) + .filter(f => f.event === "response.custom_tool_call_input.delta") + .map(f => String(f.data.delta)); + for (const delta of deltas) { + // A delta ending on an unpaired high surrogate is not decodable on its own. + const last = delta.charCodeAt(delta.length - 1); + expect(last >= 0xd800 && last <= 0xdbff).toBe(false); + } + expect(deltas.join("")).toBe(`x${emoji}y`); + } + }); + + test("a decorated apply_patch envelope streams nothing rather than markers completion rewrites", async () => { + // normalizeApplyPatchDelimiters strips the trailing *** from the outer lines of a complete + // envelope. Streaming the decorated markers first and completing with the normalized ones is + // the same rewind the wrapper and fence holds exist to avoid. + const decorated = ["*** Begin Patch ***", "*** Add File: a.txt", "+x", "*** End Patch ***"].join("\n"); + const normalized = ["*** Begin Patch", "*** Add File: a.txt", "+x", "*** End Patch"].join("\n"); + const frames = await collectSse(bridgeToResponsesSSE(replay([ + { type: "tool_call_start", id: "c1", name: "apply_patch" } as AdapterEvent, + { type: "tool_call_delta", arguments: decorated } as AdapterEvent, + { type: "tool_call_end" } as AdapterEvent, + { type: "done" } as AdapterEvent, + ]), "model", undefined, new Set(["apply_patch"]))); + expect(inputView(frames)).toEqual({ concatenated: "", done: normalized, itemInput: normalized }); + }); + test("a stream that dies inside a held wrapper manufactures no tool call", async () => { // The held buffer is suppressed output, never content. An aborted turn must not turn it // into a completed call, and must not release it as raw JSON either. diff --git a/tests/lib/socks5-fetch.test.ts b/tests/lib/socks5-fetch.test.ts index 6299e0cbb8f..ec6246f7c14 100644 --- a/tests/lib/socks5-fetch.test.ts +++ b/tests/lib/socks5-fetch.test.ts @@ -2,6 +2,7 @@ import { afterEach, describe, expect, test } from "bun:test"; import { createServer as createHttpServer } from "node:http"; import net, { createConnection, createServer as createTcpServer, Socket, type Server as TcpServer } from "node:net"; import type { AddressInfo } from "node:net"; +import { deflateSync, gzipSync } from "node:zlib"; import { configuredOutboundFetch, effectiveProxyFor, configureSocks5Fetch } from "../../src/lib/proxy-env"; import { providerOutboundGet } from "../../src/lib/provider-outbound"; import { socks5Fetch } from "../../src/lib/socks5-fetch"; @@ -242,7 +243,7 @@ describe("socks5Fetch", () => { test("a gzip body is decoded and stops advertising a coding it no longer carries", async () => { const body = JSON.stringify({ ok: true, note: "compressed" }); - const { server: target } = codedTarget("gzip", Bun.gzipSync(new TextEncoder().encode(body))); + const { server: target } = codedTarget("gzip", gzipSync(Buffer.from(body, "utf8"))); const proxy = socksProxy(); const [targetPort, proxyPort] = await Promise.all([listen(target), listen(proxy)]); try { @@ -262,7 +263,10 @@ describe("socks5Fetch", () => { test("a deflate body is decoded the same way", async () => { const body = JSON.stringify({ ok: true }); - const { server: target } = codedTarget("deflate", Bun.deflateSync(new TextEncoder().encode(body))); + // HTTP `deflate` is the zlib container, not raw DEFLATE, and that is what + // `DecompressionStream("deflate")` reads. `node:zlib` states the framing explicitly rather + // than leaving it to a runtime default. + const { server: target } = codedTarget("deflate", deflateSync(Buffer.from(body, "utf8"))); const proxy = socksProxy(); const [targetPort, proxyPort] = await Promise.all([listen(target), listen(proxy)]); try { @@ -322,6 +326,98 @@ describe("socks5Fetch", () => { await Promise.all([close(proxy), close(bare.server), close(chosen.server)]); } }); + + test("a bodyless response keeps its representation headers and is never decoded", async () => { + // A 304 describes the representation it is not sending. Refusing it for naming a coding + // this transport cannot decode would reject a correct answer that carries no coded bytes. + const target = createTcpServer(socket => { + socket.once("error", () => undefined); + socket.on("data", chunk => { + if (!chunk.toString("latin1").includes("\r\n\r\n")) return; + socket.write("HTTP/1.1 304 Not Modified\r\ncontent-encoding: br\r\n\r\n"); + }); + }); + const proxy = socksProxy(); + const [targetPort, proxyPort] = await Promise.all([listen(target), listen(proxy)]); + try { + const response = await socks5Fetch( + `http://provider.invalid:${targetPort}/not-modified`, + undefined, + `socks5://127.0.0.1:${proxyPort}`, + ); + expect(response.status).toBe(304); + expect(response.body).toBeNull(); + expect(response.headers.get("content-encoding")).toBe("br"); + } finally { + await Promise.all([close(proxy), close(target)]); + } + }); + + test("a corrupt coded body errors the reader and still releases the socket", async () => { + // The decode runs in a transform between the socket stream and the caller. If an error + // there did not travel back through the pipe, the source would never cancel and the socket + // would outlive the request. + let targetConnection: Socket | undefined; + const garbage = Buffer.from([0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03, 0x41, 0x42, 0x43]); + const target = createTcpServer(socket => { + targetConnection = socket; + socket.once("error", () => undefined); + socket.on("data", chunk => { + if (!chunk.toString("latin1").includes("\r\n\r\n")) return; + const head = "HTTP/1.1 200 OK\r\ncontent-encoding: gzip\r\ncontent-length: " + + garbage.byteLength + "\r\nconnection: keep-alive\r\n\r\n"; + socket.write(Buffer.concat([Buffer.from(head, "latin1"), garbage])); + }); + }); + const proxy = socksProxy(); + const [targetPort, proxyPort] = await Promise.all([listen(target), listen(proxy)]); + try { + const response = await socks5Fetch( + `http://provider.invalid:${targetPort}/corrupt-gzip`, + undefined, + `socks5://127.0.0.1:${proxyPort}`, + ); + await expect(response.text()).rejects.toThrow(); + expect(await awaitDestroyed(targetConnection)).toBe(true); + } finally { + targetConnection?.destroy(); + await Promise.all([close(proxy), close(target)]); + } + }); + + test("cancelling a coded body releases the socket without waiting for the coding to finish", async () => { + let targetConnection: Socket | undefined; + const payload = gzipSync(Buffer.from("x".repeat(64 * 1024), "utf8")); + const target = createTcpServer(socket => { + targetConnection = socket; + socket.once("error", () => undefined); + socket.on("data", chunk => { + if (!chunk.toString("latin1").includes("\r\n\r\n")) return; + // Declare more than is ever written: the body stays open until the caller cancels. + const head = "HTTP/1.1 200 OK\r\ncontent-encoding: gzip\r\ncontent-length: " + + (payload.byteLength + 1024) + "\r\nconnection: keep-alive\r\n\r\n"; + socket.write(Buffer.concat([Buffer.from(head, "latin1"), payload.subarray(0, 32)])); + }); + }); + const proxy = socksProxy(); + const [targetPort, proxyPort] = await Promise.all([listen(target), listen(proxy)]); + try { + const response = await socks5Fetch( + `http://provider.invalid:${targetPort}/cancel-gzip`, + undefined, + `socks5://127.0.0.1:${proxyPort}`, + ); + const reader = response.body?.getReader(); + if (!reader) throw new Error("coded response had no body"); + const pending = reader.read(); + await reader.cancel(); + await Promise.race([pending.catch(() => undefined), Bun.sleep(250)]); + expect(await awaitDestroyed(targetConnection)).toBe(true); + } finally { + targetConnection?.destroy(); + await Promise.all([close(proxy), close(target)]); + } + }); }); test("rejects a request stuck in body backpressure when the socket errors", async () => { From 558d0080cb72e778bde5b69166fc2bf12bb6e709 Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 19 Sep 2026 00:22:04 +0900 Subject: [PATCH 3/3] test(bridge): pin what a wrapper that turns invalid mid-stream is allowed to do A valid prefix is already published when an undefined escape arrives, and no mechanism takes a delta back. The guarantee is that nothing further is invented: the preview stops at the last decodable character and the completed item carries the raw text. Holding the whole wrapper instead would mean never streaming a canonical wrapper progressively, which the neighbouring test requires. --- src/bridge/sse.ts | 7 +++++++ tests/adapters/bridge.test.ts | 12 ++++++++++++ 2 files changed, 19 insertions(+) diff --git a/src/bridge/sse.ts b/src/bridge/sse.ts index cbb7111fb27..66d15f58fce 100644 --- a/src/bridge/sse.ts +++ b/src/bridge/sse.ts @@ -265,6 +265,13 @@ export function bridgeToResponsesSSE( * progressive streaming this path exists to provide. The completed item stays authoritative, * and no model emits a duplicate key in practice. * + * A wrapper that turns invalid AFTER streaming has committed has the same shape and the same + * answer. `{"input":"a` followed by `\\qb"}` has already published a when the undefined escape + * arrives, and completion returns the raw text because nothing parses. The preview stops + * there: no rewind, and no decoded text the completed item does not contain. Bounding the + * damage is what is available without giving up progressive streaming, and the args are + * unusable in that case whichever representation wins. + * * A fallback key is not. It only unwraps when it is the SINGLE string field, and a second * key can still arrive — so a value emitted early would have to be taken back. That is the * rewind this holds instead: stream nothing until the object closes, then publish the one diff --git a/tests/adapters/bridge.test.ts b/tests/adapters/bridge.test.ts index a5e657d1110..9b51a25c5a1 100644 --- a/tests/adapters/bridge.test.ts +++ b/tests/adapters/bridge.test.ts @@ -1875,6 +1875,18 @@ describe("fallback freeform wrappers stream one stable representation (#5047)", expect(view).toEqual({ concatenated: "", done: wrapper, itemInput: wrapper }); }); + test("a wrapper that turns invalid after streaming committed stops rather than continuing", async () => { + // The valid prefix has already been published when the undefined escape arrives, and no + // mechanism can take a delta back. What is guaranteed is that nothing further is invented: + // the preview stops at the last decodable character and the completed item carries the raw + // text. Holding the whole wrapper instead would mean never streaming any canonical wrapper + // progressively, which the test above this one requires. + const view = inputView(await streamExec(['{"input":"a', '\\qb"}'])); + expect(view.concatenated).toBe("a"); + expect(view.done).toBe('{"input":"a\\qb"}'); + expect(view.itemInput).toBe('{"input":"a\\qb"}'); + }); + test("a surrogate pair split across chunks is emitted whole, never as a lone half", async () => { const emoji = "\u{1F600}"; const wrapper = '{"input":"x\\ud83d\\ude00y"}';