From 71cfe28b56f48071d57cabfdb3034f587fded59e Mon Sep 17 00:00:00 2001 From: ework-agent Date: Fri, 28 Aug 2026 00:16:55 +0800 Subject: [PATCH] fix: detect chained bili instances via x-bili-hop marker (#300) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When two bili instances are chained (A forward -> B), both run the compression pipeline on the same request, double-compressing and corrupting session state (root cause of #292). - Outbound: stamp x-bili-hop: on every request this instance actually processed (main forward + preflight summarization + compress-loop calls via shared headers). A passthrough forward does NOT claim processing, so a dumb-proxy bili in front of a processing one is unaffected; any inbound marker keeps propagating down the chain. - Inbound: if a request already carries x-bili-hop, an upstream bili already processed it. Skip ALL processing (no tool/tag injection, no acp-loop, no session state) and pass it through verbatim, logging a prominent [warn] naming the upstream instance (self-loop vs foreign chain). - e2e: two tests — B in isolation with a foreign marker (body reaches the LLM byte-identical, no tool/tag injection, marker propagated), and the full A->B chain (single-layer processing, B warns, one LLM request, marker is A's not B's). Clients never send this header, so single-instance use is unaffected. --- src/server.ts | 62 +++++++++-- tests/e2e-bili-chain.test.ts | 202 +++++++++++++++++++++++++++++++++++ 2 files changed, 258 insertions(+), 6 deletions(-) create mode 100644 tests/e2e-bili-chain.test.ts diff --git a/src/server.ts b/src/server.ts index 63b91f6e6..5e4af70f1 100644 --- a/src/server.ts +++ b/src/server.ts @@ -1,5 +1,6 @@ import http from "node:http"; import fs from "node:fs"; +import { randomUUID } from "node:crypto"; import { createCore, type CompressionCore, type CompressionState, type Config, type CoreMessage, type NudgeDecision, type Prompts, defaultPrompts, estimateTokensFast, renderNudgeText, deactivateBlock, viableRanges } from "acp-kernel"; import { resolveCompress, resolveCompressPrompts, resolveRequestConfig } from "./compress-settings.js"; import type { ProxyOptions } from "./config.js"; @@ -72,6 +73,15 @@ function bodyDumpEnabled(): boolean { return process.env.ACP_DUMP_BODY === "1"; } +// #300: bili→bili chain marker. When a bili instance forwards a request it has +// processed upstream, it stamps this header with its own instance id. A bili +// instance that RECEIVES a request already carrying it knows an upstream bili +// already ran the compression pipeline on this request — processing it again +// would double-compress and corrupt session state (issue #292). Clients never +// send this header, so its presence on an inbound request always means "came +// from a bili instance". +export const BILI_HOP_HEADER = "x-bili-hop"; + // Per-model context windows handed over by a `bili ` launcher // (BILI_LAUNCHER_MODEL_WINDOWS, JSON model-id → window), read from the // client's OWN config (pi models.json / omp models.yml / …) at launch time. @@ -216,6 +226,11 @@ export async function startServer(opts: ProxyOptions): Promise { const core = createCore(); const config: Config = opts.kernelConfig; const log = (level: string, msg: string) => logMsg(opts, level, msg); + // #300: per-server identity stamped into the x-bili-hop marker on outbound + // forwards. Per-server (not module-level) so two servers in one process + // (tests) are distinct instances; a restart changing the id is harmless + // (the chain check only compares against the other running instance). + const instanceId = randomUUID(); // Reload persisted compression state before accepting traffic so sessions // that survived a restart keep their folded view (otherwise long sessions // re-send oversized raw history and hang). @@ -231,7 +246,7 @@ export async function startServer(opts: ProxyOptions): Promise { void loadRegistry(); const server = http.createServer(async (req, res) => { try { - await handle(req, res, opts, core, config, log); + await handle(req, res, opts, core, config, log, instanceId); } catch (err) { const msg = String(err); log("error", msg); @@ -453,6 +468,7 @@ async function handle( core: CompressionCore, config: Config, log: (level: string, msg: string) => void, + instanceId: string, ): Promise { // SECURITY: the /__bili/ management endpoints (config read/write, reload, // session stats) are privileged — a remote caller who can reach them can @@ -646,6 +662,20 @@ async function handle( res.end(JSON.stringify({ error: { type: "invalid_request", message: String(err) } })); return; } + // #300: bili→bili chain detection. If the inbound request already carries + // the x-bili-hop marker, an upstream bili instance already ran the + // compression pipeline on it. Processing it again would double-compress + // and corrupt session state (#292). Skip ALL processing (no tool/tag + // injection, no acp-loop, no session state) and pass the request through + // verbatim. Clients never send this header, so its presence on an inbound + // request always means "came from a bili instance". + const hopMarker = headerValue(req, BILI_HOP_HEADER); + if (hopMarker !== undefined) { + const selfLoop = hopMarker === instanceId; + log("warn", selfLoop + ? `[chain] inbound request carries THIS instance's ${BILI_HOP_HEADER} marker (${hopMarker}) — self-loop detected. Passing through without processing; check your upstream config (it may point back to this instance).` + : `[chain] inbound request carries ${BILI_HOP_HEADER} from another bili instance (${hopMarker}) — bili→bili chain detected. Passing through without processing to avoid double compression; keep only one bili instance in the chain.`); + } const countTokens = isCountTokensRequest(req.method ?? "GET", urlPath, bodyBuffer.length > 0); // Per-request context limit: look up body.model against the per-route model // declaration in providers.json first (same model can have different @@ -740,7 +770,10 @@ async function handle( } } let prepared: Prepared | null = null; - if (!opts.passthrough && protocol && parsed && typeof parsed === "object") { + // #300: `hopMarker !== undefined` means an upstream bili already processed + // this request — skip the whole pipeline (prepared stays null) so the + // passthrough path below forwards it verbatim. + if (!opts.passthrough && hopMarker === undefined && protocol && parsed && typeof parsed === "object") { const sessionHeader = headerValue(req, opts.sessionHeader); // Plugin mode (issue #1, "内外呼应"): a cooperative agent-side plugin // announces itself with x-bili-plugin. The proxy then treats the @@ -969,9 +1002,10 @@ async function handle( route, affinity, log, + instanceId, ); } - await forward(req, res, opts, prepared!.body, prepared!, core, reqConfig, log, route, affinity); + await forward(req, res, opts, prepared!.body, prepared!, core, reqConfig, log, route, instanceId, affinity); // Remember for ALL modes (not just plugin): wire clients (dsh, // hermes, unplug'd pi) read the same panel via /__bili/plugin/status // and need the nudge/breakdown sections too. @@ -987,7 +1021,7 @@ async function handle( if (protocol === null && !opts.passthrough) { log("warn", `unrecognized path ${maskUrlsInText(req.url ?? "")} — not a known protocol (/chat/completions, /v1/messages, /responses, /responses/compact); forwarding unchanged`); } - await forward(req, res, opts, bodyBuffer, null, core, reqConfig, log, route, undefined); + await forward(req, res, opts, bodyBuffer, null, core, reqConfig, log, route, instanceId, undefined); } } @@ -1577,6 +1611,7 @@ function buildForwardTarget( opts: ProxyOptions, route: ReturnType, affinity?: string, + hopMarker?: string, ): ForwardTarget { // rewrittenUrl may use a `mitm://` scheme (for config-lookup distinction // — see resolveUpstream). fetch needs the real https:// scheme, so strip @@ -1592,6 +1627,11 @@ function buildForwardTarget( if (UPSTREAM_HOP_HEADERS.has(lower) || reqConnNamed.has(lower) || v === undefined) continue; headers[k] = Array.isArray(v) ? v.join(", ") : v; } + // #300: stamp the chain marker AFTER copying inbound headers so it wins + // over any inbound value (only set when this instance processed the + // request; a passthrough leaves the inbound marker — if any — intact so it + // keeps propagating down the chain). + if (hopMarker !== undefined) headers[BILI_HOP_HEADER] = hopMarker; headers["host"] = new URL(upstreamUrl).host; // codex advertises its own server-side context compaction via this beta // feature. It conflicts with bili's client-side compress (bili IS the @@ -1634,6 +1674,7 @@ async function preflightCompressIfNeeded( route: ReturnType, affinity: string | undefined, log: (level: string, msg: string) => void, + instanceId: string, ): Promise { const session = prepared.session; const limit = config.modelContextLimit; @@ -1644,7 +1685,9 @@ async function preflightCompressIfNeeded( if (limit <= 0 || !model || tokenCount < limit) return prepared; if ((prepared.nudge?.compressibleRanges ?? []).length === 0) return prepared; log("warn", `[${session.id}] context ${tokenCount} tokens exceeds model window ${limit} (model=${model}); preflight compressing before forward`); - const { upstreamUrl, headers, proxyUrl } = buildForwardTarget(req, opts, route, affinity); + // #300: stamp the chain marker so a downstream bili skips these + // summarization calls too (preflight always processes). + const { upstreamUrl, headers, proxyUrl } = buildForwardTarget(req, opts, route, affinity, instanceId); const clientAbort = new AbortController(); res.on("close", () => { if (!res.writableEnded) clientAbort.abort(); @@ -1688,9 +1731,16 @@ async function forward( config: Config, log: (level: string, msg: string) => void, route: ReturnType, + instanceId: string, affinity?: string, ): Promise { - const { upstreamUrl, headers, proxyUrl } = buildForwardTarget(req, opts, route, affinity); + // #300: stamp the chain marker ONLY when this instance actually processed + // the request (prepared !== null). A passthrough forward (prepared === null) + // must NOT claim processing — otherwise a downstream processing bili would + // 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. + const { upstreamUrl, headers, proxyUrl } = buildForwardTarget(req, opts, route, affinity, prepared !== null ? instanceId : 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 diff --git a/tests/e2e-bili-chain.test.ts b/tests/e2e-bili-chain.test.ts new file mode 100644 index 000000000..a5fb21ef3 --- /dev/null +++ b/tests/e2e-bili-chain.test.ts @@ -0,0 +1,202 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import http from "node:http"; +import { once } from "node:events"; +import { randomUUID } from "node:crypto"; +import { defaultConfig } from "acp-kernel"; +import { startServer, BILI_HOP_HEADER } from "../src/server.ts"; +import { SessionStore, _setStoreForTest } from "../src/persist.ts"; +import type { ProxyOptions } from "../src/config.ts"; +import { _setForTest as setRegistryForTest } from "../src/registry.ts"; +import { setLogCapture } from "../src/logger.ts"; + +/** #300: chained bili instances (A forward → B) must not double-process. + * A stamps x-bili-hop on processed forwards; B, seeing a foreign marker, + * skips ALL processing and passes the request through verbatim (logging a + * prominent warn). Two tests: (1) B in isolation with a foreign marker — + * proves B rewrites nothing; (2) the full A→B chain — proves single-layer + * processing end to end. */ + +function listen(server: http.Server): Promise { + if (server.listening) return Promise.resolve(); + return once(server, "listening").then(() => undefined); +} + +function close(server: http.Server): Promise { + return new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve()))); +} + +const MODEL = "gpt-test"; + +function makeOpts(port: number, upstream: string): ProxyOptions { + return { + port, + host: "127.0.0.1", + upstream, + routes: { [upstream]: { models: { [MODEL]: { context: 400_000 } } } }, + modelContextLimit: 400_000, + kernelConfig: defaultConfig(400_000), + compress: { injectTool: true, injectNudge: true }, + promptCache: { routing: "auto" }, + sessionHeader: "x-acp-session", + log: true, + debug: false, + passthrough: false, + autoUpdate: false, + logFile: "off", + mitm: { enabled: false, domains: [] }, + }; +} + +type Captured = { url: string; headers: http.IncomingHttpHeaders; body: string }; + +function makeUpstream(captured: Captured[]): http.Server { + return http.createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (c: Buffer) => chunks.push(c)); + req.on("end", () => { + const body = Buffer.concat(chunks).toString("utf8"); + captured.push({ url: req.url ?? "", headers: req.headers, body }); + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ + id: "chatcmpl-test", + object: "chat.completion", + model: MODEL, + choices: [{ index: 0, message: { role: "assistant", content: "ok" }, finish_reason: "stop" }], + usage: { prompt_tokens: 10, completion_tokens: 5, total_tokens: 15 }, + })); + }); + }); +} + +function chainWarnings(logs: { level: string; msg: string }[]): { level: string; msg: string }[] { + return logs.filter((l) => l.level === "warn" && l.msg.includes("[chain]") && l.msg.includes(BILI_HOP_HEADER)); +} + +test("#300: inbound foreign x-bili-hop → B passes through WITHOUT processing", async () => { + _setStoreForTest(new SessionStore({ enabled: false })); + setRegistryForTest({}); + const logs: { level: string; msg: string }[] = []; + setLogCapture((level, msg) => logs.push({ level, msg })); + + const captured: Captured[] = []; + const upstream = makeUpstream(captured); + upstream.listen(0, "127.0.0.1"); + await listen(upstream); + const llmUrl = `http://127.0.0.1:${(upstream.address() as { port: number }).port}`; + + const B = await startServer(makeOpts(0, llmUrl)); + await listen(B); + const bPort = (B.address() as { port: number }).port; + + const otherMarker = randomUUID(); + const bodyJson = JSON.stringify({ + model: MODEL, + stream: false, + messages: [ + { role: "system", content: "You are a test assistant." }, + { role: "user", content: "hello world" }, + ], + }); + + try { + const resp = await fetch(`http://127.0.0.1:${bPort}/v1/chat/completions`, { + method: "POST", + headers: { + "content-type": "application/json", + "x-acp-session": "chain-isolated", + [BILI_HOP_HEADER]: otherMarker, + }, + body: bodyJson, + }); + assert.equal(resp.status, 200); + await resp.text(); + + // B detected the chain and logged a prominent warn naming the upstream instance. + const warns = chainWarnings(logs); + assert.equal(warns.length, 1, `expected exactly one chain warning, got ${warns.length}: ${JSON.stringify(warns)}`); + assert.ok(warns[0]!.msg.includes(otherMarker), "chain warning should name the upstream instance marker"); + + // B forwarded the request to the LLM exactly once… + assert.equal(captured.length, 1, `expected exactly one LLM request, got ${captured.length}`); + const atLlm = captured[0]!; + // …with the body BYTE-IDENTICAL (B did not run the pipeline: no tag/tool + // injection, no re-serialization). + assert.equal(atLlm.body, bodyJson, "body must reach the LLM byte-identical (B must not rewrite it)"); + const parsed = JSON.parse(atLlm.body) as { tools?: unknown[] }; + assert.ok(!parsed.tools || parsed.tools.length === 0, "B must not inject tools"); + assert.ok(!atLlm.body.includes("\x3cacp "), "B must not inject acp tags"); + // The foreign marker propagated through B unchanged (B did not re-stamp it). + assert.equal(atLlm.headers[BILI_HOP_HEADER], otherMarker, "foreign marker must propagate through B unchanged"); + } finally { + setLogCapture(null); + B.closeAllConnections?.(); + await close(B); + upstream.closeAllConnections?.(); + await close(upstream); + } +}); + +test("#300: A→B chain → single-layer processing, B warns + passes through", async () => { + _setStoreForTest(new SessionStore({ enabled: false })); + setRegistryForTest({}); + const logs: { level: string; msg: string }[] = []; + setLogCapture((level, msg) => logs.push({ level, msg })); + + const captured: Captured[] = []; + const upstream = makeUpstream(captured); + upstream.listen(0, "127.0.0.1"); + await listen(upstream); + const llmUrl = `http://127.0.0.1:${(upstream.address() as { port: number }).port}`; + + const B = await startServer(makeOpts(0, llmUrl)); + await listen(B); + const bUrl = `http://127.0.0.1:${(B.address() as { port: number }).port}`; + + const A = await startServer(makeOpts(0, bUrl)); + await listen(A); + const aPort = (A.address() as { port: number }).port; + + const body = { + model: MODEL, + stream: false, + messages: [ + { role: "system", content: "You are a test assistant." }, + { role: "user", content: "hello world" }, + ], + }; + + try { + // Client → A via zero-config /bili//… so A's upstream is B. + const resp = await fetch(`http://127.0.0.1:${aPort}/bili/${bUrl}/v1/chat/completions`, { + method: "POST", + headers: { "content-type": "application/json", "x-acp-session": "chain-e2e" }, + body: JSON.stringify(body), + }); + assert.equal(resp.status, 200); + await resp.text(); + + // B saw A's marker and warned (exactly once — only B is chained here). + const warns = chainWarnings(logs); + assert.equal(warns.length, 1, `expected exactly one chain warning (from B), got ${warns.length}: ${JSON.stringify(warns)}`); + + // The request reached the LLM exactly once (no duplicated processing hop). + assert.equal(captured.length, 1, `expected exactly one LLM request, got ${captured.length}`); + const atLlm = captured[0]!; + // A processed + stamped the request (marker present ⇒ A ran the pipeline). + const markerAtLlm = atLlm.headers[BILI_HOP_HEADER]; + assert.ok(markerAtLlm, "A's x-bili-hop marker must reach the LLM (A processed the request)"); + // B passed A's marker through verbatim (did NOT re-stamp with its own id) + // — i.e. the marker B warned about is the same one that reached the LLM. + assert.ok(warns[0]!.msg.includes(markerAtLlm as string), + "B's chain warning must name the same marker that reached the LLM (A's, not B's)"); + } finally { + setLogCapture(null); + A.closeAllConnections?.(); + await close(A); + B.closeAllConnections?.(); + await close(B); + upstream.closeAllConnections?.(); + await close(upstream); + } +});