From 9603e42c668acbb2b5fe314e5dcbcb98d35f851b Mon Sep 17 00:00:00 2001 From: ranxianglei Date: Sun, 6 Sep 2026 19:32:16 +0800 Subject: [PATCH 1/3] fix: compat.roles wire role rewrite for role-rejecting upstreams (#552) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New opt-in compat.roles map (global + per-provider, provider wins per key) rewrites message roles on the FINAL forwarded openai/responses body at the forward boundary — covering client-sent roles and bili's own injected developer-role prompt alike — and the compress-retry loops apply the same transform to every re-sent body via RequestOptions.wireTransform. Default {} forwards byte-for-byte unchanged; no kernel changes (kernel already normalizes developer internally, so session/ACP state is unaffected). --- CONFIGURATION.md | 14 ++ CONFIGURATION.zh-CN.md | 14 ++ README.md | 17 +++ src/compat-roles.ts | 88 +++++++++++ src/compress-loop-responses.ts | 5 +- src/config.ts | 22 ++- src/loop/core.ts | 7 +- src/server.ts | 68 +++++++-- tests/compat-roles.test.ts | 263 +++++++++++++++++++++++++++++++++ 9 files changed, 485 insertions(+), 13 deletions(-) create mode 100644 src/compat-roles.ts create mode 100644 tests/compat-roles.test.ts diff --git a/CONFIGURATION.md b/CONFIGURATION.md index e1a6db7fc..9ed14d1b7 100644 --- a/CONFIGURATION.md +++ b/CONFIGURATION.md @@ -107,6 +107,13 @@ Top-level keys that control how the proxy listens and behaves globally. - **Status:** ACTIVE - **Description:** Forward every request to the upstream **without** compression, tool injection, or nudging. Equivalent to `ACP_PASSTHROUGH=1`. Handy for A/B comparison against the uncompressed baseline. +### `compat` + +- **Type:** `{ roles?: Record }` +- **Default:** `{}` (disabled) +- **Status:** ACTIVE +- **Description:** Global wire-compat role map. `roles` maps message roles to the role name your upstream accepts, e.g. `{"compat":{"roles":{"developer":"system"}}}` rewrites `developer` → `system` on the final forwarded body for upstreams that reject the `developer` role (#552, newer codex clients). Applies to `openai` chat-completions and `responses` requests; exact-match roles only, everything else in the body is untouched; re-sent compress-retry bodies carry the same rewrite. Per-provider `compat.roles` entries (see [Providers](#providers)) win per key. Default `{}` forwards bodies byte-for-byte unchanged. + ### `proxy` - **Type:** `string` @@ -171,6 +178,13 @@ A shallow key (`https://open.bigmodel.cn`) matches every path on that host. A de - **Status:** ACTIVE - **Description:** Per-provider compression overrides. This is **level 2 of 3** in the merge hierarchy — see [Compression Tuning](#compression-tuning). +### `compat` + +- **Type:** `{ roles?: Record }` +- **Default:** `{}` (disabled) +- **Status:** ACTIVE +- **Description:** Per-provider wire-compat overrides. `roles` maps message roles to the role name this upstream accepts, e.g. `{"developer": "system"}` for upstreams that reject the `developer` role newer codex clients send (#552). Applied to the final forwarded `openai`/`responses` body — client-sent roles and bili's own injected prompt alike — and to every body the compress-retry loops re-send. Wins per key over the global `compat` block (see [Server Settings](#server-settings)). Default `{}` forwards byte-for-byte unchanged. + --- ## Compression Tuning diff --git a/CONFIGURATION.zh-CN.md b/CONFIGURATION.zh-CN.md index f322bda1c..20740bb0b 100644 --- a/CONFIGURATION.zh-CN.md +++ b/CONFIGURATION.zh-CN.md @@ -107,6 +107,13 @@ - **状态：** ACTIVE - **说明：** 将每个请求**不经过**压缩、工具注入或 nudge，直接转发到上游。等价于 `ACP_PASSTHROUGH=1`。便于与未压缩基线做 A/B 对比。 +### `compat` + +- **类型：** `{ roles?: Record }` +- **默认值：** `{}`（禁用） +- **状态：** ACTIVE +- **说明：** 全局线上兼容角色映射。`roles` 把消息角色映射为上游接受的角色名，例如 `{"compat":{"roles":{"developer":"system"}}}` 把 `developer` → `system`，用于拒绝 `developer` 角色的上游（#552，新版 codex 客户端会发送）。作用于 `openai` chat-completions 与 `responses` 请求；仅精确匹配角色，体内其它内容不动；压缩重试重发的请求体同样携带。按 provider 的 `compat.roles`（见 [Providers](#providers)）按键优先。默认 `{}` 逐字节透明转发。 + ### `proxy` - **类型：** `string` @@ -169,6 +176,13 @@ - **状态：** ACTIVE - **说明：** 按 provider 的压缩覆盖项。这是三层合并中的**第 2 层** —— 见[压缩调优](#压缩调优)。 +### `compat` + +- **类型：** `{ roles?: Record }` +- **默认值：** `{}`（禁用） +- **状态：** ACTIVE +- **说明：** 按 provider 的线上兼容覆盖。`roles` 把消息角色映射为该上游接受的角色名，例如 `{"developer": "system"}` —— 用于拒绝 `developer` 角色的上游（#552，新版 codex 客户端会发这个角色）。作用于最终转发的 `openai`/`responses` 请求体 —— 客户端发送的角色和 bili 自己注入的提示一视同仁 —— 压缩重试循环重发的请求体同样携带该改写。按键覆盖全局 `compat` 块（见[服务端设置](#服务端设置)）。默认 `{}` 逐字节透明转发。 + --- ## 压缩调优 diff --git a/README.md b/README.md index 6b4913dbd..f2f42013e 100644 --- a/README.md +++ b/README.md @@ -355,6 +355,23 @@ So you can give ZCode its own proxy without affecting API-key clients: } ``` +### Wire-compat role rewrite (`compat.roles`) + +Some upstreams reject the `developer` role newer codex clients send on the +Responses API (`400 Invalid role: developer`). `compat.roles` maps roles to +what the upstream accepts — applied at the forward boundary to the final +`openai`/`responses` body (client-sent roles **and** bili's own injected +prompt alike), global or per-provider, default off = byte-for-byte: + +```jsonc +{ + "compat": { "roles": { "developer": "system" } }, + "providers": { + "https://picky.example.com": { "compat": { "roles": { "developer": "user" } } } + } +} +``` + ## How sessions work The proxy needs a stable per-conversation identifier to isolate compression diff --git a/src/compat-roles.ts b/src/compat-roles.ts new file mode 100644 index 000000000..27c0d1c08 --- /dev/null +++ b/src/compat-roles.ts @@ -0,0 +1,88 @@ +import type { ProviderRoutes } from "./config.js"; +import { findRoute } from "./config.js"; + +/** A resolved wire-compat role map: source role → role name upstream accepts. */ +export type CompatRoles = Record; + +/** Validate a `compat.roles`-shaped value: an object of string → string. + * Non-string entries are dropped (a malformed partial never breaks the + * proxy); returns undefined when there is nothing usable. */ +export function parseCompatRoles(v: unknown): Record | undefined { + if (!v || typeof v !== "object" || Array.isArray(v)) return undefined; + const obj = v as Record; + let out: Record | undefined; + for (const [k, val] of Object.entries(obj)) { + if (typeof val !== "string" || val.length === 0) continue; + out ??= {}; + out[k] = val; + } + return out; +} + +/** Merge the wire-compat role map for one request: global `compat.roles` + * (config root) overlaid by the per-provider route entry (longest-URL-prefix + * match, identical to the compress-settings lookup). Provider entries win + * per key; global keys not overridden still apply. Empty when unconfigured — + * the default, byte-for-byte transparent forward (#552). */ +export function resolveCompatRoles( + routes: ProviderRoutes, + upstreamUrl: string | undefined, + globalRoles: Record | undefined, +): Record { + const providerRoles = findRoute(routes, upstreamUrl)?.compat?.roles; + if (!globalRoles && !providerRoles) return {}; + return { ...globalRoles, ...providerRoles }; +} + +/** Apply the role map to a serialized chat-completions or Responses body. + * Rewrites exact-match roles only — position, content and every other field + * are untouched. Returns the original string (no re-stringify) when nothing + * matched, so the default path stays byte-identical. + * + * Applied at the FINAL forward boundary on purpose: every emission site — + * client-sent items, bili's injected compress prompt, instructions hoisting, + * compress-loop items — is visible there, and the kernel already normalizes + * developer→system internally (session state / block IDs / prefix-cache are + * computed from pre-wire core messages), so only outbound bytes change. */ +/** Object-level role rewrite shared by the forward boundary (string body) + * and the compress-retry loops (parsed body). Mutates `parsed` in place; + * returns the number of roles rewritten. */ +export function applyCompatRolesJson( + parsed: Record, + protocol: "openai" | "responses", + roles: Record, +): number { + if (Object.keys(roles).length === 0) return 0; + const items = protocol === "openai" ? parsed.messages : parsed.input; + if (!Array.isArray(items)) return 0; + let rewritten = 0; + for (const item of items) { + if (!item || typeof item !== "object" || Array.isArray(item)) continue; + const msg = item as Record; + if (protocol === "responses" && msg.type !== undefined && msg.type !== "message") continue; + const role = msg.role; + if (typeof role !== "string") continue; + const mapped = roles[role]; + if (mapped === undefined || mapped === role) continue; + msg.role = mapped; + rewritten++; + } + return rewritten; +} + +export function applyCompatRoles( + body: string, + protocol: "openai" | "responses", + roles: Record, +): { body: string; rewritten: number } { + if (Object.keys(roles).length === 0) return { body, rewritten: 0 }; + let parsed: Record; + try { + parsed = JSON.parse(body) as Record; + } catch { + return { body, rewritten: 0 }; + } + const rewritten = applyCompatRolesJson(parsed, protocol, roles); + if (rewritten === 0) return { body, rewritten: 0 }; + return { body: JSON.stringify(parsed), rewritten }; +} diff --git a/src/compress-loop-responses.ts b/src/compress-loop-responses.ts index 93db2f923..d3416f4aa 100644 --- a/src/compress-loop-responses.ts +++ b/src/compress-loop-responses.ts @@ -67,6 +67,9 @@ interface CompressLoopResponsesCtx { interface RequestOptions { url: string; headers: Record; + /** Final-stage wire transform (compat.roles, #552) — same contract as the + * unified loop's RequestOptions.wireTransform. */ + wireTransform?: (body: Record) => Record; } interface FunctionCallAccumulator { @@ -215,7 +218,7 @@ export async function compressLoopResponsesJson( const result = await fetchWithRetry(requestOptions.url, { method: "POST", headers: requestOptions.headers, - body: JSON.stringify(requestBody), + body: JSON.stringify(requestOptions.wireTransform ? requestOptions.wireTransform(requestBody) : requestBody), ...(ctx.proxyUrl ? { dispatcher: proxyDispatcher(ctx.proxyUrl) } : {}), }, undefined, undefined, (info) => { // #189: correlate the rejection with the rewrite that preceded it. diff --git a/src/config.ts b/src/config.ts index 93e2bb780..dc9a178a2 100644 --- a/src/config.ts +++ b/src/config.ts @@ -5,6 +5,8 @@ import { configFile } from "./paths.js"; import { log as loggerLog } from "./logger.js"; import { validateHttpProxy, type ProxyFallbackOptions } from "./upstream-proxy.js"; +import { parseCompatRoles } from "./compat-roles.js"; + export function safeReadJson(path: string): unknown { try { // Strip a leading UTF-8 BOM: Windows Notepad saves UTF-8 "with BOM", @@ -41,6 +43,12 @@ export type ProviderRoute = { compressProtocol?: "tools" | "marker"; /** Per-provider compression overrides (level 2 of 3). See CompressSettings. */ compress?: CompressSettings; + /** Per-provider wire-compat overrides. `roles` maps message roles to the + * role name this upstream accepts (e.g. `"developer": "system"`) — + * applied at the forward boundary to the FINAL wire body, covering + * client-sent roles and bili's own injected prompt alike (#552). Wins + * per key over the global `compat` block. */ + compat?: { roles?: Record }; }; export type ProviderRoutes = Record; // key = upstream URL prefix (the /bili/ string) @@ -234,6 +242,11 @@ export type ProxyOptions = { injectNudge: boolean; }; promptCache: { routing: PromptCacheRouting }; + /** Wire-compat role map (global level; per-provider `compat.roles` overlays + * it per key). `{"developer":"system"}` rewrites developer→system on the + * forwarded body for upstreams without the developer role (#552). Empty = + * byte-for-byte transparent. */ + compat: { roles: Record }; sessionHeader: string; log: boolean; debug: boolean; @@ -373,6 +386,7 @@ export function loadOptions(env: NodeJS.ProcessEnv = process.env): ProxyOptions promptCache: { routing: parsePromptCacheRouting(env.ACP_PROMPT_CACHE_ROUTING ?? fileConfig.promptCache?.routing), }, + compat: { roles: parseCompatRoles(fileConfig.compat?.roles) ?? {} }, sessionHeader: env.ACP_SESSION_HEADER ?? fileConfig.sessionHeader ?? "x-acp-session", log: env.ACP_LOG !== "0" && fileConfig.log !== false, debug: (env.ACP_DEBUG ?? (fileConfig.debug ? "1" : "0")) === "1", @@ -421,6 +435,10 @@ type FileConfig = { compress?: CompressSettings & { injectTool?: boolean; injectNudge?: boolean }; promptCache?: { routing?: string }; mitm?: { enabled?: boolean; domains?: string[] }; + /** Global wire-compat block. `roles` maps message roles to the role name + * upstreams accept (e.g. `{"developer":"system"}`) — applied to the + * final forwarded body for openai/responses requests (#552). */ + compat?: { roles?: Record }; }; function nonEmpty(value: string | undefined): string | undefined { @@ -495,11 +513,13 @@ export function parseRouteEntry(v: unknown): ProviderRoute | undefined { // is the KEY in the providers map (identical to the /bili/ string), // so it is NOT repeated inside the value. if (v && typeof v === "object" && !Array.isArray(v)) { - const obj = v as { models?: Record; proxy?: string; compressProtocol?: string; compress?: CompressSettings }; + const obj = v as { models?: Record; proxy?: string; compressProtocol?: string; compress?: CompressSettings; compat?: { roles?: unknown } }; const route: ProviderRoute = { models: obj.models }; if (typeof obj.proxy === "string") route.proxy = obj.proxy; if (obj.compressProtocol === "marker" || obj.compressProtocol === "tools") route.compressProtocol = obj.compressProtocol; if (obj.compress) route.compress = obj.compress; + const compatRoles = parseCompatRoles(obj.compat?.roles); + if (compatRoles) route.compat = { roles: compatRoles }; return route; } // A bare value (e.g. null) means "this upstream exists, no overrides". diff --git a/src/loop/core.ts b/src/loop/core.ts index 53acfd8fc..25df9f0a0 100644 --- a/src/loop/core.ts +++ b/src/loop/core.ts @@ -79,6 +79,11 @@ export interface LoopCtx { export interface RequestOptions { url: string; headers: Record; + /** Final-stage wire transform (e.g. compat.roles role rewrite, #552). + * Applied to every outgoing body this loop re-sends — truncation retry, + * degraded retry, rebuilt rounds — so re-sent bodies carry the same + * compat the initial forward() applied. Returns the body to serialize. */ + wireTransform?: (body: Record) => Record; } export type ParsedStreamEvent = @@ -211,7 +216,7 @@ export async function* runCompressLoop( { method: "POST", headers: requestOptions.headers, - body: JSON.stringify(body), + body: JSON.stringify(requestOptions.wireTransform ? requestOptions.wireTransform(body) : body), ...(ctx.proxyUrl ? { dispatcher: proxyDispatcher(ctx.proxyUrl) } : {}), }, undefined, diff --git a/src/server.ts b/src/server.ts index c31e76154..132690790 100644 --- a/src/server.ts +++ b/src/server.ts @@ -74,6 +74,7 @@ import { systemToUser, isLoopbackAddress, inspectContextOverflow, reserveOutputH import { BILI_TUNNEL_HEADER, checkTunnelDestination, tunnelAllowlistFromEnv } from "./tunnel-guard.js"; import { decodeRequestBody } from "./content-encoding.js"; +import { applyCompatRoles, applyCompatRolesJson, resolveCompatRoles, type CompatRoles } from "./compat-roles.js"; // Body dumps (dumps/req-*.json, raw/*-REQ.txt, raw/*-RES.txt, raw/*-INCOMING.txt, // req-*-REREQUEST.json) write the full plaintext request body and are off by @@ -791,6 +792,7 @@ async function handle( opts.proxySource = fresh.proxySource; opts.proxyFallback = fresh.proxyFallback; opts.compress = fresh.compress; + opts.compat = fresh.compat; resetProxyCache(); for (const k of Object.keys(opts.routes)) delete opts.routes[k]; Object.assign(opts.routes, loadRoutes()); @@ -2424,6 +2426,16 @@ function logUpstreamProxyDecision(opts: ProxyOptions, upstreamUrl: string | unde logMsg(opts, "info", `[upstream-proxy] ${maskHostPortForLog(host)} ${via} (source=${decision.source})`); } +/** Infer the wire protocol from the request path for compat-role rewrites on + * requests the pipeline did not prepare (passthrough). Mirrors the path + * checks in handleRequest; returns null when unknown (no rewrite). */ +function inferWireProtocol(path: string): "openai" | "responses" | null { + const p = path.split("?", 2)[0]; + if (p.endsWith("/chat/completions")) return "openai"; + if (p.endsWith("/responses") || p.endsWith("/responses/compact")) return "responses"; + return null; +} + function buildForwardTarget( req: http.IncomingMessage, opts: ProxyOptions, @@ -2658,7 +2670,42 @@ async function forward( // wrongly skip and the user loses compression. When prepared is null any // inbound marker (from an upstream bili) is preserved verbatim by // buildForwardTarget, so the marker keeps propagating down the chain. + // #552: optional wire-compat role rewrite at the FINAL forward boundary — + // the only choke point that sees every emission site (client items, + // bili's injected compress prompt, instructions hoisting, compress-loop + // items). Opt-in via compat.roles (global + per-provider); empty map = + // byte-for-byte passthrough. + let wireBody: Buffer | string = body; + // #552 resolved compat map + protocol, shared with the compress-retry + // loops below (re-sent bodies must carry the same rewrite as the initial + // forward, or a developer-role 400 would hit mid-stream on retry). + let compatRoles: CompatRoles | null = null; + let compatProtocol: "openai" | "responses" | null = null; const { upstreamUrl, headers, proxyUrl } = buildForwardTarget(req, opts, route, affinity, prepared !== null ? instanceId : undefined); + if (typeof body === "string") { + // upstreamUrl (the real destination) — not route?.rewrittenUrl, which + // is undefined for zero-config requests and would skip provider compat. + const roles = resolveCompatRoles(opts.routes, upstreamUrl, opts.compat?.roles); + const protocol = prepared?.protocol ?? route?.explicitProtocol ?? inferWireProtocol(req.url ?? ""); + if (Object.keys(roles).length > 0 && (protocol === "openai" || protocol === "responses")) { + compatRoles = roles; + compatProtocol = protocol; + const applied = applyCompatRoles(body, protocol, roles); + if (applied.rewritten > 0) { + wireBody = applied.body; + log("info", `[${prepared?.session.id ?? "passthrough"}] [compat] rewrote ${applied.rewritten} message role(s) per compat.roles (${Object.entries(roles).map(([f, t]) => `${f}→${t}`).join(",")})`); + } + } + } + // #552: wire transform shared by ALL re-send paths (compress-retry loops + // below) so re-sent bodies carry the same rewrite as the initial forward — + // otherwise a developer-role 400 would hit mid-stream on the first retry. + const wireTransform = compatRoles && compatProtocol + ? (b: Record): Record => { + applyCompatRolesJson(b, compatProtocol, compatRoles); + return b; + } + : undefined; // Show the final proxied URL (where the request actually lands) as the // primary signal. The provider label is appended only for named routes — // zero-config requests have a single routing mode now, so the final @@ -2680,9 +2727,9 @@ async function forward( } } } - if (typeof body === "string" && (opts.debug || bodyDumpEnabled())) { + if (typeof wireBody === "string" && (opts.debug || bodyDumpEnabled())) { try { - const parsed = JSON.parse(body); + const parsed = JSON.parse(wireBody); if (opts.debug) { const toolNames = (parsed.tools ?? []).map((t: Record) => { const fn = t.function as { name?: string } | undefined; @@ -2697,10 +2744,10 @@ async function forward( const sid = prepared?.session.id ?? "unknown"; const out = `${dumpDir}/req-${Date.now()}-${safeSessionId(sid)}.json`; try { - const pretty = JSON.stringify(JSON.parse(body), null, 2); + const pretty = JSON.stringify(JSON.parse(wireBody), null, 2); fs.writeFileSync(out, pretty); } catch { - fs.writeFileSync(out, body); + fs.writeFileSync(out, wireBody); } log("info", `[debug] forwarded body written to ${out}`); } @@ -2741,9 +2788,9 @@ async function forward( const bodyText = req.method === "GET" || req.method === "HEAD" ? "" - : typeof body === "string" - ? body - : Buffer.from(body).toString("utf8"); + : typeof wireBody === "string" + ? wireBody + : Buffer.from(wireBody).toString("utf8"); const reqPath = `${rawBase}-REQ.txt`; fs.writeFileSync(reqPath, `${req.method ?? "POST"} ${maskUrlForLog(upstreamUrl)}\n${hdrText}\n\n${bodyText}`); log("info", `[debug] RAW request dump: ${reqPath}`); @@ -2753,7 +2800,7 @@ async function forward( const init: Omit & { dispatcher?: object } = { method: req.method ?? "GET", headers, - body: req.method === "GET" || req.method === "HEAD" ? undefined : body, + body: req.method === "GET" || req.method === "HEAD" ? undefined : wireBody, }; if (dispatcher) init.dispatcher = dispatcher; // Must be created before fetchWithTimeout: the signal aborts the upstream @@ -3091,7 +3138,7 @@ async function forward( streamToRead, { core, config, messages: prepared.processedMessages.length > 0 ? prepared.processedMessages : prepared.originalMessages, compressMessages: prepared.originalMessages, session: prepared.session, log: ctx.log, proxyUrl, protocol: prepared.protocol, textProtocol, debug: opts.debug, refreshFolded }, parsedReq, - { url: upstreamUrl, headers: reqHeaders }, + { url: upstreamUrl, headers: reqHeaders, wireTransform }, adapter, systemPrompt, clientAbort.signal, @@ -3146,7 +3193,7 @@ async function forward( json, { core, config, messages: prepared.originalMessages, session: prepared.session, log: ctx.log, proxyUrl, textProtocol: true }, requestBody, - { url: upstreamUrl, headers: requestHeaders }, + { url: upstreamUrl, headers: requestHeaders, wireTransform }, ); } // Capture upstream usage so tokenCount (which drives nudge + @@ -3370,6 +3417,7 @@ function handleConfigReload(opts: ProxyOptions, res: http.ServerResponse, log: ( for (const k of Object.keys(opts.routes)) delete opts.routes[k]; Object.assign(opts.routes, fresh); opts.compress = loadOptions().compress; + opts.compat = loadOptions().compat; // Release cached ProxyAgents so agents for proxy URLs that were // removed/changed don't leak for the process lifetime. The next request // re-creates the needed agent lazily via proxyDispatcher(). diff --git a/tests/compat-roles.test.ts b/tests/compat-roles.test.ts new file mode 100644 index 000000000..19503485c --- /dev/null +++ b/tests/compat-roles.test.ts @@ -0,0 +1,263 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import http from "node:http"; +import { once } from "node:events"; +import { mkdirSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { defaultConfig } from "acp-kernel"; +import { startServer } from "../src/server.ts"; +import { loadRoutes, type ProxyOptions } from "../src/config.ts"; +import { SessionStore, _setStoreForTest } from "../src/persist.ts"; +import { _setForTest as setRegistryForTest } from "../src/registry.ts"; +import { applyCompatRoles, applyCompatRolesJson, parseCompatRoles, resolveCompatRoles } from "../src/compat-roles.ts"; + +function close(server: http.Server): Promise { + return new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); +} + +async function freePort(): Promise { + const server = http.createServer(); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const port = (server.address() as { port: number }).port; + await close(server); + return port; +} + +const ROLES = { developer: "system" }; + +test("parseCompatRoles drops non-string entries", () => { + assert.equal(parseCompatRoles(undefined), undefined); + assert.equal(parseCompatRoles("developer"), undefined); + assert.equal(parseCompatRoles([]), undefined); + assert.equal(parseCompatRoles({}), undefined); + assert.deepEqual(parseCompatRoles({ developer: "system", bad: 42, empty: "", ok: "user" }), { developer: "system", ok: "user" }); +}); + +test("resolveCompatRoles: provider wins per key, global fills the rest", () => { + const routes = { + "https://api.a.com": { compat: { roles: { developer: "user", assistant: "user" } } }, + }; + // No match anywhere. + assert.deepEqual(resolveCompatRoles(routes, "https://api.b.com/v1/chat/completions", undefined), {}); + // Provider only. + assert.deepEqual(resolveCompatRoles(routes, "https://api.a.com/v1/chat/completions", undefined), { developer: "user", assistant: "user" }); + // Global + provider: provider wins per key. + assert.deepEqual( + resolveCompatRoles(routes, "https://api.a.com/v1/chat/completions", { developer: "system", extra: "user" }), + { developer: "user", assistant: "user", extra: "user" }, + ); +}); + +test("applyCompatRoles rewrites openai messages[].role", () => { + const body = JSON.stringify({ model: "m", messages: [{ role: "developer", content: "sys" }, { role: "user", content: "hi" }] }); + const out = applyCompatRoles(body, "openai", ROLES); + assert.equal(out.rewritten, 1); + assert.deepEqual(JSON.parse(out.body).messages[0], { role: "system", content: "sys" }); + assert.deepEqual(JSON.parse(out.body).messages[1], { role: "user", content: "hi" }); +}); + +test("applyCompatRoles rewrites responses input[] message roles, skips typed items", () => { + const body = JSON.stringify({ + model: "m", + input: [ + { type: "message", role: "developer", content: "sys" }, + { role: "developer", content: "sys2" }, + { type: "function_call", name: "f", call_id: "c", arguments: "{}" }, + { type: "message", role: "user", content: "hi" }, + ], + }); + const out = applyCompatRoles(body, "responses", ROLES); + assert.equal(out.rewritten, 2); + const input = JSON.parse(out.body).input as Array>; + assert.equal(input[0].role, "system"); + assert.equal(input[1].role, "system"); + assert.equal(input[2].name, "f"); + assert.equal(input[3].role, "user"); +}); + +test("applyCompatRoles default no-op returns the original string", () => { + const body = JSON.stringify({ messages: [{ role: "developer", content: "x" }] }); + assert.deepEqual(applyCompatRoles(body, "openai", {}), { body, rewritten: 0 }); + // No matching role → original bytes, no re-stringify. + const other = JSON.stringify({ messages: [{ role: "user", content: "x" }] }); + const out = applyCompatRoles(other, "openai", ROLES); + assert.equal(out.body, other); + assert.equal(out.rewritten, 0); + // Invalid JSON → untouched. + assert.deepEqual(applyCompatRoles("not json", "openai", ROLES), { body: "not json", rewritten: 0 }); +}); + +test("applyCompatRolesJson mutates parsed bodies for the retry loops", () => { + const parsed = { messages: [{ role: "developer", content: "x" }] } as Record; + assert.equal(applyCompatRolesJson(parsed, "openai", ROLES), 1); + assert.equal((parsed.messages as Array<{ role: string }>)[0].role, "system"); +}); + +function upstreamServer(status: number, onBody: (path: string, body: unknown) => void): Promise { + const server = http.createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (c: Buffer) => chunks.push(c)); + req.on("end", () => { + let parsed: unknown = null; + try { parsed = JSON.parse(Buffer.concat(chunks).toString("utf8")); } catch { /* ignore */ } + onBody(req.url ?? "", parsed); + res.writeHead(status, { "content-type": "application/json" }); + res.end(JSON.stringify({ ok: true })); + }); + }); + return new Promise((resolve) => server.listen(0, "127.0.0.1", () => resolve(server))); +} + +interface StartOpts { + compatJson: string; +} + +async function startProxy(upstream: http.Server, { compatJson }: StartOpts): Promise<{ port: number; opts: ProxyOptions; stop: () => Promise; cleanup: () => void }> { + _setStoreForTest(new SessionStore({ enabled: false })); + setRegistryForTest({}); + const root = path.join(tmpdir(), `bili-compat-roles-${process.pid}-${Date.now()}`); + const biliConfig = path.join(root, "billion-context.json"); + mkdirSync(root, { recursive: true }); + writeFileSync(biliConfig, compatJson, "utf8"); + const previous = process.env.BILI_CONFIG_FILE; + process.env.BILI_CONFIG_FILE = biliConfig; + const upstreamPort = (upstream.address() as { port: number }).port; + const port = await freePort(); + const opts: ProxyOptions = { + port, + host: "127.0.0.1", + upstream: `http://127.0.0.1:${upstreamPort}`, + routes: loadRoutes(), + proxy: "", + proxyMode: "direct", + proxySource: "direct", + modelContextLimit: 400_000, + kernelConfig: defaultConfig(400_000), + compress: { injectTool: true, injectNudge: true }, + promptCache: { routing: "auto" }, + compat: { roles: parseCompatRoles(JSON.parse(compatJson).compat?.roles) ?? {} }, + sessionHeader: "x-acp-session", + log: false, + debug: false, + passthrough: false, + passthroughSource: null, + autoUpdate: false, + mitm: { enabled: false, domains: [] }, + }; + const proxy = await startServer(opts); + if (!proxy.listening) await once(proxy, "listening"); + return { + port, + opts, + stop: async () => { await close(proxy); }, + cleanup: () => { + if (previous === undefined) delete process.env.BILI_CONFIG_FILE; else process.env.BILI_CONFIG_FILE = previous; + rmSync(root, { recursive: true, force: true }); + }, + }; +} + +test("e2e #552 A: responses developer role rewritten on forward", async () => { + const seen: Array<{ path: string; roles: string[] }> = []; + const upstream = await upstreamServer(200, (path, body) => { + const roles = ((body as { input?: Array<{ role?: string }> })?.input ?? []).map((i) => i.role ?? "?"); + seen.push({ path, roles }); + }); + const harness = await startProxy(upstream, { compatJson: `{"compat":{"roles":{"developer":"system"}}}` }); + try { + const res = await fetch(`http://127.0.0.1:${harness.port}/v1/responses`, { + method: "POST", + headers: { "content-type": "application/json", "x-session-id": "compat-roles-e2e" }, + body: JSON.stringify({ model: "test", input: [{ type: "message", role: "developer", content: "be terse" }, { type: "message", role: "user", content: "hi" }] }), + }); + assert.equal(res.status, 200); + assert.equal(seen.length, 1); + assert.deepEqual(seen[0].roles, ["system", "user"]); + } finally { + await harness.stop(); + harness.cleanup(); + await close(upstream); + } +}); + +test("e2e #552 B: chat-completions roles rewritten on the rebuilt wire body", async () => { + const seen: Array = []; + const upstream = await upstreamServer(200, (_path, body) => { + seen.push(((body as { messages?: Array<{ role?: string }> })?.messages ?? []).map((m) => m.role ?? "?")); + }); + // Note: the chat rebuild pipeline (kernel coreToOpenai) already normalizes + // developer→system before compat even runs — the LIVE surface of #552 is + // the responses path. Mapping system→user here proves the openai boundary + // rewrite genuinely fires on the rebuilt body. + const harness = await startProxy(upstream, { compatJson: `{"compat":{"roles":{"system":"user"}}}` }); + try { + const res = await fetch(`http://127.0.0.1:${harness.port}/v1/chat/completions`, { + method: "POST", + headers: { "content-type": "application/json", "x-session-id": "compat-roles-e2e" }, + body: JSON.stringify({ model: "test", messages: [{ role: "developer", content: "be terse" }, { role: "user", content: "hi" }] }), + }); + assert.equal(res.status, 200); + assert.equal(seen[0][0], "user", "rebuilt system role rewritten per compat"); + } finally { + await harness.stop(); + harness.cleanup(); + await close(upstream); + } +}); + +test("e2e #552 C: default (no compat config) is byte-for-byte transparent", async () => { + const seen: Array = []; + const upstream = await upstreamServer(200, (_path, body) => seen.push(body)); + const harness = await startProxy(upstream, { compatJson: `{"providers":{}}` }); + try { + const payload = { model: "test", input: [{ type: "message", role: "developer", content: "be terse" }] }; + const res = await fetch(`http://127.0.0.1:${harness.port}/v1/responses`, { + method: "POST", + headers: { "content-type": "application/json", "x-session-id": "compat-roles-e2e" }, + body: JSON.stringify(payload), + }); + assert.equal(res.status, 200); + // The pipeline may append its own compress nudge to message content — + // unrelated to compat. What compat guarantees is the ROLE reaching the + // upstream unchanged when no rewrite is configured. + const input = (seen[0] as { input: Array<{ role: string; content: string }> }).input; + assert.equal(input[0].role, "developer", "role untouched by default"); + assert.ok(input[0].content.startsWith("be terse"), "original content preserved"); + } finally { + await harness.stop(); + harness.cleanup(); + await close(upstream); + } +}); + +test("e2e #552 D: per-provider compat.roles wins over global", async () => { + const seen: Array = []; + const upstreamPortHolder: { port: number } = { port: 0 }; + const upstream = await upstreamServer(200, (_path, body) => { + const items = (body as { messages?: Array<{ role?: string }>; input?: Array<{ role?: string }> }); + seen.push([...(items.messages ?? []), ...(items.input ?? [])].map((m) => m.role ?? "?")); + }); + upstreamPortHolder.port = (upstream.address() as { port: number }).port; + const config = { + compat: { roles: { developer: "system" } }, + providers: { [`http://127.0.0.1:${upstreamPortHolder.port}`]: { compat: { roles: { developer: "user" } } } }, + }; + const harness = await startProxy(upstream, { compatJson: JSON.stringify(config) }); + try { + // responses path: developer survives the rebuild to the wire (unlike + // chat, which normalizes it), so provider precedence is observable. + const res = await fetch(`http://127.0.0.1:${harness.port}/v1/responses`, { + method: "POST", + headers: { "content-type": "application/json", "x-session-id": "compat-roles-e2e" }, + body: JSON.stringify({ model: "test", input: [{ type: "message", role: "developer", content: "be terse" }] }), + }); + assert.equal(res.status, 200); + assert.deepEqual(seen[0], ["user"], "provider compat entry wins per key"); + } finally { + await harness.stop(); + harness.cleanup(); + await close(upstream); + } +}); From d2b15b3affebe431b39d97801583e96c1270853b Mon Sep 17 00:00:00 2001 From: ranxianglei Date: Sun, 6 Sep 2026 20:38:01 +0800 Subject: [PATCH 2/3] feat: auto-retry role-rejecting upstreams with learned session mapping (#552) Learn-on-failure for compat roles: an upstream 400 'Invalid role: ...' triggers ONE auto-retry with the offending role rewritten to 'system'. On success the mapping is remembered session-scoped (in-memory on the session, never written to config) so later requests skip the 400 round-trip; the info log carries the permanent per-provider snippet. Retry failures pass the original 400 through verbatim. Detection is conservative (two body patterns + stopword guard). --- CONFIGURATION.md | 1 + CONFIGURATION.zh-CN.md | 1 + README.md | 8 +++ src/compat-roles.ts | 31 ++++++++++++ src/server.ts | 94 ++++++++++++++++++++++++++++++---- tests/compat-roles.test.ts | 101 ++++++++++++++++++++++++++++++++++++- 6 files changed, 225 insertions(+), 11 deletions(-) diff --git a/CONFIGURATION.md b/CONFIGURATION.md index 9ed14d1b7..59e6842ad 100644 --- a/CONFIGURATION.md +++ b/CONFIGURATION.md @@ -113,6 +113,7 @@ Top-level keys that control how the proxy listens and behaves globally. - **Default:** `{}` (disabled) - **Status:** ACTIVE - **Description:** Global wire-compat role map. `roles` maps message roles to the role name your upstream accepts, e.g. `{"compat":{"roles":{"developer":"system"}}}` rewrites `developer` → `system` on the final forwarded body for upstreams that reject the `developer` role (#552, newer codex clients). Applies to `openai` chat-completions and `responses` requests; exact-match roles only, everything else in the body is untouched; re-sent compress-retry bodies carry the same rewrite. Per-provider `compat.roles` entries (see [Providers](#providers)) win per key. Default `{}` forwards bodies byte-for-byte unchanged. +- **Learn-on-failure:** with no compat configured, an upstream `400 Invalid role: …` is auto-fixed — bili rewrites the offending role to `system`, retries once, and remembers the mapping **session-scoped** (in-memory on the session; never written to config). Later requests in that session skip the 400 round-trip. The info log emitted when the fix fires carries the permanent per-provider snippet. ### `proxy` diff --git a/CONFIGURATION.zh-CN.md b/CONFIGURATION.zh-CN.md index 20740bb0b..a8d959867 100644 --- a/CONFIGURATION.zh-CN.md +++ b/CONFIGURATION.zh-CN.md @@ -113,6 +113,7 @@ - **默认值：** `{}`（禁用） - **状态：** ACTIVE - **说明：** 全局线上兼容角色映射。`roles` 把消息角色映射为上游接受的角色名，例如 `{"compat":{"roles":{"developer":"system"}}}` 把 `developer` → `system`，用于拒绝 `developer` 角色的上游（#552，新版 codex 客户端会发送）。作用于 `openai` chat-completions 与 `responses` 请求；仅精确匹配角色，体内其它内容不动；压缩重试重发的请求体同样携带。按 provider 的 `compat.roles`（见 [Providers](#providers)）按键优先。默认 `{}` 逐字节透明转发。 +- **失败自学习：** 未配置 compat 时，上游返回 `400 Invalid role: …` 会被自动修复 —— bili 把被拒角色改写为 `system`，重试一次，并把学到的映射记在**会话上**（仅内存，绝不写入配置）。该会话后续请求免 400 往返。修复生效时打印的 info 日志附带可永久化的 per-provider 片段。 ### `proxy` diff --git a/README.md b/README.md index f2f42013e..2954dea91 100644 --- a/README.md +++ b/README.md @@ -372,6 +372,14 @@ prompt alike), global or per-provider, default off = byte-for-byte: } ``` +**No configuration needed for the common case.** When an upstream answers a +request with `400 Invalid role: …`, bili auto-rewrites the offending role to +`system`, retries the request once, and — if the retry succeeds — remembers +the mapping **for that session only** (nothing is written to your config). +Later requests in the session skip the 400 round-trip. The log line printed +when the auto-fix fires includes a copy-paste per-provider snippet if you +want the mapping permanently. + ## How sessions work The proxy needs a stable per-conversation identifier to isolate compression diff --git a/src/compat-roles.ts b/src/compat-roles.ts index 27c0d1c08..3b1a8e5f6 100644 --- a/src/compat-roles.ts +++ b/src/compat-roles.ts @@ -70,6 +70,37 @@ export function applyCompatRolesJson( return rewritten; } +/** Stop-words that follow "role" in prose ("Invalid role must be one…"): + * a captured token equal to one of these means the regex grabbed a word + * from the sentence, not a role name. */ +const ROLE_CAPTURE_STOPWORDS = new Set([ + "must", "should", "be", "is", "one", "of", "the", "a", "an", "in", "for", "not", "was", "and", "or", "to", "only", "allowed", "supported", "valid", "value", "message", "messages", +]); + +const ROLE_REJECTION_PATTERNS = [ + /(?:invalid|unknown|unsupported|unrecognized|unexpected)[a-z ]{0,24}?role\s*[:=]?\s*["'`]?([a-z][a-z0-9_-]{1,31})["'`]?/i, + /role\s*[:=]?\s*["'`]?([a-z][a-z0-9_-]{1,31})["'`]?\s+(?:is\s+)?not\s+(?:supported|allowed|recognized|valid|accepted)/i, +]; + +/** Detect an upstream 400 that exists only because of a message role the + * provider does not support (e.g. codex ≥0.153 sends "developer"; + * converting backends answer 400 "Invalid role: developer"). Conservative by + * design: status must be 400, the body must blame a role, and the captured + * token must look like a role name. Returns the offending role so the + * caller can auto-rewrite it to a supported one. */ +export function detectRoleRejection(status: number, bodyText: string): { role: string } | null { + if (status !== 400) return null; + const head = bodyText.slice(0, 4096); + for (const re of ROLE_REJECTION_PATTERNS) { + const m = re.exec(head); + if (!m) continue; + const role = m[1].toLowerCase(); + if (ROLE_CAPTURE_STOPWORDS.has(role)) continue; + return { role }; + } + return null; +} + export function applyCompatRoles( body: string, protocol: "openai" | "responses", diff --git a/src/server.ts b/src/server.ts index 132690790..43ec89c87 100644 --- a/src/server.ts +++ b/src/server.ts @@ -74,7 +74,7 @@ import { systemToUser, isLoopbackAddress, inspectContextOverflow, reserveOutputH import { BILI_TUNNEL_HEADER, checkTunnelDestination, tunnelAllowlistFromEnv } from "./tunnel-guard.js"; import { decodeRequestBody } from "./content-encoding.js"; -import { applyCompatRoles, applyCompatRolesJson, resolveCompatRoles, type CompatRoles } from "./compat-roles.js"; +import { applyCompatRoles, applyCompatRolesJson, detectRoleRejection, resolveCompatRoles, type CompatRoles } from "./compat-roles.js"; // Body dumps (dumps/req-*.json, raw/*-REQ.txt, raw/*-RES.txt, raw/*-INCOMING.txt, // req-*-REREQUEST.json) write the full plaintext request body and are off by @@ -2685,24 +2685,36 @@ async function forward( if (typeof body === "string") { // upstreamUrl (the real destination) — not route?.rewrittenUrl, which // is undefined for zero-config requests and would skip provider compat. - const roles = resolveCompatRoles(opts.routes, upstreamUrl, opts.compat?.roles); + const configured = resolveCompatRoles(opts.routes, upstreamUrl, opts.compat?.roles); + // #552 learn-on-failure: roles this session learned from a role- + // rejection 400 overlay the configured map (empirical wins per key), + // so later requests skip the 400 round-trip. Session-scoped only — + // config stays user-owned. + const learned = (prepared?.session.metadata.learnedCompatRoles as CompatRoles | undefined) ?? {}; + const roles = { ...configured, ...learned }; const protocol = prepared?.protocol ?? route?.explicitProtocol ?? inferWireProtocol(req.url ?? ""); - if (Object.keys(roles).length > 0 && (protocol === "openai" || protocol === "responses")) { - compatRoles = roles; + // compatProtocol is armed even with zero roles: the learn-on-failure + // retry below needs it, and roles may be learned mid-request. + if (protocol === "openai" || protocol === "responses") { compatProtocol = protocol; - const applied = applyCompatRoles(body, protocol, roles); - if (applied.rewritten > 0) { - wireBody = applied.body; - log("info", `[${prepared?.session.id ?? "passthrough"}] [compat] rewrote ${applied.rewritten} message role(s) per compat.roles (${Object.entries(roles).map(([f, t]) => `${f}→${t}`).join(",")})`); + if (Object.keys(roles).length > 0) { + compatRoles = roles; + const applied = applyCompatRoles(body, protocol, roles); + if (applied.rewritten > 0) { + wireBody = applied.body; + log("info", `[${prepared?.session.id ?? "passthrough"}] [compat] rewrote ${applied.rewritten} message role(s) per compat.roles (${Object.entries(roles).map(([f, t]) => `${f}→${t}`).join(",")})`); + } } } } // #552: wire transform shared by ALL re-send paths (compress-retry loops // below) so re-sent bodies carry the same rewrite as the initial forward — // otherwise a developer-role 400 would hit mid-stream on the first retry. - const wireTransform = compatRoles && compatProtocol + // Reads compatRoles at CALL time: a role learned mid-request (retry below) + // applies to later re-sends within the same request. + const wireTransform = compatProtocol ? (b: Record): Record => { - applyCompatRolesJson(b, compatProtocol, compatRoles); + if (compatRoles) applyCompatRolesJson(b, compatProtocol, compatRoles); return b; } : undefined; @@ -2819,6 +2831,68 @@ async function forward( recordUpstreamConnection(upstreamUrl, proxyUrl, error); throw new Error(`upstream request failed: ${formatUpstreamError(error, upstreamUrl, proxyUrl)}`, { cause: error }); } + // #552 learn-on-failure: a converting upstream that rejects a role (codex + // ≥0.153 sends "developer"; vLLM/SGLang-style backends answer 400 + // "Invalid role: developer") gets ONE auto-retry with the offending role + // rewritten to "system". On success the mapping is remembered on the + // session (never written to config — the log carries the permanent + // per-provider snippet instead). On failure the original response + // continues downstream verbatim. 400 bodies are small (buffered below). + if ( + compatProtocol && + typeof wireBody === "string" && + upstreamResult.response.status === 400 && + upstreamResult.response.body + ) { + let roleErrText: string | null = null; + try { + roleErrText = (await readStreamToBuffer(upstreamResult.response.body)).toString("utf8"); + } catch { + roleErrText = null; + } + if (roleErrText !== null) { + // Rebuild the consumed body so the error path below re-reads the + // same bytes verbatim (fetchWithTimeout ships rebuilt Responses + // itself, so this shape is established). + upstreamResult = { + response: new Response(roleErrText, { + status: upstreamResult.response.status, + statusText: upstreamResult.response.statusText, + headers: new Headers(upstreamResult.response.headers), + }), + clearTimer: upstreamResult.clearTimer, + }; + const rejection = detectRoleRejection(upstreamResult.response.status, roleErrText); + if (rejection && rejection.role !== "system") { + const fixed = applyCompatRoles(wireBody, compatProtocol, { [rejection.role]: "system" }); + if (fixed.rewritten > 0) { + try { + const retry = await fetchWithTimeout(upstreamUrl, { ...init, body: fixed.body }, undefined, clientAbort.signal); + if (retry.response.ok) { + upstreamResult.clearTimer(); + const s = prepared?.session; + if (s) { + const prev = (s.metadata.learnedCompatRoles as CompatRoles | undefined) ?? {}; + s.metadata.learnedCompatRoles = { ...prev, [rejection.role]: "system" }; + markDirty(s); + } + // Same-request re-sends (compress-retry loops) must + // carry the rewrite too — wireTransform reads this + // variable at call time. + compatRoles = { ...compatRoles, [rejection.role]: "system" }; + const providerKey = new URL(upstreamUrl).origin; + log("info", `[${prepared?.session.id ?? "passthrough"}] [compat] upstream rejected role "${rejection.role}" — auto-rewrote ${fixed.rewritten} message role(s) to "system", retry OK (remembered for this session only). To make permanent, add: {"providers":{"${providerKey}":{"compat":{"roles":{"${rejection.role}":"system"}}}}`); + upstreamResult = retry; + } else { + retry.clearTimer(); + } + } catch { + // retry transport failure — keep the original 400 + } + } + } + } + } const { response: upstream, clearTimer: clearUpstreamTimer } = upstreamResult; const respHeaders: Record = {}; const respConnNamed = connectionNamedHeaders(upstream.headers.get("connection") ?? undefined); diff --git a/tests/compat-roles.test.ts b/tests/compat-roles.test.ts index 19503485c..265e0dc04 100644 --- a/tests/compat-roles.test.ts +++ b/tests/compat-roles.test.ts @@ -10,7 +10,7 @@ import { startServer } from "../src/server.ts"; import { loadRoutes, type ProxyOptions } from "../src/config.ts"; import { SessionStore, _setStoreForTest } from "../src/persist.ts"; import { _setForTest as setRegistryForTest } from "../src/registry.ts"; -import { applyCompatRoles, applyCompatRolesJson, parseCompatRoles, resolveCompatRoles } from "../src/compat-roles.ts"; +import { applyCompatRoles, applyCompatRolesJson, detectRoleRejection, parseCompatRoles, resolveCompatRoles } from "../src/compat-roles.ts"; function close(server: http.Server): Promise { return new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); @@ -261,3 +261,102 @@ test("e2e #552 D: per-provider compat.roles wins over global", async () => { await close(upstream); } }); + +test("detectRoleRejection: extracts the offending role, conservative otherwise", () => { + assert.deepEqual(detectRoleRejection(400, '{"error":{"message":"Invalid role: developer"}}'), { role: "developer" }); + assert.deepEqual(detectRoleRejection(400, "Invalid message role: 'system'"), { role: "system" }); + assert.deepEqual(detectRoleRejection(400, "400 Bad Request: role developer is not supported"), { role: "developer" }); + assert.deepEqual(detectRoleRejection(400, 'Unexpected role=assistant for input'), { role: "assistant" }); + // Prose traps: captured token is a stopword → no detection. + assert.equal(detectRoleRejection(400, "Invalid role must be one of: system, user"), null); + // Wrong status / unrelated bodies. + assert.equal(detectRoleRejection(401, '{"error":{"message":"Invalid role: developer"}}'), null); + assert.equal(detectRoleRejection(400, '{"error":{"message":"insufficient quota"}}'), null); + // Pydantic-style validation error: the rejected role name is not in the + // text at all — must NOT produce a bogus detection. + assert.equal(detectRoleRejection(400, '[{"loc":["body","messages",0,"role"],"msg":"Input should be \'system\', \'user\', \'tool\' or \'assistant\'"}]'), null); +}); + +function roleRejectingUpstream(): Promise<{ server: http.Server; seen: string[][] }> { + const seen: string[][] = []; + const server = http.createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (c: Buffer) => chunks.push(c)); + req.on("end", () => { + let roles: string[] = []; + try { + const parsed = JSON.parse(Buffer.concat(chunks).toString("utf8")) as { input?: Array<{ role?: string }>; messages?: Array<{ role?: string }> }; + roles = [...(parsed.input ?? []), ...(parsed.messages ?? [])].map((m) => m.role ?? "?"); + } catch { /* ignore */ } + seen.push(roles); + if (roles.includes("developer")) { + res.writeHead(400, { "content-type": "application/json" }); + res.end(JSON.stringify({ error: { message: "Invalid role: developer" } })); + } else { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ ok: true })); + } + }); + }); + return new Promise((resolve) => server.listen(0, "127.0.0.1", () => resolve({ server, seen }))); +} + +test("e2e #552 E: role-rejection 400 auto-retries, learns session-scoped, skips the round-trip next time", async () => { + const { server: upstream, seen } = await roleRejectingUpstream(); + const harness = await startProxy(upstream, { compatJson: `{"providers":{}}` }); + try { + const payload = JSON.stringify({ model: "test", input: [{ type: "message", role: "developer", content: "be terse" }] }); + const res1 = await fetch(`http://127.0.0.1:${harness.port}/v1/responses`, { + method: "POST", + headers: { "content-type": "application/json", "x-session-id": "compat-roles-learn" }, + body: payload, + }); + assert.equal(res1.status, 200, "client sees a transparent 200 after the auto-retry"); + const res2 = await fetch(`http://127.0.0.1:${harness.port}/v1/responses`, { + method: "POST", + headers: { "content-type": "application/json", "x-session-id": "compat-roles-learn" }, + body: payload, + }); + assert.equal(res2.status, 200); + // 3 upstream hits total: req1 rejected (developer), req1-retried (system), + // req2 rewritten BEFORE fetch via the learned session map (system). + assert.equal(seen.length, 3, `expected 3 upstream hits, got ${JSON.stringify(seen)}`); + assert.deepEqual(seen[0], ["developer"], "first hit carries the client's developer role (no compat configured)"); + assert.deepEqual(seen[1], ["system"], "auto-retry rewrote developer→system"); + assert.deepEqual(seen[2], ["system"], "second request skipped the 400 round-trip (session learned)"); + } finally { + await harness.stop(); + harness.cleanup(); + await close(upstream); + } +}); + +test("e2e #552 F: retry that still fails passes the original 400 through verbatim", async () => { + const hits: number[] = []; + const upstream = http.createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (c: Buffer) => chunks.push(c)); + req.on("end", () => { + hits.push(1); + res.writeHead(400, { "content-type": "application/json" }); + res.end(JSON.stringify({ error: { message: "Invalid role: developer" } })); + }); + }); + await new Promise((resolve) => upstream.listen(0, "127.0.0.1", () => resolve())); + const harness = await startProxy(upstream, { compatJson: `{"providers":{}}` }); + try { + const res = await fetch(`http://127.0.0.1:${harness.port}/v1/responses`, { + method: "POST", + headers: { "content-type": "application/json", "x-session-id": "compat-roles-fail" }, + body: JSON.stringify({ model: "test", input: [{ type: "message", role: "developer", content: "be terse" }] }), + }); + assert.equal(res.status, 400, "client receives the upstream 400"); + const text = await res.text(); + assert.ok(text.includes("Invalid role: developer"), `original error body preserved verbatim, got: ${text}`); + assert.equal(hits.length, 2, "exactly one retry, no loop"); + } finally { + await harness.stop(); + harness.cleanup(); + await close(upstream); + } +}); From 9b6efc2fe9e19ab2be0132a47200b2d2f4ff4293 Mon Sep 17 00:00:00 2001 From: ranxianglei Date: Sun, 6 Sep 2026 21:07:16 +0800 Subject: [PATCH 3/3] fix: late body chunk must not re-arm the idle timer after clearTimer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A response body's pending pull can resolve AFTER the consumer called clearTimer (abandoned body — a failed retry whose response is never read, or a client abort racing the last chunk). The old code re-armed a fresh UPSTREAM_TIMEOUT_MS (10 min) timer in that window, pinning the event loop for ten minutes per leaked fetch; the new e2e #552 F suite hung exactly this way. Guard rearm() with a cleared flag set by cleanup(). --- src/fetch-util.ts | 7 +++++++ tests/compat-roles.test.ts | 11 +++++++++++ 2 files changed, 18 insertions(+) diff --git a/src/fetch-util.ts b/src/fetch-util.ts index 06baa9154..1c018ee70 100644 --- a/src/fetch-util.ts +++ b/src/fetch-util.ts @@ -52,6 +52,7 @@ export async function fetchWithTimeout( externalSignal?: AbortSignal, ): Promise<{ response: Response; clearTimer: () => void }> { const controller = new AbortController(); + let cleared = false; const armTimer = () => { const t = setTimeout(() => { liveUpstreamTimers.delete(t); @@ -62,6 +63,11 @@ export async function fetchWithTimeout( }; let timer = armTimer(); const rearm = () => { + // A late body chunk can resolve after the consumer already called + // clearTimer (abandoned response — e.g. a failed retry whose body is + // never read, or a client abort). Re-arming after cleanup would leak + // a fresh 10-minute timer and pin the event loop (#552 e2e hang). + if (cleared) return; clearTimeout(timer); liveUpstreamTimers.delete(timer); timer = armTimer(); @@ -75,6 +81,7 @@ export async function fetchWithTimeout( } } const cleanup = () => { + cleared = true; clearTimeout(timer); liveUpstreamTimers.delete(timer); if (onExternalAbort && externalSignal) externalSignal.removeEventListener("abort", onExternalAbort); diff --git a/tests/compat-roles.test.ts b/tests/compat-roles.test.ts index 265e0dc04..b3a952a7a 100644 --- a/tests/compat-roles.test.ts +++ b/tests/compat-roles.test.ts @@ -11,6 +11,7 @@ import { loadRoutes, type ProxyOptions } from "../src/config.ts"; import { SessionStore, _setStoreForTest } from "../src/persist.ts"; import { _setForTest as setRegistryForTest } from "../src/registry.ts"; import { applyCompatRoles, applyCompatRolesJson, detectRoleRejection, parseCompatRoles, resolveCompatRoles } from "../src/compat-roles.ts"; +import { _liveUpstreamTimersForTest } from "../src/fetch-util.ts"; function close(server: http.Server): Promise { return new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); @@ -354,9 +355,19 @@ test("e2e #552 F: retry that still fails passes the original 400 through verbati const text = await res.text(); assert.ok(text.includes("Invalid role: developer"), `original error body preserved verbatim, got: ${text}`); assert.equal(hits.length, 2, "exactly one retry, no loop"); + await waitFor(() => _liveUpstreamTimersForTest() === 0); + assert.equal(_liveUpstreamTimersForTest(), 0, "abandoned retry body must not re-arm the idle timer after clearTimer"); } finally { await harness.stop(); harness.cleanup(); await close(upstream); } }); + +async function waitFor(probe: () => boolean, deadlineMs = 3000): Promise { + const start = Date.now(); + while (!probe()) { + if (Date.now() - start > deadlineMs) return; + await new Promise((resolve) => setTimeout(resolve, 25)); + } +}