diff --git a/CHANGELOG.md b/CHANGELOG.md index bbb6f57c..726c4c45 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ Versions follow the merge of a `*_release-v*` branch; CI publishes to npm on tag ### Fixes +- **Proactive nudge fires on upstreams that NEVER report usage — no more unbounded climb to the hard limit (#728)**: ChatGPT-login-style backends omit `usage.input_tokens` entirely (`response.completed` carries none), so `lastInputTokens` stayed 0 for the whole session — and the kernel's `decideNudge` is structurally unfireable at tokenCount == 0 (growth reference falls back to tokenCount itself → growth ≡ 0; mass-ready and all pressure bands require usage ≥ their pct lines). Proactive compression therefore never fired and contexts climbed unbounded until preflight's hard-limit emergency path took over (#726 reached ~1.32M tokens against a 1.05M window before the ~630K-token emergency summarization — which then 502-looped). `effectiveTokenCount` fed 0 for explicit-identity zero-baseline sessions on the "self-heals via the next measured usage report" assumption, which breaks for never-reporting upstreams. The fix mirrors how OpenAI handles exactly this in codex itself (server-observed usage first-class, local estimator fallback underneath — verified against `openai/codex` sources during triage): each `prepare*` now records the char-count upper bound of THAT turn's outbound payload (post-fold processed messages + system/tools overhead + image tokens) into `session.stats.localInputEstimate`, and `effectiveTokenCount` feeds it ONLY while `lastInputTokens == 0`, capped by the current request's inbound upper bound so a stale-high reading can't outlive a client-side shrink. Real usage always takes precedence the moment any report lands (same invariant as #604's `armFailureShrink` exception — the upper-bound estimator only errs early, compresses earlier, never later); turn 1 of a fresh explicit session is byte-identical to before (nothing measured yet → still 0); anonymous-prefix-affinity forks keep their raw-inbound-bound regime untouched; the estimate self-corrects after every fold (post-fold outbound payload shrinks → next reading drops, no inflation even though proxy clients re-send full raw history every turn); it is cleared on the native-compaction-boundary reset. Net effect: silent-backend sessions get incremental compression from the moment the local estimate crosses the nudge thresholds instead of one giant emergency fold at the wall. Rebase onto current master exposed an interaction with #817's per-body preflight dead-end cooldown key: the nudge text injected into the system prompt changes every turn, so keying the cooldown on the sha256 of the proxy-REBUILT wire body misses on identical client retries once the fallback arms the nudge — every retry re-burned the doomed summarization walk, defeating #726's quota protection on exactly the silent backends this fix targets. The cooldown identity now hashes the client's raw inbound request body instead, so proxy-side injections (nudge/system/tools/tags) can no longer defeat it. Tests pin: silent-backend multi-turn (turn 1 idle → turn 2 nudges, preflight silent), real-usage precedence (reported tokens beat a ~90% payload), stale-high cap (shrunk history does not re-trigger). - **Bump acp-kernel 0.0.74 → 0.0.77; `acp_cache` held off every wire surface until adoption (#800)**: the pin jumps three kernel releases — 0.0.75 (cache-reconciliation core: `acp_cache` tool + `buildCacheReport`; recommend tail-below-gate fix), 0.0.76 (Google/Gemini wire codec; recommend range integrity), 0.0.77 (nudge ranges ordered/merged by position instead of ref number — fixes #887). Kernel ≥0.0.75 ships a fifth proxy tool, `acp_cache`, but this proxy has no execution path for it: an intercepted call fell through `executeProxyTool` into the unknown-tool branch, which reports SUCCESS with `[Unknown proxy tool: acp_cache]` — a model calling it would get a plausible-looking result that did nothing. Until adoption lands (feature #800, owner decision pending), `src/compress-tool.ts` derives `PROXY_TOOL_NAMES` and the four `BILI_ACP_TOOLS_*` arrays from the kernel constants minus `UNADOPTED_KERNEL_TOOLS` (`{acp_cache}`) — the single choke point every serving surface consumes (Responses/OpenAI/Anthropic tool arrays, plugin manifest + allowed list, absorb gate, Responses-adapter classification), so the tool stays off the wire everywhere without touching call sites. `MUTATING_PROXY_TOOLS`/`READONLY_PROXY_TOOLS` stay raw re-exports (classification sets only; `acp_cache` is correctly classified readonly and no serving surface reads them). The #841 parity test (`tests/search-context-cross-session.test.ts`) extends the "no other tool may change" contract against the FILTERED kernel side only, so a future BILI-side leak of an unadopted tool still fails. - **A render tag wrapped around the model's whole turn no longer reaches the client as orphan markup (PR #872)**: a model imitating the render tag can write its entire turn — tool call included — where the opening's attributes are still open (recorded shape: `= their pct lines). #728 fix: such sessions fall back to +// the PREVIOUS turn's locally-measured outbound payload upper bound +// (session.stats.localInputEstimate, recorded in prepare* each turn). The +// estimator only errs EARLY (char-count upper bound → compress earlier, never +// later), is active only while lastInputTokens == 0 (a real usage report takes +// precedence immediately — same invariant as #604's armFailureShrink +// exception), and self-corrects after every fold (the post-fold payload +// shrinks → the next estimate drops). Turn 1 of a fresh explicit session +// still feeds 0 (nothing measured yet; nothing pending either), so +// first-turn behavior is byte-identical to pre-#728. function effectiveTokenCount(session: Session, msgs: CoreMessage[]): number { if (session.stats.lastInputTokens > 0) return session.stats.lastInputTokens; - if (!session.metadata.anonymousPrefixAffinity) return 0; - return estimateCoreMessagesUpper(msgs); + if (session.metadata.anonymousPrefixAffinity) return estimateCoreMessagesUpper(msgs); + const est = session.stats.localInputEstimate ?? 0; + if (est <= 0) return 0; + // Cap by THIS request's inbound upper bound: the recorded estimate lags by + // one turn, so after a client-side shrink (native compaction echo, history + // edit) the previous turn's payload can be larger than what is in front of + // us now — never claim more context than the current request could hold. + // In steady state est <= raw bound always (the outbound fold is never + // larger than the inbound history), so this is a no-op there. + const raw = estimateCoreMessagesUpper(msgs); + return Math.min(est, raw); } function prepareAnthropic( @@ -1956,6 +1979,16 @@ function prepareAnthropic( // identity chain (#268), not part of the Anthropic Messages API — strip it // so the real upstream never sees a field it doesn't know. delete (rebuilt as Record).prompt_cache_key; + // #728: record the char-count upper bound of THIS turn's outbound payload + // (post-fold messages + system/tools overhead + images) as the fallback + // token source for upstreams that never report usage — read only while + // lastInputTokens == 0 (effectiveTokenCount). When the kernel transform + // above failed, processedMessages is empty and the forwarded body is the + // UNPROCESSED projection — measure that instead so the fallback isn't + // blinded to a system+tools-only floor. + session.stats.localInputEstimate = estimateCoreMessagesUpper(processedMessages.length > 0 ? processedMessages : originalMessages) + + countSystemAndToolsTokens(extractSystem(systemOut), toolsOut) + + imageTokensInParsedBody("anthropic", rebuilt); return { body: JSON.stringify(rebuilt), session, processedMessages, originalMessages, anthropicSystem: parsed.system, protocol: "anthropic", stream, compressInjected: injectTools, pluginMode, nudge, prompts, surface, renderTags: "text-only" } as Prepared; } @@ -2113,6 +2146,15 @@ function prepareOpenai( if (!isTitleGen && openaiOutboundSystem !== undefined) { session.metadata.systemPromptTokens = countSystemAndToolsTokens(openaiOutboundSystem, toolsOut); } + // #728: record this turn's outbound payload upper bound as the fallback + // token source for upstreams that never report usage (see effectiveTokenCount). + // Title-gen side requests are skipped like the overhead row above — their + // tiny payload would clobber the conversation's measurement. + if (!isTitleGen) { + session.stats.localInputEstimate = estimateCoreMessagesUpper(processedMessages.length > 0 ? processedMessages : originalMessages) + + countSystemAndToolsTokens(openaiOutboundSystem || openaiSystemText, toolsOut) + + imageTokensInParsedBody("openai", rebuilt); + } snapshotMessages(session, originalMessages); markDirty(session); return { body: JSON.stringify(rebuilt), session, processedMessages, originalMessages, protocol: "openai", stream, compressInjected: injectTools, pluginMode, nudge, prompts, surface, openaiSystemText, renderTags: "text-only" } as Prepared; @@ -2373,6 +2415,15 @@ function prepareResponses( if (transformOk) { session.metadata.systemPromptTokens = countSystemAndToolsTokens(responsesDevContent ?? "", toolsOut); } + // #728: record this turn's outbound payload upper bound as the fallback + // token source for upstreams that never report usage (see effectiveTokenCount). + // Compaction-trigger requests are the compression mechanism itself — no + // incremental decision hangs off them, so don't leave a stale reading. + if (!isCompactionTrigger) { + session.stats.localInputEstimate = estimateCoreMessagesUpper(processedMessages.length > 0 ? processedMessages : originalMessages) + + countSystemAndToolsTokens(responsesDevContent ?? "", toolsOut) + + imageTokensInParsedBody("responses", rebuilt); + } snapshotMessages(session, originalMessages); markDirty(session); return { @@ -2829,6 +2880,7 @@ async function preflightCompressIfNeeded( prepared: Prepared, runPrepare: () => Prepared, req: http.IncomingMessage, + inboundBody: Buffer, res: http.ServerResponse, opts: ProxyOptions, core: CompressionCore, @@ -2950,7 +3002,12 @@ async function preflightCompressIfNeeded( // stale-baseline fit): those return without running the walk, so the // cooldown must not convert a fitting payload into a false fail-fast while // a marker from a larger earlier request is still warm. - const deadEndKey = `${model}\u0000${limit}\u0000${createHash("sha256").update(prepared.body).digest("hex")}`; + // #726 identity is the CLIENT request, not the rebuilt wire body: proxy-side + // injections vary turn to turn (the nudge text changes every turn; #728's + // silent-backend fallback arms the nudge on exactly these never-report-usage + // upstreams), so hashing prepared.body misses the cooldown on retry and + // re-burns upstream quota — the failure #726 exists to prevent. + const deadEndKey = `${model}\u0000${limit}\u0000${createHash("sha256").update(inboundBody).digest("hex")}`; const deadEnd = session.metadata.preflightDeadEnd; if (deadEnd && typeof deadEnd === "object") { const de = deadEnd as Record; diff --git a/src/session.ts b/src/session.ts index c5549274..8b4abba4 100644 --- a/src/session.ts +++ b/src/session.ts @@ -107,6 +107,16 @@ export type Session = { pendingFoldUsage?: boolean; /** Current in-context (uncompressed) token count at last processTurn. */ contextTokens: number; + /** #728: char-count upper bound of the LAST turn's outbound payload + * (post-fold processed messages + system/tools overhead + images), + * recorded locally in prepare* each turn. Read ONLY while + * lastInputTokens == 0, as the fallback tokenCount for upstreams + * that never report usage (ChatGPT-login-style backends — see + * effectiveTokenCount in server.ts). Self-correcting: a successful + * fold shrinks the next turn's payload and thus the estimate. + * Cleared by resetSessionCompression (native-compaction boundary). + * Persisted (survives restart like the rest of stats). */ + localInputEstimate?: number; }; /** Free-form escape hatch for future fields not yet promoted to typed * members. Persisted as-is (must be JSON-serializable). Use sparingly — @@ -378,6 +388,10 @@ export function resetSessionCompression(session: Session): void { session.stats.lastInputTokens = 0; // #857: a zeroed baseline carries no provenance — drop any stale flag. delete session.stats.lastInputTokensSource; + // #728: the pre-compaction outbound payload is gone — the old estimate + // (measured against the pre-compaction wire) would read high and blind + // the nudge fallback early; let the next prepare* re-measure. + session.stats.localInputEstimate = 0; session.stats.contextTokens = 0; session.metadata.nativeCompactionAt = Date.now(); markDirty(session); diff --git a/tests/silent-backend-nudge.test.ts b/tests/silent-backend-nudge.test.ts new file mode 100644 index 00000000..d2ea527d --- /dev/null +++ b/tests/silent-backend-nudge.test.ts @@ -0,0 +1,223 @@ +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 { _resetSessionsForTest } from "../src/session.ts"; + +// #728: upstreams that NEVER report usage (ChatGPT-login backends — their +// response.completed carries no usage.input_tokens) leave lastInputTokens == 0 +// for the whole session, and the kernel's decideNudge is structurally +// unfireable at tokenCount == 0 (growth reference falls back to tokenCount +// itself → growth ≡ 0; mass-ready and pressure bands all require usage at or +// above their pct lines). Context then grows unbounded until the hard limit — +// incident #726 sat at ~1.32M tokens before preflight finally kicked in. Fix +// (host-side, refined option 1 from the issue triage): prepare* records the +// PREVIOUS turn's LOCAL outbound payload upper bound +// (session.stats.localInputEstimate) each turn, and effectiveTokenCount feeds +// IT — capped by this request's inbound upper bound — only while +// lastInputTokens == 0. Real usage always takes precedence (the estimator only +// errs early, mirroring #604's armFailureShrink exception); the value +// self-corrects after every fold because the post-fold outbound payload +// shrinks. These e2e pins (explicit identity, Anthropic wire): +// A. silent backend, multi-turn growth → turn 1 idle (nothing measured yet, +// byte-identical to pre-fix first-turn behavior), turn 2 NUDGES once the +// recorded estimate crosses the kernel thresholds; preflight stays silent +// (optimistic estimate under the window). +// B. same conversation but upstream reports real usage every turn → NO +// nudge anywhere (real usage always beats the local estimate). +// C. stale-high cap: after A's nudged turn 2, turn 3 sends a SHRUNKEN +// history → no nudge (the previous turn's high estimate must not outlive +// the content it was measured from). +// The anonymous-prefix-affinity regression lives in fork-nudge-trigger.test.ts +// (untouched branch — that regime keeps feeding the raw inbound upper bound). + +const NUDGE_MARKER = "Context limit reached"; +const WINDOW = 60_000; + +const sseLine = (event: string, data: unknown): string => + `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; + +// message_start/message_delta omit `usage` entirely when inputTokens is null — +// the ChatGPT-login wire shape (field absent, not zero). +function okSse(inputTokens: number | null): string { + const startMsg = inputTokens == null + ? { type: "message_start", message: { id: "m1", role: "assistant" } } + : { type: "message_start", message: { id: "m1", role: "assistant", usage: { input_tokens: inputTokens } } }; + const deltaObj = inputTokens == null + ? { type: "message_delta", delta: { stop_reason: "end_turn", stop_sequence: null } } + : { type: "message_delta", delta: { stop_reason: "end_turn", stop_sequence: null }, usage: { output_tokens: 3 } }; + return ( + sseLine("message_start", startMsg) + + sseLine("content_block_start", { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } }) + + sseLine("content_block_delta", { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "ok" } }) + + sseLine("content_block_stop", { type: "content_block_stop", index: 0 }) + + sseLine("message_delta", deltaObj) + + sseLine("message_stop", { type: "message_stop" }) + ); +} + +function msgText(m: { content?: unknown }): string { + const c = m.content; + if (typeof c === "string") return c; + if (!Array.isArray(c)) return ""; + return c + .map((b) => (b && typeof b === "object" && typeof (b as { text?: unknown }).text === "string" ? (b as { text: string }).text : "")) + .join(""); +} + +function msgsOf(raw: string): Array<{ role?: string; content?: unknown }> { + try { + const parsed = JSON.parse(raw) as { messages?: Array<{ role?: string; content?: unknown }> }; + return parsed.messages ?? []; + } catch { + return []; + } +} + +// Same shape as fork-nudge-trigger.test.ts: 7 OLD code-heavy messages (~5.5k +// chars each) + 5 tiny recent ones. Char-count upper bound ≈ 54k/60k ≈ 90% of +// the window (over-limit for decideNudge), optimistic chars/4 estimate well +// under it (preflight stays silent). +const LINE = (i: number) => + `const handler_${i} = (req: Request, res: Response) => { res.status(200).json({ status: "ok", id: ${i}, ts: Date.now() }); };`; +const HEAVY = (i: number) => `CODE_${i}_` + LINE(i).repeat(62); + +function baseConversation(): Array<{ role: string; content: string }> { + const msgs: Array<{ role: string; content: string }> = []; + for (let i = 0; i < 12; i++) { + msgs.push({ role: i % 2 === 0 ? "user" : "assistant", content: i < 7 ? HEAVY(i) : `CODE_${i}_tiny note ${i}` }); + } + return msgs; +} + +async function runCase(opts: { inputTokens: number | null }): Promise> { + const streamed: string[] = []; + let nonStream = 0; + 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) { + streamed.push(raw); + res.writeHead(200, { "content-type": "text/event-stream" }); + res.end(okSse(opts.inputTokens)); + } else { + nonStream++; + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ + id: "msg_summary", + type: "message", + role: "assistant", + model: "claude-small", + content: [{ type: "text", text: "SUMMARY TEXT" }], + stop_reason: "end_turn", + usage: { input_tokens: 500, output_tokens: 50 }, + })); + } + }); + }); + upstream.listen(0, "127.0.0.1"); + await once(upstream, "listening"); + const upstreamPort = upstream.address().port; + + _setStoreForTest(new SessionStore({ enabled: false })); + _resetSessionsForTest(); + 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}`]: { models: { "claude-small": { context: WINDOW } } } }, + modelContextLimit: 400_000, + kernelConfig: defaultConfig(400_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; + + try { + const post = async (messages: Array<{ role: string; content: string }>): Promise => { + const resp = await fetch(`http://127.0.0.1:${proxyPort}/bili/http://127.0.0.1:${upstreamPort}/v1/messages`, { + method: "POST", + headers: { "content-type": "application/json", "x-acp-session": "silent-backend-sess" }, + body: JSON.stringify({ model: "claude-small", max_tokens: 1024, stream: true, messages }), + }); + await resp.text(); + return resp.status; + }; + + // Multi-turn: the client re-sends its FULL growing history every turn + // (the proxy-mode wire contract). Turn 1 = base conversation; turn 2 + // adds an assistant reply + one heavy follow-up user message (~same + // scale); turn 3 SHRINKS to the recent small tail (a client-side + // compaction/edit — the stale-estimate regime). + const base = baseConversation(); + const statuses: number[] = []; + statuses.push(await post(base)); + statuses.push(await post([ + ...base, + { role: "assistant", content: "ok, done with step one." }, + { role: "user", content: HEAVY(99) + " now step two" }, + ])); + statuses.push(await post([ + { role: "user", content: "short recent slice one" }, + { role: "assistant", content: "ok fine" }, + { role: "user", content: "tiny question only" }, + ])); + return { statuses, streamed, nonStream }; + } finally { + proxy.close(); + upstream.close(); + } +} + +test("#728A: silent-backend explicit session — turn 1 idle, turn 2 nudge via local estimate, preflight silent", async () => { + const { statuses, streamed, nonStream } = await runCase({ inputTokens: null }); + assert.deepEqual(statuses, [200, 200, 200]); + assert.equal(nonStream, 0, "preflight must stay silent (optimistic estimate under window)"); + assert.equal(streamed.length, 3); + assert.ok(!streamed[0].includes(NUDGE_MARKER), "turn 1 must stay idle — nothing measured yet (pre-fix first-turn behavior)"); + const fwd = msgsOf(streamed[1]); + const last = fwd.at(-1)!; + assert.ok( + last.role === "user" && msgText(last).includes(NUDGE_MARKER), + `turn 2 must carry the trailing nudge once the recorded local estimate crosses the threshold, got: ${streamed[1].slice(-800)}`, + ); + for (let i = 0; i < 12; i++) { + assert.ok(streamed[1].includes(`CODE_${i}_`), `CODE_${i}_ must survive un-folded (nudge is advisory, no fold happened)`); + } +}); + +test("#728B: real usage always takes precedence — reported input_tokens beats the local estimate", async () => { + const { statuses, streamed } = await runCase({ inputTokens: 2000 }); + assert.deepEqual(statuses, [200, 200, 200]); + for (let i = 0; i < streamed.length; i++) { + assert.ok(!streamed[i].includes(NUDGE_MARKER), `turn ${i + 1}: reported usage (2000 << ${WINDOW}) must drive tokenCount, not the ~90% payload estimate`); + } +}); + +test("#728C: stale-high cap — shrunk history does not re-trigger on the previous turn's estimate", async () => { + const { statuses, streamed } = await runCase({ inputTokens: null }); + assert.deepEqual(statuses, [200, 200, 200]); + assert.ok(streamed[1].includes(NUDGE_MARKER), "precondition: turn 2 nudged (see #728A)"); + assert.ok(!streamed[2].includes(NUDGE_MARKER), "turn 3 (shrunk history) must not nudge — the previous turn's high estimate is capped by this request's inbound upper bound"); +});