Merge pull request #865 from alvinunreal/omos/cache-safety-strategy-matrix

Close the two detection gaps that let the v2.2.5 checkpoint-board cache regression ship silently

docs/cache-verification.md

@@ -45,12 +45,23 @@ the Prompt Cache Safety section in `AGENTS.md` for the authoring rules.
 ## Runtime cache monitoring
 
 `src/hooks/cache-monitor/` observes `message.updated` events and the
-provider-reported `tokens.cache.read` / `tokens.cache.write` counters. When a
-session that previously hit the cache reports zero cache-read tokens on a
-sizeable request, it logs a `[cache-monitor] possible prompt-cache bust`
-warning (once per bust streak) to the plugin log. Providers that never report
-cache telemetry produce no warnings. This is the field safety net for
-provider-side behavior no offline test can model.
+provider-reported `tokens.cache.read` / `tokens.cache.write` counters. It
+logs two warning shapes (observation only, to the plugin log):
+
+- `possible prompt-cache bust` — a session that previously hit the cache
+  reports zero cache-read tokens on a sizeable request (once per bust
+  streak). The signature of a mid-session prompt-prefix change.
+- `never hit the provider cache` — a session reports zero cached tokens on
+  every sizeable request from its first turn, past both a consecutive-request
+  and a cumulative-uncached-input threshold (once per session). The signature
+  of a prefix that changes on *every* request — how the v2.2.5
+  checkpoint-board regression looked in the field. OpenCode coalesces missing
+  provider telemetry to zeros, so a cache-less provider is indistinguishable
+  from this; the thresholds and hedged wording keep that ambiguity from
+  becoming noise, and modest sessions on cache-less providers stay silent.
+
+This is the field safety net for provider-side behavior no offline test can
+model.
 
 ## Prerequisites
 

src/hooks/cache-monitor/index.test.ts

@@ -100,9 +100,12 @@ describe('createCacheMonitorHook', () => {
     expect(warnings).toHaveLength(2);
   });
 
-  test('stays silent for providers that never report cache tokens', async () => {
+  test('stays silent for modest sessions that never report cache tokens', async () => {
     const { hook, warnings } = createHarness();
 
+    // Cache-less providers are indistinguishable from busted sessions
+    // (OpenCode coalesces missing telemetry to zeros); below the cumulative
+    // input threshold the monitor must give them the benefit of the doubt.
     for (const id of ['c1', 'c2', 'c3']) {
       await hook.event(
         assistantMessageEvent({ messageID: id, input: 20000, cacheRead: 0 }),
@@ -112,6 +115,59 @@ describe('createCacheMonitorHook', () => {
     expect(warnings).toHaveLength(0);
   });
 
+  test('warns once for a large session that never hits the cache', async () => {
+    const { hook, warnings } = createHarness();
+
+    // The v2.2.5 checkpoint-board signature: consecutive ~146K-input
+    // requests, zero cache reads from the very first turn.
+    for (const id of ['g1', 'g2']) {
+      await hook.event(
+        assistantMessageEvent({ messageID: id, input: 146000, cacheRead: 0 }),
+      );
+    }
+    expect(warnings).toHaveLength(0);
+
+    await hook.event(
+      assistantMessageEvent({ messageID: 'g3', input: 146000, cacheRead: 0 }),
+    );
+    expect(warnings).toHaveLength(1);
+    expect(warnings[0].message).toContain('never hit the provider cache');
+    expect(warnings[0].data).toMatchObject({
+      sessionID: 'ses_monitor',
+      consecutiveUncachedRequests: 3,
+      uncachedInputTokens: 438000,
+    });
+
+    // Once per session, even as the streak keeps growing.
+    await hook.event(
+      assistantMessageEvent({ messageID: 'g4', input: 146000, cacheRead: 0 }),
+    );
+    expect(warnings).toHaveLength(1);
+  });
+
+  test('any reported cache activity disarms the never-cached warning', async () => {
+    const { hook, warnings } = createHarness();
+
+    // An Anthropic-style first request reports a cache write; later misses
+    // are the everReportedCache bust signature, not the never-cached one.
+    await hook.event(
+      assistantMessageEvent({
+        messageID: 'h1',
+        input: 146000,
+        cacheRead: 0,
+        cacheWrite: 140000,
+      }),
+    );
+    for (const id of ['h2', 'h3', 'h4']) {
+      await hook.event(
+        assistantMessageEvent({ messageID: id, input: 146000, cacheRead: 0 }),
+      );
+    }
+
+    expect(warnings).toHaveLength(1);
+    expect(warnings[0].message).toContain('prompt-cache bust');
+  });
+
   test('stays silent on the first request and on tiny prompts', async () => {
     const { hook, warnings } = createHarness();
 

src/hooks/cache-monitor/index.ts

@@ -25,11 +25,30 @@ const MIN_INPUT_TOKENS_FOR_WARNING = 2048;
 const MAX_TRACKED_SESSIONS = 256;
 const MAX_TRACKED_MESSAGES_PER_SESSION = 512;
 
+/**
+ * A session busted from its very first request never trips the
+ * `everReportedCache` warning below — that was the field signature of the
+ * v2.2.5 checkpoint board regression, where every request re-paid full
+ * input from turn one and the monitor stayed silent.
+ *
+ * OpenCode coalesces missing provider cache telemetry to zeros, so explicit
+ * zeros cannot distinguish "prefix changes every request" from "provider
+ * has no prompt cache". Both thresholds must be met before warning — at
+ * least this many consecutive sizeable zero-cache requests AND this much
+ * cumulative uncached input — so the warning only fires where a working
+ * cache would have saved a large amount, and the wording stays hedged.
+ */
+const NEVER_CACHED_STREAK_FOR_WARNING = 3;
+const NEVER_CACHED_INPUT_TOKENS_FOR_WARNING = 100_000;
+
 interface SessionCacheState {
   completedRequests: number;
   everReportedCache: boolean;
   lastCacheRead: number;
   warnedSinceLastHit: boolean;
+  neverCachedStreak: number;
+  neverCachedInputTokens: number;
+  neverCachedWarned: boolean;
   processedMessageIDs: Set<string>;
 }
 
@@ -118,6 +137,9 @@ export function createCacheMonitorHook(options: CacheMonitorOptions = {}) {
       everReportedCache: false,
       lastCacheRead: 0,
       warnedSinceLastHit: false,
+      neverCachedStreak: 0,
+      neverCachedInputTokens: 0,
+      neverCachedWarned: false,
       processedMessageIDs: new Set(),
     };
     sessions.set(sessionID, state);
@@ -152,6 +174,39 @@ export function createCacheMonitorHook(options: CacheMonitorOptions = {}) {
       );
     }
 
+    // A session that never serves a single cached token, over enough
+    // sizeable requests that a working cache would have saved a large
+    // amount, is busted from turn one — it never arms the
+    // everReportedCache warning above. Small requests neither extend nor
+    // reset the streak: they sit under provider minimum-prefix thresholds
+    // and legitimately miss.
+    if (!state.everReportedCache) {
+      if (
+        message.cacheRead === 0 &&
+        message.cacheWrite === 0 &&
+        message.inputTokens >= MIN_INPUT_TOKENS_FOR_WARNING
+      ) {
+        state.neverCachedStreak += 1;
+        state.neverCachedInputTokens += message.inputTokens;
+      }
+      if (
+        !state.neverCachedWarned &&
+        state.neverCachedStreak >= NEVER_CACHED_STREAK_FOR_WARNING &&
+        state.neverCachedInputTokens >= NEVER_CACHED_INPUT_TOKENS_FOR_WARNING
+      ) {
+        state.neverCachedWarned = true;
+        logger(
+          '[cache-monitor] session has never hit the provider cache: every sizeable request reported 0 cache-read tokens. If this provider supports prompt caching, the prompt prefix is likely changing on every request; if not, this session is re-paying full input each turn — see docs/cache-verification.md.',
+          {
+            sessionID: message.sessionID,
+            requestNumber: state.completedRequests,
+            consecutiveUncachedRequests: state.neverCachedStreak,
+            uncachedInputTokens: state.neverCachedInputTokens,
+          },
+        );
+      }
+    }
+
     if (message.cacheRead > 0) state.warnedSinceLastHit = false;
     state.everReportedCache =
       state.everReportedCache ||

src/hooks/cache-safety-harness.test.ts

@@ -31,6 +31,13 @@ export const FIXTURE_NOW = 1_700_000_000_000;
 
 export type TransformOutput = { messages: unknown[] };
 
+export type BoardStrategy = 'latest' | 'checkpoint-compatible';
+
+export interface PipelineOptions {
+  /** Board injection strategy under test; defaults to the production default. */
+  strategy?: BoardStrategy;
+}
+
 export interface Pipeline {
   run: (output: TransformOutput) => Promise<void>;
   markFileToolPending: () => void;
@@ -42,7 +49,7 @@ export interface Pipeline {
  * cache-safety.property.test.ts fails when the two fall out of sync — update
  * BOTH when adding, removing, or reordering a transform step.
  */
-export function createPipeline(): Pipeline {
+export function createPipeline(options: PipelineOptions = {}): Pipeline {
   const sessionAgentMap = new Map<string, string>();
   const board = new BackgroundJobBoard();
   const lifecycle = new SessionLifecycle(() => {});
@@ -67,6 +74,7 @@ export function createPipeline(): Pipeline {
     {
       maxSessionsPerAgent: 2,
       maxRetainedSnapshots: DEFAULT_MAX_RETAINED_SNAPSHOTS,
+      ...(options.strategy ? { strategy: options.strategy } : {}),
       backgroundJobBoard: board,
       shouldManageSession: (sessionID) =>
         sessionAgentMap.get(sessionID) === 'orchestrator',

src/hooks/cache-safety.property.test.ts

@@ -21,9 +21,11 @@
 import { afterEach, describe, expect, setSystemTime, test } from 'bun:test';
 import { readFileSync } from 'node:fs';
 import path from 'node:path';
+import { BackgroundJobsConfigSchema } from '../config';
 import { isVolatileTaggedMessage } from './cache-safe-injection';
 import {
   assistantTurn,
+  type BoardStrategy,
   buildHistory,
   createPipeline,
   FIXTURE_NOW,
@@ -34,49 +36,116 @@ import {
   turnEndIndices,
 } from './cache-safety-harness.test';
 import { BACKGROUND_JOB_BOARD_METADATA_KEY } from './task-session-manager';
+import type { MessageWithParts } from './types';
 
 afterEach(() => {
   setSystemTime();
 });
 
-describe('cache-safety: turn-over-turn prefix stability', () => {
-  test('re-rendering a growing conversation reproduces byte-identical history', async () => {
-    const pipeline = createPipeline();
-    const history = buildHistory();
-    const turns = turnEndIndices(history);
-
-    let previous: string[] | undefined;
-    for (const [turnNumber, endIndex] of turns.entries()) {
-      // Exercise cross-turn hook state: a file-tool nudge fires before the
-      // second turn, and background jobs churn (launch, then drop) while
-      // later turns render — none of it may touch stable bytes.
-      if (turnNumber === 1) pipeline.markFileToolPending();
-      if (turnNumber === 2) {
-        pipeline.board.registerLaunch({
-          taskID: 'task-alpha',
-          parentSessionID: SESSION_ID,
-          agent: 'explorer',
-          description: 'churn fixture',
-          now: FIXTURE_NOW,
-        });
-      }
-      if (turnNumber === 3) pipeline.board.drop('task-alpha');
+/**
+ * Per-strategy definition of "the bytes that must never be rewritten".
+ *
+ * - `latest`: board state lives in a single volatile trailing message that
+ *   is stripped and re-appended every request, so the stable prefix is
+ *   every non-volatile message.
+ * - `checkpoint-compatible`: board snapshots are append-only stable bytes
+ *   by design — that is the strategy's entire purpose — so the stable
+ *   prefix is the WHOLE provider-visible payload. Filtering tagged messages
+ *   here would hide exactly the snapshot drop/reinsert rewrite that shipped
+ *   in v2.2.5. Replayed snapshot messages are rebuilt each request from a
+ *   varying base message, so only provider-visible fields (role, agent,
+ *   parts) participate — `info` never reaches the provider.
+ *
+ * A new `BackgroundJobsConfigSchema` strategy must add an entry here (the
+ * drift guard below fails until it does), forcing an explicit decision
+ * about its cache-safety semantics before it can ship.
+ */
+const STRATEGY_STABLE_FINGERPRINTS: Record<
+  BoardStrategy,
+  (messages: unknown[]) => string[]
+> = {
+  latest: stableFingerprints,
+  'checkpoint-compatible': (messages) =>
+    (messages as MessageWithParts[]).map((message) =>
+      JSON.stringify({
+        role: message.info.role,
+        agent: message.info.agent,
+        parts: message.parts,
+      }),
+    ),
+};
+
+const BOARD_STRATEGIES = Object.keys(
+  STRATEGY_STABLE_FINGERPRINTS,
+) as BoardStrategy[];
+
+describe('cache-safety: board strategy coverage drift guard', () => {
+  test('every configurable board strategy has property coverage', () => {
+    const schemaStrategies =
+      BackgroundJobsConfigSchema.shape.strategy.unwrap().options;
+    expect([...BOARD_STRATEGIES].sort()).toEqual([...schemaStrategies].sort());
+  });
+});
+
+describe.each(BOARD_STRATEGIES)(
+  'cache-safety: turn-over-turn prefix stability (%s)',
+  (strategy) => {
+    test('re-rendering a growing conversation reproduces byte-identical history', async () => {
+      const pipeline = createPipeline({ strategy });
+      const history = buildHistory();
+      const turns = turnEndIndices(history);
+      const fingerprintsFor = STRATEGY_STABLE_FINGERPRINTS[strategy];
 
-      const output = await renderTurn(pipeline, history, endIndex);
-      const fingerprints = stableFingerprints(output.messages);
+      let previous: string[] | undefined;
+      for (const [turnNumber, endIndex] of turns.entries()) {
+        // Exercise cross-turn hook state: a file-tool nudge fires and a
+        // background job launches before the second turn (a real user turn,
+        // so checkpoint mode creates a snapshot), the job is dropped before
+        // the internal-initiator turn renders with an empty board, and a
+        // second job launches before the fourth turn. Snapshot creation,
+        // replay across internal-initiator and empty-board turns, and
+        // unchanged-board dedupe all must leave stable bytes untouched —
+        // the v2.2.5 checkpoint regression rewrote them on exactly these
+        // transitions.
+        if (turnNumber === 1) {
+          pipeline.markFileToolPending();
+          pipeline.board.registerLaunch({
+            taskID: 'task-alpha',
+            parentSessionID: SESSION_ID,
+            agent: 'explorer',
+            description: 'churn fixture',
+            now: FIXTURE_NOW,
+          });
+        }
+        if (turnNumber === 2) pipeline.board.drop('task-alpha');
+        if (turnNumber === 3) {
+          pipeline.board.registerLaunch({
+            taskID: 'task-beta',
+            parentSessionID: SESSION_ID,
+            agent: 'fixer',
+            description: 'second churn fixture',
+            now: FIXTURE_NOW,
+          });
+        }
 
-      if (previous) {
-        if (fingerprints.length < previous.length) {
-          throw new Error(
-            'A transform removed stable messages between turns — this rewrites the cached prefix. Route the content through src/hooks/cache-safe-injection.ts instead.',
-          );
+        const output = await renderTurn(pipeline, history, endIndex);
+        const fingerprints = fingerprintsFor(output.messages);
+
+        if (previous) {
+          if (fingerprints.length < previous.length) {
+            throw new Error(
+              'A transform removed stable messages between turns — this rewrites the cached prefix. Route the content through src/hooks/cache-safe-injection.ts instead.',
+            );
+          }
+          expect(fingerprints.slice(0, previous.length)).toEqual(previous);
         }
-        expect(fingerprints.slice(0, previous.length)).toEqual(previous);
+        previous = fingerprints;
       }
-      previous = fingerprints;
-    }
-  });
+    });
+  },
+);
 
+describe('cache-safety: turn-over-turn prefix stability', () => {
   test('a consumed file-tool nudge is reproduced by the phase reminder on the next turn', async () => {
     const pipeline = createPipeline();
     const history = buildHistory();
@@ -97,41 +166,44 @@ describe('cache-safety: turn-over-turn prefix stability', () => {
   });
 });
 
-describe('cache-safety: specialist sessions', () => {
-  test('non-orchestrator payloads pass through byte-identical', async () => {
-    const pipeline = createPipeline();
-    const specialistSession = 'ses_specialist_fixture';
-    const history = [
-      {
-        info: {
-          role: 'user',
-          agent: 'explorer',
-          sessionID: specialistSession,
-          id: 's01',
+describe.each(BOARD_STRATEGIES)(
+  'cache-safety: specialist sessions (%s)',
+  (strategy) => {
+    test('non-orchestrator payloads pass through byte-identical', async () => {
+      const pipeline = createPipeline({ strategy });
+      const specialistSession = 'ses_specialist_fixture';
+      const history = [
+        {
+          info: {
+            role: 'user',
+            agent: 'explorer',
+            sessionID: specialistSession,
+            id: 's01',
+          },
+          parts: [{ type: 'text', text: 'find the config loader' }],
         },
-        parts: [{ type: 'text', text: 'find the config loader' }],
-      },
-      assistantTurn('s02', 'Searching now.'),
-      {
-        info: {
-          role: 'user',
-          agent: 'explorer',
-          sessionID: specialistSession,
-          id: 's03',
+        assistantTurn('s02', 'Searching now.'),
+        {
+          info: {
+            role: 'user',
+            agent: 'explorer',
+            sessionID: specialistSession,
+            id: 's03',
+          },
+          parts: [{ type: 'text', text: 'summarize what you found' }],
         },
-        parts: [{ type: 'text', text: 'summarize what you found' }],
-      },
-    ];
-    const before = history.map((message) => JSON.stringify(message));
+      ];
+      const before = history.map((message) => JSON.stringify(message));
 
-    const output: TransformOutput = { messages: structuredClone(history) };
-    await pipeline.run(output);
+      const output: TransformOutput = { messages: structuredClone(history) };
+      await pipeline.run(output);
 
-    expect(output.messages.map((message) => JSON.stringify(message))).toEqual(
-      before,
-    );
-  });
-});
+      expect(output.messages.map((message) => JSON.stringify(message))).toEqual(
+        before,
+      );
+    });
+  },
+);
 
 describe('cache-safety: volatile content isolation', () => {
   test('background-job state only ever changes the tagged trailing message', async () => {
@@ -167,41 +239,81 @@ describe('cache-safety: volatile content isolation', () => {
       ),
     ).toBe(false);
   });
-});
 
-describe('cache-safety: determinism under ambient inputs', () => {
-  test('wall clock and randomness never leak into the payload', async () => {
+  test('checkpoint-compatible board state only ever adds tagged snapshot messages', async () => {
     const history = buildHistory();
     const lastTurn = history.length - 1;
-    const originalRandom = Math.random;
-
-    const render = async (time: number, random: number): Promise<string[]> => {
-      setSystemTime(new Date(time));
-      Math.random = () => random;
-      try {
-        const pipeline = createPipeline();
-        pipeline.board.registerLaunch({
-          taskID: 'task-gamma',
-          parentSessionID: SESSION_ID,
-          agent: 'oracle',
-          description: 'determinism fixture',
-          now: FIXTURE_NOW,
-        });
-        const output = await renderTurn(pipeline, history, lastTurn);
-        return output.messages.map((message) => JSON.stringify(message));
-      } finally {
-        Math.random = originalRandom;
-        setSystemTime();
-      }
-    };
 
-    const first = await render(FIXTURE_NOW, 0.1234);
-    const second = await render(FIXTURE_NOW + 987_654_321, 0.9876);
+    const emptyBoard = createPipeline({ strategy: 'checkpoint-compatible' });
+    const busyBoard = createPipeline({ strategy: 'checkpoint-compatible' });
+    busyBoard.board.registerLaunch({
+      taskID: 'task-beta',
+      parentSessionID: SESSION_ID,
+      agent: 'fixer',
+      description: 'checkpoint isolation fixture',
+      now: FIXTURE_NOW,
+    });
 
-    expect(second).toEqual(first);
+    const withoutJobs = await renderTurn(emptyBoard, history, lastTurn);
+    const withJobs = await renderTurn(busyBoard, history, lastTurn);
+
+    // Real message bytes must be identical; board content may only appear
+    // as tagged snapshot messages (append-only by design, so they are part
+    // of the stable prefix rather than a volatile tail).
+    expect(stableFingerprints(withJobs.messages)).toEqual(
+      stableFingerprints(withoutJobs.messages),
+    );
+    const snapshots = withJobs.messages.filter((message) =>
+      isVolatileTaggedMessage(message, BACKGROUND_JOB_BOARD_METADATA_KEY),
+    );
+    expect(snapshots.length).toBeGreaterThan(0);
+    expect(
+      withoutJobs.messages.some((message) =>
+        isVolatileTaggedMessage(message, BACKGROUND_JOB_BOARD_METADATA_KEY),
+      ),
+    ).toBe(false);
   });
 });
 
+describe.each(BOARD_STRATEGIES)(
+  'cache-safety: determinism under ambient inputs (%s)',
+  (strategy) => {
+    test('wall clock and randomness never leak into the payload', async () => {
+      const history = buildHistory();
+      const lastTurn = history.length - 1;
+      const originalRandom = Math.random;
+
+      const render = async (
+        time: number,
+        random: number,
+      ): Promise<string[]> => {
+        setSystemTime(new Date(time));
+        Math.random = () => random;
+        try {
+          const pipeline = createPipeline({ strategy });
+          pipeline.board.registerLaunch({
+            taskID: 'task-gamma',
+            parentSessionID: SESSION_ID,
+            agent: 'oracle',
+            description: 'determinism fixture',
+            now: FIXTURE_NOW,
+          });
+          const output = await renderTurn(pipeline, history, lastTurn);
+          return output.messages.map((message) => JSON.stringify(message));
+        } finally {
+          Math.random = originalRandom;
+          setSystemTime();
+        }
+      };
+
+      const first = await render(FIXTURE_NOW, 0.1234);
+      const second = await render(FIXTURE_NOW + 987_654_321, 0.9876);
+
+      expect(second).toEqual(first);
+    });
+  },
+);
+
 describe('cache-safety: pipeline drift guard', () => {
   const srcRoot = path.resolve(import.meta.dir, '..');