Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions reference/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ billion-context/
│ │ ├── headers.ts # Client-provided id sanitization for dump filenames (#286)
│ │ ├── inject.ts # Per-wire tool-injection wrappers + FORCE_TEXT_PROTOCOL switch (#1440)
│ │ ├── observability.ts # Body dumps (dumps/req-*.json, raw/*), unrecognized-path stats
│ │ ├── outbound-prefix.ts # Outbound-prefix LCP triage line: per-message fingerprints vs previous turn localize prompt-cache breaks (#5385)
│ │ ├── prepare-anthropic.ts # Anthropic request preparation (fold → rebuild → system/nudge injection) (#1440)
│ │ ├── prepare-google.ts # Gemini request preparation incl. :countTokens twin (#1440)
│ │ ├── prepare-openai.ts # OpenAI chat request preparation (#1440)
Expand Down
81 changes: 81 additions & 0 deletions src/server/outbound-prefix.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
import { bodyDumpEnabled } from "../knobs.js";

// #5385: observation-only instrument for prompt-cache-break triage. Anthropic
// caching is explicit and PREFIX-based, so when a rebuild changes a body byte
// ahead of the tail (e.g. a mid-history fold) the stable prefix shrinks and the
// next request re-caches from the first changed message to the tail — the exact
// symptom #5385 reports (cache_read pinned at the system+tools prefix while
// cache_creation grows with session length). Each turn we compare the rebuilt
// outbound message list against the PREVIOUS turn's by per-message fingerprint
// and report the longest common prefix + what sits at the boundary + where the
// cache_control breakpoints landed. Reads the rebuilt body, NEVER touches the
// wire bytes. Gated behind diagnostics.dumpBody / ACP_DUMP_BODY (off by default)
// so there is zero overhead unless you are already inspecting bodies; levels:
// debug every turn, info exactly on shrinkage (the cache-break event), landing
// in the same bili.log users send for triage.

function djb2(s: string): string {
let h = 5381;
for (let i = 0; i < s.length; i++) h = ((h << 5) + h + s.charCodeAt(i)) >>> 0;
return h.toString(36);
}

/** Compact order-stable fingerprint of one outbound wire message: role, content
* shape, block-type run, thinking/cache-control presence, total text length and
* a content hash. Equal fingerprints ⇒ equal bytes for caching purposes. */
function fingerprint(m: unknown): string {
const msg = m as { role?: unknown; content?: unknown };
const role = typeof msg.role === "string" ? msg.role : "?";
const c = msg.content;
if (typeof c === "string") {
return `${role}|str|len=${c.length}|h=${djb2(c)}`;
}
if (!Array.isArray(c)) {
return `${role}|${c === undefined ? "nil" : typeof c}`;
}
let textLen = 0;
let think = false;
let cc = false;
const types: string[] = [];
for (const b of c as Array<Record<string, unknown>>) {
const t = typeof b?.type === "string" ? b.type : "?";
types.push(t);
if (t === "thinking" || t === "redacted_thinking") think = true;
if (b?.cache_control !== undefined) cc = true;
if (typeof b?.text === "string") textLen += b.text.length;
else if (typeof b?.thinking === "string") textLen += b.thinking.length;
else if (b?.input !== undefined || b?.content !== undefined) textLen += JSON.stringify(b.input ?? b.content ?? "").length;
}
return `${role}|blk|types=[${types.join(",")}]|think=${think ? 1 : 0}|cc=${cc ? 1 : 0}|len=${textLen}|h=${djb2(JSON.stringify(c))}`;
}

export function observeOutboundPrefix(
sessionId: string,
messages: ReadonlyArray<unknown>,
meta: Record<string, unknown>,
log: (level: string, msg: string) => void,
): void {
if (!bodyDumpEnabled()) return;
try {
const cur = messages.map(fingerprint);
const key = "outboundPrefixFingerprint";
const prev = Array.isArray(meta[key]) ? (meta[key] as string[]) : null;
meta[key] = cur;
if (!prev || prev.length === 0) return;
const n = Math.min(prev.length, cur.length);
let lcp = 0;
while (lcp < n && prev[lcp] === cur[lcp]) lcp++;
// A mid-history byte change vs last turn: the whole previous list no
// longer matches from some index on (a healthy append keeps lcp == prev).
const shrank = lcp < prev.length;
const ccIdx: number[] = [];
for (let i = 0; i < cur.length; i++) if (cur[i].includes("|cc=1")) ccIdx.push(i);
const detail = shrank
? ` BREAK@${lcp} prev[${prev.length}]="${prev[lcp] ?? "<end>"}" -> cur[${cur.length}]="${cur[lcp] ?? "<end>"}"`
: "";
log(shrank ? "info" : "debug", `[${sessionId}] outbound-prefix: lcp=${lcp}/${cur.length} (prev ${prev.length}) cc@[${ccIdx.join(",")}]${detail} (#5385)`);
} catch (err) {
// observation-only: never disturb the request path
log("warn", `[${sessionId}] outbound-prefix: observation failed: ${String(err)} (#5385)`);
}
}
2 changes: 2 additions & 0 deletions src/server/prepare-anthropic.ts
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ import { MSG_IDS_BODY_FIELD, MSG_SEQS_BODY_FIELD, applyMsgIdPin, applyMsgSeqStam
import { collectRequestIds, probeRequestMembership } from "../request-membership.js";
import { alignIdSet, looksLikeSummaryCarrier } from "../msgid-algebra.js";
import { yieldToEventLoop } from "../yield-loop.js";
import { observeOutboundPrefix } from "./outbound-prefix.js";

export async function prepareAnthropic(
parsed: AnthropicRequestBody,
Expand Down Expand Up @@ -629,6 +630,7 @@ export async function prepareAnthropic(
// trailing-user tail #5604 established stays last.
const healedMessages = healAnthropicSystemPositions(dedupedMessages);
warnAnthropicThinkingPairs(parsed.messages, healedMessages, log, sessionId);
observeOutboundPrefix(sessionId, healedMessages, session.metadata, log);
const rebuilt: AnthropicRequestBody = { ...parsed, messages: healedMessages, system: systemOut, tools: toolsOut };
// prompt_cache_key is the omp plugin's session id stamped for the proxy's
// identity chain (#268), not part of the Anthropic Messages API — strip it
Expand Down
103 changes: 103 additions & 0 deletions tests/issue5385-outbound-prefix-diff.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
// #5385: outbound-prefix (LCP) triage instrument — both gate directions.
// Fires (info) exactly when the stable prefix shrinks vs the previous turn
// (a mid-history byte change); stays quiet (debug) on a healthy tail append.
import assert from "node:assert/strict";
import test from "node:test";
import { observeOutboundPrefix } from "../src/server/outbound-prefix.ts";

type Log = { level: string; msg: string };
const capture = (): Log[] => [];
const makeLog = (out: Log[]) => (level: string, msg: string) => out.push({ level, msg });

const m0 = () => ({ role: "user", content: "hello" });
const m1 = () => ({ role: "assistant", content: [{ type: "text", text: "hi there" }] });
const m2 = () => ({ role: "user", content: [{ type: "tool_result", tool_use_id: "t1", content: "ok" }, { type: "text", text: "next", cache_control: { type: "ephemeral" } }] });

process.env.ACP_DUMP_BODY = "1";

test("disabled by default: logs nothing and writes no session state", () => {
const prev = process.env.ACP_DUMP_BODY;
delete process.env.ACP_DUMP_BODY;
try {
const out: Log[] = [];
const meta: Record<string, unknown> = {};
observeOutboundPrefix("s0", [m0(), m1()], meta, makeLog(out));
assert.equal(out.length, 0);
assert.equal(meta["outboundPrefixFingerprint"], undefined, "off ⇒ no session-state write");
} finally {
if (prev === undefined) delete process.env.ACP_DUMP_BODY;
else process.env.ACP_DUMP_BODY = prev;
}
});

test("first turn stores fingerprints and emits nothing", () => {
const out: Log[] = [];
const meta: Record<string, unknown> = {};
const before = JSON.stringify([m0(), m1()]);
observeOutboundPrefix("s1", [m0(), m1()], meta, makeLog(out));
assert.equal(out.length, 0);
assert.ok(Array.isArray(meta["outboundPrefixFingerprint"]), "stores the fingerprint list");
assert.equal(JSON.stringify([m0(), m1()]), before, "does not mutate the wire");
});

test("healthy tail append keeps the whole prefix: debug, no BREAK", () => {
const out: Log[] = [];
const meta: Record<string, unknown> = {};
const log = makeLog(out);
observeOutboundPrefix("s2", [m0(), m1()], meta, log);
observeOutboundPrefix("s2", [m0(), m1(), m2()], meta, log);
assert.equal(out.length, 1, "only the second turn logs");
assert.equal(out[0].level, "debug", "append is NOT a shrinkage event");
assert.match(out[0].msg, /lcp=2\/3 \(prev 2\)/);
assert.doesNotMatch(out[0].msg, /BREAK/);
});

test("mid-history byte change at the head: info, BREAK@0", () => {
const out: Log[] = [];
const meta: Record<string, unknown> = {};
const log = makeLog(out);
observeOutboundPrefix("s3", [m0(), m1(), m2()], meta, log);
observeOutboundPrefix("s3", [{ role: "user", content: "hello CHANGED" }, m1(), m2()], meta, log);
assert.equal(out.length, 1);
assert.equal(out[0].level, "info", "shrinkage escalates to info so it lands in the default bili.log");
assert.match(out[0].msg, /lcp=0\/3 \(prev 3\)/);
assert.match(out[0].msg, /BREAK@0/);
});

test("break at index 1 reports that boundary and the cache_control positions", () => {
const out: Log[] = [];
const meta: Record<string, unknown> = {};
const log = makeLog(out);
observeOutboundPrefix("s4", [m0(), m1(), m2()], meta, log);
observeOutboundPrefix("s4", [m0(), { role: "assistant", content: [{ type: "text", text: "reworded" }] }, m2()], meta, log);
assert.equal(out.length, 1);
assert.equal(out[0].level, "info");
assert.match(out[0].msg, /lcp=1\/3 \(prev 3\)/);
assert.match(out[0].msg, /BREAK@1/);
assert.match(out[0].msg, /cc@\[2\]/, "reports where the current outbound carries cache_control");
});

test("same-length re-encoding with different bytes still breaks (content hash)", () => {
const out: Log[] = [];
const meta: Record<string, unknown> = {};
const log = makeLog(out);
observeOutboundPrefix("s5", [m0(), m1()], meta, log);
// same length "hi there" -> "hi there!" minus one char elsewhere; force a byte diff of equal length
observeOutboundPrefix("s5", [m0(), { role: "assistant", content: [{ type: "text", text: "hi therX" }] }], meta, log);
assert.equal(out.length, 1);
assert.equal(out[0].level, "info");
assert.match(out[0].msg, /BREAK@1/);
});

test("history shrinkage reports the break boundary with an end sentinel", () => {
const out: Log[] = [];
const meta: Record<string, unknown> = {};
const log = makeLog(out);
observeOutboundPrefix("s6", [m0(), m1(), m2()], meta, log);
observeOutboundPrefix("s6", [m0(), m1()], meta, log);
assert.equal(out.length, 1);
assert.equal(out[0].level, "info");
assert.match(out[0].msg, /lcp=2\/2 \(prev 3\)/);
assert.match(out[0].msg, /BREAK@2/);
assert.match(out[0].msg, /<end>/, "the shorter list reads as <end> at the break boundary");
});
Loading