fix: detect chained bili instances via x-bili-hop marker (#300)

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 <client>` 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<http.Server> {
     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<http.Server> {
     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<void> {
     // 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<typeof resolveUpstream>,
     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<typeof resolveUpstream>,
     affinity: string | undefined,
     log: (level: string, msg: string) => void,
+    instanceId: string,
 ): Promise<Prepared> {
     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<typeof resolveUpstream>,
+    instanceId: string,
     affinity?: string,
 ): Promise<void> {
-    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

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<void> {
+    if (server.listening) return Promise.resolve();
+    return once(server, "listening").then(() => undefined);
+}
+
+function close(server: http.Server): Promise<void> {
+    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/<B-url>/… 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);
+    }
+});