fix(cache): keep the per-turn billing line out of the cached prefix (+3 attribution fixes)

src/core/proxy.ts

@@ -1365,6 +1365,11 @@ export function createProxy(config: ProxyConfig = {}) {
     }
 
     const outHeaders = filterHeaders(req.headers, STRIP_REQ_HEADERS);
+    // Claude Code folds `x-anthropic-billing-header: ...` into its system text
+    // and it carries a per-turn request id; transform lifts it out of the
+    // cached prefix and hands it back here, so it travels as a header (which
+    // is what its own name says it is) and stops busting the prompt cache.
+    if (info?.billingHeader) outHeaders.set('x-anthropic-billing-header', info.billingHeader);
     if (isOpenAIPath || bridgedGptMessages || bridgedChatMessages) {
       outHeaders.delete('x-api-key');
       // Never forward a Messages client's bearer credential across providers.

src/core/tracker.ts

@@ -98,6 +98,24 @@ export interface TrackEvent {
   cache_prefix_sha8?: string;
   /** Approx chars in that pinned prefix (growth vs pure-invalidation split). */
   cache_prefix_bytes?: number;
+  /** Per-layer digests of the same pinned prefix. Whichever one moves between
+   *  two turns of a session IS the cache-bust cause: tools (client loaded a
+   *  deferred tool), system (volatile text inside the pinned span), head
+   *  (collapse boundary / marker placement moved). */
+  cache_prefix_tools_sha8?: string;
+  cache_prefix_system_sha8?: string;
+  /** Narrows a system-layer bust to the block that moved: a text-only digest
+   *  (ignores the block framing) plus one `index[*]:len:hash4` fingerprint per
+   *  block, where `*` marks a cache_control anchor. Content never lands here. */
+  cache_prefix_system_text_sha8?: string;
+  cache_prefix_system_blocks?: string[];
+  cache_prefix_head_sha8?: string;
+  /** The span Anthropic really caches (through the last cache_control marker),
+   *  its size, and the marker's position. Unstable marked digest ⇒ pxpipe-side
+   *  bust; stable digest with cache_read 0 ⇒ look upstream, not at the rewrite. */
+  cache_prefix_marked_sha8?: string;
+  cache_prefix_marked_bytes?: number;
+  cache_prefix_marker_pos?: string;
 
   // From TransformInfo.env:
   cwd?: string;
@@ -278,6 +296,15 @@ export function toTrackEvent(ev: ProxyEvent): TrackEvent {
     }
     if (info.cachePrefixSha8) out.cache_prefix_sha8 = info.cachePrefixSha8;
     if (info.cachePrefixBytes !== undefined) out.cache_prefix_bytes = info.cachePrefixBytes;
+    if (info.cachePrefixToolsSha8) out.cache_prefix_tools_sha8 = info.cachePrefixToolsSha8;
+    if (info.cachePrefixSystemSha8) out.cache_prefix_system_sha8 = info.cachePrefixSystemSha8;
+    if (info.cachePrefixHeadSha8) out.cache_prefix_head_sha8 = info.cachePrefixHeadSha8;
+    if (info.cachePrefixMarkedSha8) out.cache_prefix_marked_sha8 = info.cachePrefixMarkedSha8;
+    if (info.cachePrefixMarkedBytes !== undefined)
+      out.cache_prefix_marked_bytes = info.cachePrefixMarkedBytes;
+    if (info.cachePrefixMarkerPos) out.cache_prefix_marker_pos = info.cachePrefixMarkerPos;
+    if (info.cachePrefixSystemTextSha8) out.cache_prefix_system_text_sha8 = info.cachePrefixSystemTextSha8;
+    if (info.cachePrefixSystemBlocks) out.cache_prefix_system_blocks = info.cachePrefixSystemBlocks;
     if (info.unknownStaticTags && info.unknownStaticTags.length > 0)
       out.unknown_static_tags = info.unknownStaticTags;
     if (info.churningStaticTags && info.churningStaticTags.length > 0)

src/dashboard/fragments.ts

@@ -766,7 +766,7 @@ export function renderStatsTableFragment(p: FullStatsPayload): string {
   const totalIn = (s.inputTokensTotal || 0) + (s.cacheCreateTokensTotal || 0) + (s.cacheReadTokensTotal || 0);
   const hitRateTok = totalIn > 0 ? ((s.cacheReadTokensTotal / totalIn) * 100).toFixed(1) + '%' : '-';
   const hitRateEv =
-    s.eventsWithBaseline > 0 ? ((s.cacheHitEvents / s.eventsWithBaseline) * 100).toFixed(1) + '%' : '-';
+    s.eventsWithUsage > 0 ? ((s.cacheHitEvents / s.eventsWithUsage) * 100).toFixed(1) + '%' : '-';
   const charRatio =
     s.origCharsTotal > 0 ? ((s.imageBytesTotal / s.origCharsTotal) * 100).toFixed(3) + 'x' : '-';
 

src/dashboard/types.ts

@@ -132,7 +132,9 @@ export interface FullStatsSummary {
   cacheReadTokensTotal: number;
   outputTokensTotal: number;
   cacheHitEvents: number;
-  eventsWithBaseline: number;
+  // Denominator for the event-based cache-hit rate: events that carried usage
+  // data at all. Emitted by stats.ts as `eventsWithUsage`.
+  eventsWithUsage: number;
   origCharsTotal: number;
   imageBytesTotal: number;
   pinCharsTotal?: number;

tests/cache-bust-attribution.test.ts

@@ -0,0 +1,191 @@
+/**
+ * Cache-bust ATTRIBUTION telemetry (#11).
+ *
+ * When cache_read collapses in production, these fields are the only evidence
+ * available after the fact. Two contracts:
+ *
+ *  1. `history_image_sha8` must describe the HISTORY images. On a Claude Code
+ *     request the slab anchor makes `messages[0]` the protected slab message,
+ *     not the synthetic history message — hashing index 0 silently reported
+ *     slab stability under the history name, so a drifting collapse boundary
+ *     looked stable in telemetry.
+ *  2. The pinned prefix must be digested per LAYER (tools / system / imaged
+ *     head), because "the prefix changed" does not say which layer moved, and
+ *     the three fail for different reasons and need different fixes.
+ *
+ * Run just this file:  pnpm vitest run tests/cache-bust-attribution.test.ts
+ */
+import { describe, expect, it } from 'vitest';
+import { transformRequest } from '../src/core/transform.js';
+import { HISTORY_SYNTHETIC_INTRO } from '../src/core/history.js';
+import type { Message } from '../src/core/types.js';
+
+const big = (n: number) => 'x'.repeat(n);
+const enc = (obj: unknown) => new TextEncoder().encode(JSON.stringify(obj));
+const dec = (b: Uint8Array): any => JSON.parse(new TextDecoder().decode(b));
+
+/** N closed plain turns — long enough that the collapse gate accepts. */
+function convo(n: number, chars = 3500): Message[] {
+  const out: Message[] = [];
+  for (let i = 0; i < n; i++) {
+    const body = `turn ${i}: ` + big(chars);
+    out.push({ role: i % 2 === 0 ? 'user' : 'assistant', content: body });
+  }
+  return out;
+}
+
+/** A Claude-Code-shaped request: big marked system slab + a long conversation. */
+function ccBody(opts: { turns?: number; tools?: unknown[]; sysSuffix?: string } = {}) {
+  return enc({
+    model: 'claude-3-5-sonnet',
+    system: [
+      {
+        type: 'text',
+        text: big(80_000) + (opts.sysSuffix ?? ''),
+        cache_control: { type: 'ephemeral' },
+      },
+    ],
+    ...(opts.tools ? { tools: opts.tools } : {}),
+    messages: convo(opts.turns ?? 15),
+  });
+}
+
+/** Concatenated base64 of the image blocks on the synthetic history message. */
+function historyImageData(out: Uint8Array): string {
+  const body = dec(out);
+  const synthetic = (body.messages ?? []).find(
+    (m: any) => Array.isArray(m.content) && m.content[0]?.type === 'text' && m.content[0].text === HISTORY_SYNTHETIC_INTRO,
+  );
+  if (!synthetic) return '';
+  return synthetic.content
+    .filter((b: any) => b?.type === 'image')
+    .map((b: any) => b.source.data)
+    .join('');
+}
+
+/** Concatenated base64 of the image blocks on the FIRST message (the slab). */
+function slabImageData(out: Uint8Array): string {
+  const first = dec(out).messages?.[0];
+  if (!first || !Array.isArray(first.content)) return '';
+  return first.content
+    .filter((b: any) => b?.type === 'image')
+    .map((b: any) => b.source.data)
+    .join('');
+}
+
+describe('cache-bust attribution telemetry', () => {
+  it('history_image_sha8 tracks the history images, not the slab message at index 0', async () => {
+    const { body: out, info } = await transformRequest(ccBody());
+    // Precondition: this really is the Claude Code shape — a slab message ahead
+    // of the synthetic history message, both carrying images.
+    const slab = slabImageData(out);
+    const history = historyImageData(out);
+    expect(info.collapsedTurns).toBeGreaterThan(0);
+    expect(slab.length).toBeGreaterThan(0);
+    expect(history.length).toBeGreaterThan(0);
+    expect(history).not.toBe(slab);
+
+    // The reported hash must be a function of the HISTORY images. Prove it by
+    // changing only the history (more collapsed turns) and requiring the hash
+    // to move, while the slab bytes stay identical.
+    const { body: out2, info: info2 } = await transformRequest(ccBody({ turns: 65 }));
+    expect(slabImageData(out2)).toBe(slab); // slab unchanged …
+    expect(historyImageData(out2)).not.toBe(history); // … history changed …
+    expect(info2.historyImageSha).not.toBe(info.historyImageSha); // … so must the hash
+  });
+
+  it('digests the pinned prefix per layer (tools / system / head)', async () => {
+    const { info } = await transformRequest(ccBody());
+    expect(info.cachePrefixSha8).toBeDefined();
+    expect(info.cachePrefixToolsSha8).toBeDefined();
+    expect(info.cachePrefixSystemSha8).toBeDefined();
+    expect(info.cachePrefixHeadSha8).toBeDefined();
+  });
+
+  it('moves ONLY the tools digest when the client adds a tool', async () => {
+    const toolsA = [{ name: 'Read', description: 'Read a file. ' + big(400), input_schema: { type: 'object' } }];
+    const toolsB = [
+      ...toolsA,
+      { name: 'Grep', description: 'Search. ' + big(400), input_schema: { type: 'object' } },
+    ];
+    const a = await transformRequest(ccBody({ tools: toolsA }));
+    const b = await transformRequest(ccBody({ tools: toolsB }));
+    expect(b.info.cachePrefixToolsSha8).not.toBe(a.info.cachePrefixToolsSha8);
+    expect(b.info.cachePrefixSystemSha8).toBe(a.info.cachePrefixSystemSha8);
+    expect(b.info.cachePrefixSha8).not.toBe(a.info.cachePrefixSha8);
+  });
+
+  it('digests the MARKED span (what Anthropic caches) and records the marker position', async () => {
+    const { info } = await transformRequest(ccBody());
+    expect(info.cachePrefixMarkedSha8).toBeDefined();
+    expect(info.cachePrefixMarkerPos).toMatch(/^m\d+\.b\d+$/);
+    // The marked span ends at the breakpoint, so it is a strict subset of the
+    // boundary-scoped prefix (which runs to the end of the history message).
+    expect(info.cachePrefixMarkedBytes!).toBeLessThanOrEqual(info.cachePrefixBytes!);
+  });
+
+  it('keeps the marked span byte-identical while the live tail grows', async () => {
+    // The contract that decides whether cache_read happens: two consecutive
+    // turns of one session must send the SAME bytes up to the breakpoint. The
+    // newest freeze chunk re-renders by design, so a boundary-scoped digest may
+    // legitimately move — the marked one may not.
+    const a = await transformRequest(ccBody({ turns: 15 }));
+    const b = await transformRequest(ccBody({ turns: 17 }));
+    expect(b.info.collapsedTurns).toBe(a.info.collapsedTurns); // same collapse window
+    expect(b.info.cachePrefixMarkerPos).toBe(a.info.cachePrefixMarkerPos);
+    expect(b.info.cachePrefixMarkedSha8).toBe(a.info.cachePrefixMarkedSha8);
+  });
+
+  it('moves the system digest when volatile text rides inside the pinned span', async () => {
+    // `# Environment` churns every turn (cwd, git status, model id). Whatever
+    // survives as system TEXT is inside the pinned prefix and must show up as a
+    // system-layer bust — that is the signal a live capture needs.
+    const a = await transformRequest(ccBody({ sysSuffix: '\n# Environment\ngit status: clean\n' }));
+    const b = await transformRequest(ccBody({ sysSuffix: '\n# Environment\ngit status: 3 files changed\n' }));
+    expect(b.info.cachePrefixSystemSha8).not.toBe(a.info.cachePrefixSystemSha8);
+    expect(b.info.cachePrefixToolsSha8).toBe(a.info.cachePrefixToolsSha8);
+  });
+});
+
+describe('billing line lift (per-turn id must never reach the prefix)', () => {
+  const billing =
+    'x-anthropic-billing-header: cc_version=2.1.220.589; cc_entrypoint=cli; cch=f9742; cc_prev_req=req_AAA;';
+  const billing2 = billing.replace('f9742', 'fa49f').replace('req_AAA', 'req_BBB');
+
+  /** Claude Code's real shape: big marked slab + the billing line as its own
+   *  UNMARKED trailing block. */
+  const trailingBlock = (line: string) =>
+    enc({
+      model: 'claude-3-5-sonnet',
+      system: [
+        { type: 'text', text: big(80_000), cache_control: { type: 'ephemeral' } },
+        { type: 'text', text: line },
+      ],
+      messages: convo(15),
+    });
+
+  it('strips the line when it arrives as an unmarked trailing block', async () => {
+    const { body, info } = await transformRequest(trailingBlock(billing));
+    const wire = new TextDecoder().decode(body);
+    expect(wire).not.toContain('x-anthropic-billing-header');
+    expect(wire).not.toContain('cc_prev_req');
+    expect(info.billingHeader).toContain('cc_prev_req=req_AAA');
+  });
+
+  it('strips it when glued onto the tail of the preceding system text', async () => {
+    const { body, info } = await transformRequest(ccBody({ sysSuffix: billing }));
+    expect(new TextDecoder().decode(body)).not.toContain('cc_prev_req');
+    expect(info.billingHeader).toContain('cc_prev_req=req_AAA');
+  });
+
+  it('two turns differing ONLY in the billing id keep an identical cached prefix', async () => {
+    const a = await transformRequest(trailingBlock(billing));
+    const b = await transformRequest(trailingBlock(billing2));
+    // Byte-for-byte identical upstream request — the only difference (the id)
+    // now travels as a header, outside the cached prefix.
+    expect(new TextDecoder().decode(b.body)).toBe(new TextDecoder().decode(a.body));
+    expect(b.info.cachePrefixMarkedSha8).toBe(a.info.cachePrefixMarkedSha8);
+    expect(b.info.cachePrefixSystemSha8).toBe(a.info.cachePrefixSystemSha8);
+    expect(b.info.billingHeader).not.toBe(a.info.billingHeader);
+  });
+});

tests/dashboard-api.test.ts

@@ -15,7 +15,11 @@ import { getAllowedModelBases, setAllowedModelBases } from '../src/core/applicab
 import type { SessionsPaths } from '../src/sessions.js';
 import type { TrackEvent } from '../src/core/tracker.js';
 import type { StatsPayload, RecentPayload } from '../src/dashboard/types.js';
-import { renderHeaderFragment, renderPage } from '../src/dashboard/fragments.js';
+import {
+  renderHeaderFragment,
+  renderPage,
+  renderStatsTableFragment,
+} from '../src/dashboard/fragments.js';
 
 function makeTmp(): SessionsPaths {
   const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'pxpipe-dashapi-'));
@@ -1004,3 +1008,35 @@ describe('server-observed warmth: text follows actual cache_read', () => {
     expect(row.session_saved_so_far_delta).toBe(9900);
   });
 });
+
+/**
+ * Regression: the dashboard's "cache hit rate (by events)" row read a summary
+ * field (`eventsWithBaseline`) that stats.ts never emitted — it emits
+ * `eventsWithUsage`. The field existed in the payload *type*, so tsc stayed
+ * quiet, and the row silently rendered "-" forever. This test walks the real
+ * path (fold -> summaryToJson -> renderStatsTableFragment) so any future
+ * rename on either side fails here instead of blanking the dashboard.
+ */
+describe('stats table: event-based cache hit rate', () => {
+  it('renders a real percentage from the summary stats.ts actually emits', async () => {
+    const { newSummary, fold, summaryToJson } = await import('../src/stats.js');
+    let s = newSummary();
+    // 3 events with usage, 2 of them cache hits -> 66.7%
+    s = fold(s, { input_tokens: 100, cache_read_tokens: 900 } as TrackEvent);
+    s = fold(s, { input_tokens: 100, cache_read_tokens: 900 } as TrackEvent);
+    s = fold(s, { input_tokens: 100, cache_read_tokens: 0 } as TrackEvent);
+    // an event carrying no usage at all must not dilute the denominator
+    s = fold(s, { cwd: '/tmp' } as TrackEvent);
+
+    const summary = summaryToJson(s);
+    expect(summary.eventsWithUsage).toBe(3);
+    expect(summary.cacheHitEvents).toBe(2);
+
+    const html = renderStatsTableFragment({ summary } as never);
+    // Pin the value to its own row: the table is a single line, so a bare
+    // "contains" would also pass on some other row's number.
+    const evRow = /<td>cache hit \(by events\)<\/td><td class="num">([^<]*)<\/td>/.exec(html);
+    expect(evRow, 'the "cache hit (by events)" row must exist').not.toBeNull();
+    expect(evRow![1]).toBe('66.7%'); // not '-', which is what the dead field produced
+  });
+});

tests/render.test.ts

@@ -1298,7 +1298,11 @@ describe('transform', () => {
     });
   });
 
-  it('strips x-anthropic-billing-header line and keeps it as text', async () => {
+  it('lifts the x-anthropic-billing-header line OUT of the prompt (header, not text)', async () => {
+    // Was: "keeps it as text". That placement zeroed the prompt cache — the
+    // line carries a fresh cc_prev_req id per turn and system[] precedes every
+    // cache_control marker, so each turn rewrote the whole prefix. It is a
+    // header the client folded into the system text; it now leaves as one.
     const sysText = 'x-anthropic-billing-header: cch=abc123\n' + 'real prompt text. '.repeat(2500);
     const req = JSON.stringify({
       model: 'claude-3-5-sonnet',
@@ -1308,10 +1312,31 @@ describe('transform', () => {
     const bytes = new TextEncoder().encode(req);
     const { body, info } = await transformRequest(bytes);
     expect(info.compressed).toBe(true);
+    expect(new TextDecoder().decode(body)).not.toContain('x-anthropic-billing-header');
+    expect(info.billingHeader).toBe('cch=abc123');
+  });
 
-    const out = JSON.parse(new TextDecoder().decode(body));
-    const textBlocks = out.system.filter((b: any) => b.type === 'text');
-    expect(textBlocks.some((b: any) => b.text.includes('x-anthropic-billing-header'))).toBe(true);
+  it('PXPIPE_KEEP_BILLING_LINE=1 restores the old in-prompt placement', async () => {
+    const sysText = 'x-anthropic-billing-header: cch=abc123\n' + 'real prompt text. '.repeat(2500);
+    const bytes = new TextEncoder().encode(
+      JSON.stringify({
+        model: 'claude-3-5-sonnet',
+        messages: [{ role: 'user', content: 'hi' }],
+        system: sysText,
+      }),
+    );
+    const prev = process.env.PXPIPE_KEEP_BILLING_LINE;
+    process.env.PXPIPE_KEEP_BILLING_LINE = '1';
+    try {
+      const { body, info } = await transformRequest(bytes);
+      const out = JSON.parse(new TextDecoder().decode(body));
+      const textBlocks = (out.system ?? []).filter((b: any) => b.type === 'text');
+      expect(textBlocks.some((b: any) => b.text.includes('x-anthropic-billing-header'))).toBe(true);
+      expect(info.billingHeader).toBeUndefined();
+    } finally {
+      if (prev === undefined) delete process.env.PXPIPE_KEEP_BILLING_LINE;
+      else process.env.PXPIPE_KEEP_BILLING_LINE = prev;
+    }
   });
 
   it('keeps <env> as text outside the image so cache_control stays stable', async () => {