fix: honor anthropic-beta context-1m to size the context window

src/server.ts

@@ -100,6 +100,31 @@ function launcherContextWindow(model: string): number | undefined {
     return LAUNCHER_MODEL_WINDOWS[model];
 }
 
+/** Parse an `anthropic-beta` header for a larger-context beta (e.g.
+ *  `context-1m-2025-08-07` → 1,000,000). The beta lets the CLIENT negotiate a
+ *  window beyond the model's standard size, so it is the most direct per-request
+ *  evidence of the window the upstream will actually serve — it outranks the
+ *  model table / registry (which list the STANDARD window, e.g. 200K for claude)
+ *  and must be re-read on every request (the header may appear/disappear between
+ *  requests of the same session, #302). `context-Nm` generalizes to future
+ *  larger-context betas (N × 1,000,000). Returns the largest requested window,
+ *  or undefined when no context beta is present. */
+export function anthropicBetaContextWindow(headers: Record<string, string | string[] | undefined>): number | undefined {
+    const raw = headers["anthropic-beta"];
+    if (raw === undefined) return undefined;
+    const list = Array.isArray(raw) ? raw.join(",") : raw;
+    let best: number | undefined;
+    for (const part of list.split(",")) {
+        const m = /^context-(\d+)m\b/.exec(part.trim().toLowerCase());
+        if (!m) continue;
+        const n = Number.parseInt(m[1], 10);
+        if (!Number.isFinite(n) || n <= 0) continue;
+        const w = n * 1_000_000;
+        if (best === undefined || w > best) best = w;
+    }
+    return best;
+}
+
 /** Session ids are client-provided verbatim (#286) — sanitize before using
  *  one in a debug-dump FILENAME so a hostile value cannot escape the dir. */
 function safeSessionId(id: string | undefined): string {
@@ -702,9 +727,13 @@ async function handle(
         const model = (parsed as { model?: string }).model;
         if (model) {
             const embeddedUrl = route?.rewrittenUrl;
-            // Native-window resolution order: (1) a cooperative plugin's
-            // report (the agent's own config — most authoritative, gated on
-            // the x-bili-plugin marker so a plain client cannot rewrite the
+            // Native-window resolution order: (0) the client's `anthropic-beta`
+            // larger-context negotiation (context-1m-… → 1,000,000) — the most
+            // direct per-request evidence of the window the upstream will serve,
+            // so it outranks every static source (the model table / registry
+            // list the STANDARD window, e.g. 200K for claude); (1) a cooperative
+            // plugin's report (the agent's own config — most authoritative, gated
+            // on the x-bili-plugin marker so a plain client cannot rewrite the
             // nudge denominator by name); (1b) the launcher's per-model
             // windows (BILI_LAUNCHER_MODEL_WINDOWS — the client's own
             // models.json/models.yml contextWindow, authoritative for this
@@ -717,20 +746,24 @@ async function handle(
             // fallback. Operator tuning via compress.modelContextLimit still
             // outranks everything inside resolveRequestConfig.
             const host = (() => { try { return embeddedUrl ? new URL(embeddedUrl).host : undefined; } catch { return undefined; } })();
+            const betaWindow = anthropicBetaContextWindow(req.headers);
             const pluginWindow = pluginReportedContextWindow(req.headers);
             const launcherWindow = launcherContextWindow(model);
             const peekWindow = host ? peekRegistryContext(model, host) : undefined;
             const configuredWindow = resolveConfiguredContextLimit(opts.routes, embeddedUrl, model);
-            let native = pluginWindow
+            let native = betaWindow
+                ?? pluginWindow
                 ?? launcherWindow
                 ?? peekWindow
                 ?? configuredWindow
                 ?? lookupContextLimit(model);
             // Fallback = no authoritative source AND the operator did not
             // explicitly tune the window via compress.modelContextLimit (an
-            // explicit tuning is owned by the operator — never floored).
+            // explicit tuning is owned by the operator — never floored). The
+            // beta window is authoritative (the client's own runtime
+            // negotiation), so it also clears the fallback flag.
             const operatorWindowTuned = resolveCompress(opts.routes, embeddedUrl, model, opts.compress).modelContextLimit !== undefined;
-            nativeFromFallback = !pluginWindow && !launcherWindow && !peekWindow && !configuredWindow && !operatorWindowTuned;
+            nativeFromFallback = !betaWindow && !pluginWindow && !launcherWindow && !peekWindow && !configuredWindow && !operatorWindowTuned;
             if (!native && host) {
                 native = await contextFromRegistry(model, host);
                 if (native) nativeFromFallback = false;

tests/anthropic-beta-window.test.ts

@@ -0,0 +1,199 @@
+import assert from "node:assert/strict";
+import http from "node:http";
+import { once } from "node:events";
+import test from "node:test";
+
+process.env.NODE_ENV = "test";
+
+import { defaultConfig } from "acp-kernel";
+import { startServer, type ProxyOptions } from "../src/server.ts";
+import { SessionStore, _setStoreForTest } from "../src/persist.ts";
+import { _setForTest as setRegistryForTest } from "../src/registry.ts";
+import { listSessions } from "../src/session.ts";
+
+// #302: an `anthropic-beta: context-1m-…` header means the client negotiated a
+// 1M-token upstream window. The proxy must size its OWN context window to 1M
+// (not the model-table 200K) so the EMERGENCY / nudge / usage bands sit against
+// the real limit. Regression: the client-facing instance missed the header and
+// fired EMERGENCY at 261K input (131% of 200K) when it was only 26% of 1M.
+//
+// The window is resolved PER-REQUEST (the beta header may appear/disappear
+// between requests of the same session), so the resolution must read the header
+// on every request, not cache it on the session.
+
+const MODEL = "claude-sonnet-4-5"; // table default 200K (config.ts:121)
+const BETA = "context-1m-2025-08-07";
+const MEASURED = 261_206; // the ~261K input from the issue's log evidence
+
+function okSse(inputTokens: number): string {
+    return (
+        `event: message_start\ndata: ${JSON.stringify({ type: "message_start", message: { id: "m1", role: "assistant", usage: { input_tokens: inputTokens } } })}\n\n` +
+        `event: content_block_start\ndata: ${JSON.stringify({ type: "content_block_start", index: 0, content_block: { type: "text", text: "" } })}\n\n` +
+        `event: content_block_delta\ndata: ${JSON.stringify({ type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "ok" } })}\n\n` +
+        `event: content_block_stop\ndata: ${JSON.stringify({ type: "content_block_stop", index: 0 })}\n\n` +
+        `event: message_delta\ndata: ${JSON.stringify({ type: "message_delta", delta: { stop_reason: "end_turn", stop_sequence: null }, usage: { output_tokens: 3 } })}\n\n` +
+        `event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}\n\n`
+    );
+}
+
+// ~30K tokens of compressible content so the kernel has a viable T1 range
+// (minCompressRange = 5000 chars) to offer if EMERGENCY fires. Without some
+// compressible content the kernel suppresses the nudge even when over limit.
+function bigConversation(): Array<{ role: string; content: string }> {
+    const msgs: Array<{ role: string; content: string }> = [];
+    for (let i = 0; i < 12; i++) {
+        const role = i % 2 === 0 ? "user" : "assistant";
+        const filler = `MARKER_${i}_content_`.repeat(250);
+        msgs.push({ role, content: `Message ${i} of the long conversation. ${filler}` });
+    }
+    return msgs;
+}
+
+interface Rig {
+    proxyPort: number;
+    upstreamPort: number;
+    forwards: string[]; // raw bodies the proxy forwarded upstream (stream calls)
+    proxy: http.Server;
+    upstream: http.Server;
+}
+
+async function startRig(): Promise<Rig> {
+    const forwards: string[] = [];
+    const upstream = http.createServer((req, res) => {
+        const chunks: Buffer[] = [];
+        req.on("data", (c: Buffer) => chunks.push(c));
+        req.on("end", () => {
+            const raw = Buffer.concat(chunks).toString("utf8");
+            let parsed: { stream?: boolean } = {};
+            try { parsed = JSON.parse(raw); } catch { /* keep {} */ }
+            if (parsed.stream) {
+                forwards.push(raw);
+                res.writeHead(200, { "content-type": "text/event-stream" });
+                // Teach the session its real context size on the first forward
+                // (the upstream's own input_tokens); later forwards report a
+                // small post-fold size so the baseline doesn't drift.
+                res.end(okSse(forwards.length === 1 ? MEASURED : 1000));
+            } else {
+                res.writeHead(200, { "content-type": "application/json" });
+                res.end(JSON.stringify({ id: "msg_s", type: "message", role: "assistant", model: MODEL, content: [{ type: "text", text: "s" }], stop_reason: "end_turn", usage: { input_tokens: 500, output_tokens: 5 } }));
+            }
+        });
+    });
+    upstream.listen(0, "127.0.0.1");
+    await once(upstream, "listening");
+    const upstreamPort = upstream.address().port;
+
+    _setStoreForTest(new SessionStore({ enabled: false }));
+    setRegistryForTest({});
+    const proxy = await startServer({
+        port: 0,
+        host: "127.0.0.1",
+        upstream: "http://127.0.0.1",
+        routes: { [`http://127.0.0.1:${upstreamPort}`]: {} },
+        modelContextLimit: 200_000,
+        kernelConfig: defaultConfig(200_000),
+        compress: { injectTool: true, injectNudge: true },
+        promptCache: { routing: "auto" },
+        sessionHeader: "x-acp-session",
+        log: false,
+        debug: false,
+        passthrough: false,
+        autoUpdate: false,
+        mitm: { enabled: false, domains: [] },
+    } as ProxyOptions);
+    await once(proxy, "listening");
+    const proxyPort = proxy.address().port;
+    return { proxyPort, upstreamPort, forwards, proxy, upstream };
+}
+
+function url(rig: Rig): string {
+    return `http://127.0.0.1:${rig.proxyPort}/bili/http://127.0.0.1:${rig.upstreamPort}/v1/messages`;
+}
+
+// The EMERGENCY nudge is rendered as a trailing user message whose text begins
+// "⚠️ Context limit reached — compress now." (acp-kernel emergencyHeader).
+function hasEmergencyNudge(body: string): boolean {
+    return body.includes("Context limit reached");
+}
+
+async function closeRig(rig: Rig): Promise<void> {
+    rig.proxy.close();
+    await once(rig.proxy, "close");
+    rig.upstream.close();
+    await once(rig.upstream, "close");
+}
+
+test("e2e: anthropic-beta context-1m → 1M window → EMERGENCY suppressed at 261K input", async () => {
+    const rig = await startRig();
+    try {
+        const headers: Record<string, string> = {
+            "content-type": "application/json",
+            "x-acp-session": "beta-sess",
+            "anthropic-beta": BETA,
+        };
+        // Request 1: tiny payload, but the upstream reports a 261K-token
+        // context → session.stats.lastInputTokens = 261206.
+        const r1 = await fetch(url(rig), { method: "POST", headers, body: JSON.stringify({ model: MODEL, max_tokens: 1024, stream: true, messages: [{ role: "user", content: "hello" }] }) });
+        assert.equal(r1.status, 200);
+        await r1.text();
+        assert.equal(listSessions()[0]?.stats.lastInputTokens, MEASURED, "session context is 261K tokens");
+
+        // Request 2: a long (compressible) conversation. Under the 1M window
+        // this is 26% usage (< 75% nudge band) → no EMERGENCY, no nudge.
+        const r2 = await fetch(url(rig), { method: "POST", headers, body: JSON.stringify({ model: MODEL, max_tokens: 1024, stream: true, messages: bigConversation() }) });
+        assert.equal(r2.status, 200);
+        await r2.text();
+
+        const lastForward = rig.forwards[rig.forwards.length - 1]!;
+        assert.ok(!hasEmergencyNudge(lastForward), "no EMERGENCY nudge at 26% of the 1M window");
+    } finally {
+        await closeRig(rig);
+    }
+});
+
+test("e2e: without beta → 200K window → EMERGENCY fires at 261K input", async () => {
+    const rig = await startRig();
+    try {
+        const headers: Record<string, string> = {
+            "content-type": "application/json",
+            "x-acp-session": "no-beta-sess",
+        };
+        const r1 = await fetch(url(rig), { method: "POST", headers, body: JSON.stringify({ model: MODEL, max_tokens: 1024, stream: true, messages: [{ role: "user", content: "hello" }] }) });
+        assert.equal(r1.status, 200);
+        await r1.text();
+        assert.equal(listSessions()[0]?.stats.lastInputTokens, MEASURED, "session context is 261K tokens");
+
+        // Same 261K input, but no beta header → the window is the model-table
+        // 200K → 131% usage (>= 95% EMERGENCY threshold) → nudge appended.
+        const r2 = await fetch(url(rig), { method: "POST", headers, body: JSON.stringify({ model: MODEL, max_tokens: 1024, stream: true, messages: bigConversation() }) });
+        assert.equal(r2.status, 200);
+        await r2.text();
+
+        const lastForward = rig.forwards[rig.forwards.length - 1]!;
+        assert.ok(hasEmergencyNudge(lastForward), "EMERGENCY nudge fires at 131% of the 200K window");
+    } finally {
+        await closeRig(rig);
+    }
+});
+
+test("plugin: effectiveContextLimit honors anthropic-beta context-1m, per-request", async () => {
+    const rig = await startRig();
+    try {
+        const base: Record<string, string> = {
+            "content-type": "application/json",
+            "x-acp-session": "plugin-sess",
+            "x-bili-plugin": "test-agent",
+        };
+        // With the beta header → the window reported to the plugin is 1M.
+        await (await fetch(url(rig), { method: "POST", headers: { ...base, "anthropic-beta": BETA }, body: JSON.stringify({ model: MODEL, max_tokens: 1024, stream: true, messages: [{ role: "user", content: "hi" }] }) })).text();
+        assert.equal(listSessions()[0]?.metadata.effectiveContextLimit, 1_000_000, "beta → 1M reported to plugin");
+
+        // Same session, header disappears → the window drops back to the model
+        // default 200K. This is the per-request resolution the issue requires:
+        // the beta may appear/disappear between requests of one session.
+        await (await fetch(url(rig), { method: "POST", headers: base, body: JSON.stringify({ model: MODEL, max_tokens: 1024, stream: true, messages: [{ role: "user", content: "hi again" }] }) })).text();
+        assert.equal(listSessions()[0]?.metadata.effectiveContextLimit, 200_000, "no beta → 200K (per-request resolution)");
+    } finally {
+        await closeRig(rig);
+    }
+});

tests/model-windows.test.ts

@@ -11,7 +11,7 @@ import {
     collectModelWindows,
     type ClientConfig,
 } from "../src/client-config.ts";
-import { parseLauncherModelWindows } from "../src/server.ts";
+import { parseLauncherModelWindows, anthropicBetaContextWindow } from "../src/server.ts";
 
 test("parseOmpYaml: captures per-model contextWindow (models after baseUrl)", () => {
     const yml = [
@@ -147,3 +147,46 @@ test("parseLauncherModelWindows: valid JSON, invalid input, non-numeric filtered
     assert.deepEqual(parseLauncherModelWindows('["array"]'), {});
     assert.deepEqual(parseLauncherModelWindows('{"bad": "str", "neg": -5, "zero": 0}'), {});
 });
+
+// #302: an `anthropic-beta: context-1m-…` header means the client negotiated a
+// 1M window with the upstream — the model table's 200K default must be
+// overridden. The parser is per-request (the header may appear/disappear
+// between requests of the same session).
+
+test("anthropicBetaContextWindow: context-1m beta → 1,000,000", () => {
+    assert.equal(anthropicBetaContextWindow({ "anthropic-beta": "context-1m-2025-08-07" }), 1_000_000);
+});
+
+test("anthropicBetaContextWindow: no header / no context beta → undefined (model default)", () => {
+    assert.equal(anthropicBetaContextWindow({}), undefined);
+    assert.equal(anthropicBetaContextWindow({ "anthropic-beta": "prompt-caching-2024-07-31, interleaved-thinking-2025-05-14" }), undefined);
+    assert.equal(anthropicBetaContextWindow({ "anthropic-beta": "" }), undefined);
+});
+
+test("anthropicBetaContextWindow: mixed beta list picks the context beta", () => {
+    assert.equal(
+        anthropicBetaContextWindow({ "anthropic-beta": "prompt-caching-2024-07-31, context-1m-2025-08-07, fine-grained-tool-streaming-2025-05-14" }),
+        1_000_000,
+    );
+});
+
+test("anthropicBetaContextWindow: array header value is joined", () => {
+    assert.equal(anthropicBetaContextWindow({ "anthropic-beta": ["context-1m-2025-08-07"] }), 1_000_000);
+    assert.equal(anthropicBetaContextWindow({ "anthropic-beta": ["prompt-caching-2024-07-31", "context-1m-2025-08-07"] }), 1_000_000);
+});
+
+test("anthropicBetaContextWindow: case-insensitive + surrounding whitespace", () => {
+    assert.equal(anthropicBetaContextWindow({ "anthropic-beta": "  Context-1M-2025-08-07  " }), 1_000_000);
+});
+
+test("anthropicBetaContextWindow: future larger-context beta generalizes (context-Nm → N×1M)", () => {
+    assert.equal(anthropicBetaContextWindow({ "anthropic-beta": "context-2m-2026-01-01" }), 2_000_000);
+    // Largest wins when several context betas are present.
+    assert.equal(anthropicBetaContextWindow({ "anthropic-beta": "context-1m-2025-08-07, context-2m-2026-01-01" }), 2_000_000);
+});
+
+test("anthropicBetaContextWindow: rejects malformed / non-context tokens", () => {
+    assert.equal(anthropicBetaContextWindow({ "anthropic-beta": "context-m-2025-08-07" }), undefined);
+    assert.equal(anthropicBetaContextWindow({ "anthropic-beta": "context-1x-2025-08-07" }), undefined);
+    assert.equal(anthropicBetaContextWindow({ "anthropic-beta": "my-context-1m-2025-08-07" }), undefined);
+});