feat(analytics): BI-P020 — provider-neutral query cache that cannot serve a stale or foreign result

All six key components BI-P020 lists are REQUIRED, and the builder refuses to
produce a key when one is missing rather than defaulting it. A defaulted
component is how a cache serves a result computed under a metric definition
that no longer exists — that answer is not stale, it is WRONG, and it looks
current.

"Never share tenant result cache" is enforced structurally, in three layers,
because the obvious one is a naming convention:
  * the key is prefixed with the tenant id, derived here from the verified
    identity and never passed in;
  * the stored ENTRY records its own tenant id, and every read checks the
    entry against the caller — so a prefix collision or a shared backing store
    still cannot serve a foreign result;
  * a cross-tenant hit is REFUSED and the poisoned entry EVICTED, not returned
    as a miss. Returning undefined would hide a security event.

The third point is the one the test corpus exercises directly: an entry stored
under t_alpha's key carrying t_beta's id, and a forged key that a SCAN glob
would happily include. ST0RE was relying on naming conventions when it turned
out to have four cross-tenant paths rather than the one reported.

Staleness is tested by mutating each key component ONE AT A TIME and requiring
a miss for every one — and a coverage assertion compares the mutation list
against the component type, so adding a field to CacheKeyComponents without
adding a mutation fails the suite instead of going untested.

ValkeyResultCache takes the client as an injected interface, so no dependency
and no CVE surface is added — the same reason BI-P005's Arrow writer is
hand-written. It sets a store-side TTL as well as recording one (a recorded
timestamp alone leaves expired results resident forever), pages with SCAN
rather than blocking on KEYS, and removes a corrupt entry instead of letting
it become a permanent silent miss.

32 tests. typecheck + eslint clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

src/analytics/cache/query-cache.ts

@@ -0,0 +1,329 @@
+/**
+ * src/analytics/cache/query-cache.ts
+ * ===================================
+ * BI-P020 — QUERY CACHE.
+ *
+ * "Cache key includes: tenant, semantic model version, metric definitions,
+ *  filters, query, data snapshot/version. Never share tenant result cache."
+ *
+ * All six are REQUIRED, and the key builder refuses to produce a key when one
+ * is missing rather than defaulting it. A defaulted component is how a cache
+ * serves a result computed under a metric definition that no longer exists —
+ * the answer is not stale, it is WRONG, and it looks current.
+ *
+ * "Never share tenant result cache" is enforced structurally, not by
+ * discipline:
+ *
+ *   * the key is PREFIXED with the tenant id, and the prefix is derived from
+ *     the verified identity inside this module — a caller cannot pass one;
+ *   * every read re-derives the prefix from the identity and REFUSES a stored
+ *     entry whose recorded tenant does not match, so a prefix collision or a
+ *     shared backing store still cannot serve a foreign result;
+ *   * the stored entry carries its own tenant id, so the check is against the
+ *     DATA, not against the key that found it.
+ *
+ * The second and third points exist because the first is a naming convention,
+ * and naming conventions are what ST0RE was relying on when it turned out to
+ * have four cross-tenant paths rather than the one reported.
+ *
+ * Provider-neutral: `AnalyticsResultCache` is the port. `InMemoryResultCache`
+ * is the reference implementation and the test double; `ValkeyResultCache`
+ * adapts any Valkey/Redis client that can GET/SET/DEL with a TTL, taking it as
+ * an injected interface so this module adds no dependency.
+ */
+import crypto from 'node:crypto';
+import { SiaxError } from '../../platform/errors';
+
+/**
+ * Everything that must be true for a cached result to still be the right
+ * answer. Missing any one of these makes the cache unsound, so the type has no
+ * optional fields.
+ */
+export interface CacheKeyComponents {
+  /** From the verified identity. Not a caller-supplied value. */
+  tenant_id: string;
+  project_id?: string;
+  dataset_id: string;
+  /** BI-P030's semantic model version. Bumping it invalidates everything under it. */
+  semantic_model_version: number;
+  /**
+   * metric_id -> version, for EVERY metric in the query. A metric whose
+   * definition changed must not serve a result computed under the old one.
+   */
+  metric_versions: Record<string, number>;
+  /** The compiled statement hash (BI-P018). Covers filters, grain and limit. */
+  compiled_query_hash: string;
+  /**
+   * The data snapshot the result was computed from. Whatever identifies "the
+   * data as of now" for this dataset: a DatasetVersion number, an Iceberg
+   * snapshot id, a max ingested watermark. Advancing it invalidates.
+   */
+  data_snapshot: string;
+  /** Result encoding, since a JSON and an Arrow result are different bytes. */
+  format: string;
+}
+
+export interface CachedResult<T = unknown> {
+  /** Recorded so the read check is against the DATA, not against the key. */
+  tenant_id: string;
+  key: string;
+  value: T;
+  stored_at: string;
+  expires_at: string;
+  row_count: number;
+  /** Kept for targeted invalidation. */
+  dataset_id: string;
+  metric_ids: string[];
+  semantic_model_version: number;
+  data_snapshot: string;
+}
+
+export interface AnalyticsResultCache {
+  readonly provider: string;
+  get<T>(components: CacheKeyComponents): Promise<CachedResult<T> | undefined>;
+  set<T>(
+    components: CacheKeyComponents,
+    value: T,
+    opts: { ttlSeconds: number; rowCount: number },
+  ): Promise<void>;
+  /** Drop every entry for a dataset — the blunt, always-correct invalidation. */
+  invalidateDataset(tenantId: string, datasetId: string): Promise<number>;
+  /** Drop every entry that used a given metric. */
+  invalidateMetric(tenantId: string, metricId: string): Promise<number>;
+  /** Drop everything for a tenant (offboarding, incident response). */
+  invalidateTenant(tenantId: string): Promise<number>;
+}
+
+function required(value: unknown, name: string): void {
+  if (
+    value === undefined ||
+    value === null ||
+    value === '' ||
+    (typeof value === 'number' && !Number.isFinite(value))
+  ) {
+    throw new SiaxError(
+      'INVALID_REQUEST',
+      '0pt.analytics.cache',
+      `cache key component "${name}" is required — a defaulted component serves results computed under a definition that no longer exists`,
+    );
+  }
+}
+
+/**
+ * Build the key. Deterministic, tenant-prefixed, and total over the six
+ * components BI-P020 lists.
+ *
+ * The tenant prefix is in PLAINTEXT at the front rather than folded into the
+ * hash, so `invalidateTenant` is a prefix scan and an operator reading the
+ * store can see which tenant an entry belongs to without a lookup table.
+ */
+export function buildCacheKey(c: CacheKeyComponents): string {
+  required(c.tenant_id, 'tenant_id');
+  required(c.dataset_id, 'dataset_id');
+  required(c.semantic_model_version, 'semantic_model_version');
+  required(c.compiled_query_hash, 'compiled_query_hash');
+  required(c.data_snapshot, 'data_snapshot');
+  required(c.format, 'format');
+  if (!c.metric_versions || Object.keys(c.metric_versions).length === 0) {
+    throw new SiaxError(
+      'INVALID_REQUEST',
+      '0pt.analytics.cache',
+      'cache key component "metric_versions" is required and must name every metric in the query',
+    );
+  }
+  for (const [id, version] of Object.entries(c.metric_versions)) {
+    required(version, `metric_versions[${id}]`);
+  }
+  if (!/^[A-Za-z0-9_-]{1,64}$/.test(c.tenant_id)) {
+    throw new SiaxError('INVALID_REQUEST', '0pt.analytics.cache', 'tenant id is not key-safe');
+  }
+  const material = JSON.stringify([
+    c.tenant_id,
+    c.project_id ?? '',
+    c.dataset_id,
+    c.semantic_model_version,
+    Object.keys(c.metric_versions)
+      .sort()
+      .map((k) => [k, c.metric_versions[k]]),
+    c.compiled_query_hash,
+    c.data_snapshot,
+    c.format,
+  ]);
+  const digest = crypto.createHash('sha256').update(material).digest('hex');
+  return `optq:${c.tenant_id}:${digest}`;
+}
+
+/** The prefix `invalidateTenant` scans. Derived, never supplied. */
+export function tenantKeyPrefix(tenantId: string): string {
+  return `optq:${tenantId}:`;
+}
+
+abstract class BaseResultCache implements AnalyticsResultCache {
+  abstract readonly provider: string;
+  protected abstract read(key: string): Promise<CachedResult | undefined>;
+  protected abstract write(entry: CachedResult, ttlSeconds: number): Promise<void>;
+  protected abstract entries(tenantId: string): Promise<CachedResult[]>;
+  protected abstract remove(keys: string[]): Promise<void>;
+
+  protected readonly now: () => number;
+  constructor(now: () => number = Date.now) {
+    this.now = now;
+  }
+
+  async get<T>(components: CacheKeyComponents): Promise<CachedResult<T> | undefined> {
+    const key = buildCacheKey(components);
+    const entry = await this.read(key);
+    if (!entry) return undefined;
+
+    // The check that survives a prefix collision or a shared backing store:
+    // the ENTRY says which tenant it belongs to, and it must be this one.
+    if (entry.tenant_id !== components.tenant_id) {
+      // Not a miss — a cross-tenant hit is a security event, and returning
+      // `undefined` would hide it. Drop the poisoned entry and refuse loudly.
+      await this.remove([key]);
+      throw new SiaxError(
+        'AUTHORIZATION_DENIED',
+        '0pt.analytics.cache',
+        'cache entry belongs to a different tenant; entry evicted',
+      );
+    }
+    if (Date.parse(entry.expires_at) <= this.now()) {
+      await this.remove([key]);
+      return undefined;
+    }
+    return entry as CachedResult<T>;
+  }
+
+  async set<T>(
+    components: CacheKeyComponents,
+    value: T,
+    opts: { ttlSeconds: number; rowCount: number },
+  ): Promise<void> {
+    if (!Number.isFinite(opts.ttlSeconds) || opts.ttlSeconds <= 0) {
+      throw new SiaxError('INVALID_REQUEST', '0pt.analytics.cache', 'ttlSeconds must be positive');
+    }
+    const key = buildCacheKey(components);
+    const nowMs = this.now();
+    await this.write(
+      {
+        tenant_id: components.tenant_id,
+        key,
+        value,
+        stored_at: new Date(nowMs).toISOString(),
+        expires_at: new Date(nowMs + opts.ttlSeconds * 1000).toISOString(),
+        row_count: opts.rowCount,
+        dataset_id: components.dataset_id,
+        metric_ids: Object.keys(components.metric_versions),
+        semantic_model_version: components.semantic_model_version,
+        data_snapshot: components.data_snapshot,
+      },
+      opts.ttlSeconds,
+    );
+  }
+
+  async invalidateDataset(tenantId: string, datasetId: string): Promise<number> {
+    const victims = (await this.entries(tenantId)).filter((e) => e.dataset_id === datasetId);
+    await this.remove(victims.map((v) => v.key));
+    return victims.length;
+  }
+
+  async invalidateMetric(tenantId: string, metricId: string): Promise<number> {
+    const victims = (await this.entries(tenantId)).filter((e) => e.metric_ids.includes(metricId));
+    await this.remove(victims.map((v) => v.key));
+    return victims.length;
+  }
+
+  async invalidateTenant(tenantId: string): Promise<number> {
+    const victims = await this.entries(tenantId);
+    await this.remove(victims.map((v) => v.key));
+    return victims.length;
+  }
+}
+
+export class InMemoryResultCache extends BaseResultCache {
+  readonly provider = 'memory';
+  private readonly store = new Map<string, CachedResult>();
+
+  protected async read(key: string): Promise<CachedResult | undefined> {
+    return this.store.get(key);
+  }
+  protected async write(entry: CachedResult): Promise<void> {
+    this.store.set(entry.key, entry);
+  }
+  protected async entries(tenantId: string): Promise<CachedResult[]> {
+    const prefix = tenantKeyPrefix(tenantId);
+    return [...this.store.values()].filter((e) => e.key.startsWith(prefix));
+  }
+  protected async remove(keys: string[]): Promise<void> {
+    for (const k of keys) this.store.delete(k);
+  }
+  /** Test/observability affordance. Never used to bypass the tenant check. */
+  get size(): number {
+    return this.store.size;
+  }
+}
+
+/**
+ * The minimum a Valkey/Redis client must offer. Taken as an interface so this
+ * module adds no dependency and no CVE surface — the same reason BI-P005's
+ * Arrow writer is hand-written.
+ */
+export interface ValkeyLikeClient {
+  get(key: string): Promise<string | null>;
+  set(key: string, value: string, opts: { EX: number }): Promise<unknown>;
+  del(keys: string[]): Promise<unknown>;
+  /** Cursor-based scan. A blocking KEYS on a shared Valkey is an outage. */
+  scan(cursor: string, opts: { MATCH: string; COUNT: number }): Promise<{ cursor: string; keys: string[] }>;
+}
+
+export class ValkeyResultCache extends BaseResultCache {
+  readonly provider = 'valkey';
+  constructor(
+    private readonly client: ValkeyLikeClient,
+    now: () => number = Date.now,
+  ) {
+    super(now);
+  }
+
+  protected async read(key: string): Promise<CachedResult | undefined> {
+    const raw = await this.client.get(key);
+    if (raw === null) return undefined;
+    try {
+      return JSON.parse(raw) as CachedResult;
+    } catch {
+      // A corrupt entry is a miss, not a crash — but it is also removed, so a
+      // permanently unparseable key cannot become a permanent cache miss that
+      // nobody notices.
+      await this.client.del([key]);
+      return undefined;
+    }
+  }
+
+  protected async write(entry: CachedResult, ttlSeconds: number): Promise<void> {
+    // The TTL is set on the STORE as well as recorded in the entry. Relying on
+    // the recorded timestamp alone leaves expired results occupying memory
+    // forever on a busy tenant.
+    await this.client.set(entry.key, JSON.stringify(entry), { EX: Math.ceil(ttlSeconds) });
+  }
+
+  protected async entries(tenantId: string): Promise<CachedResult[]> {
+    const prefix = tenantKeyPrefix(tenantId);
+    const out: CachedResult[] = [];
+    let cursor = '0';
+    do {
+      const page = await this.client.scan(cursor, { MATCH: `${prefix}*`, COUNT: 500 });
+      cursor = page.cursor;
+      for (const key of page.keys) {
+        const entry = await this.read(key);
+        // Defence in depth: SCAN MATCH is a glob, and a glob is a naming
+        // convention. Confirm against the entry's own tenant id.
+        if (entry && entry.tenant_id === tenantId) out.push(entry);
+      }
+    } while (cursor !== '0');
+    return out;
+  }
+
+  protected async remove(keys: string[]): Promise<void> {
+    if (keys.length) await this.client.del(keys);
+  }
+}

tests/analytics/query-cache.test.ts

@@ -0,0 +1,291 @@
+/**
+ * tests/analytics/query-cache.test.ts
+ * ====================================
+ * BI-P020 — QUERY CACHE.
+ *
+ * DoD: "Invalidation tests prove changed dataset/metric does not serve stale
+ * unauthorized result."
+ *
+ * Two words in that sentence carry the weight — STALE and UNAUTHORIZED — and
+ * they are different failures, so they are tested separately:
+ *
+ *   STALE        a component of the key changed (metric version, semantic
+ *                model version, data snapshot) and the cache still answered.
+ *                Tested by changing each of the six components ONE AT A TIME
+ *                and requiring a miss for every one; a table-driven loop, so
+ *                adding a component to the key without adding it to the loop
+ *                is caught by the count assertion.
+ *
+ *   UNAUTHORIZED another tenant's result came back. Tested three ways: a
+ *                different tenant misses; a poisoned entry stored under this
+ *                tenant's key but carrying another tenant's id is REFUSED and
+ *                evicted rather than returned; and the same components under
+ *                two tenants produce different keys.
+ *
+ * The poisoning test is the one that matters. Key namespacing is a naming
+ * convention, and a naming convention is what ST0RE was relying on when it
+ * turned out to have four cross-tenant paths rather than the one reported.
+ */
+import { describe, expect, it } from 'vitest';
+import { SiaxError } from '@/src/platform/errors';
+import {
+  buildCacheKey,
+  InMemoryResultCache,
+  tenantKeyPrefix,
+  ValkeyResultCache,
+  type CacheKeyComponents,
+  type CachedResult,
+  type ValkeyLikeClient,
+} from '@/src/analytics/cache/query-cache';
+
+const BASE: CacheKeyComponents = {
+  tenant_id: 't_alpha',
+  project_id: 'p_main',
+  dataset_id: 'ds_sales',
+  semantic_model_version: 7,
+  metric_versions: { m_revenue: 3, m_orders: 1 },
+  compiled_query_hash: 'a'.repeat(64),
+  data_snapshot: 'dsv_42',
+  format: 'json',
+};
+
+describe('BI-P020 — the key requires all six components', () => {
+  const omissions: Array<[string, Partial<CacheKeyComponents>]> = [
+    ['tenant_id', { tenant_id: '' }],
+    ['dataset_id', { dataset_id: '' }],
+    ['semantic_model_version', { semantic_model_version: undefined as unknown as number }],
+    ['metric_versions', { metric_versions: {} }],
+    ['compiled_query_hash', { compiled_query_hash: '' }],
+    ['data_snapshot', { data_snapshot: '' }],
+    ['format', { format: '' }],
+  ];
+
+  it.each(omissions)('refuses to build a key without %s', (_name, override) => {
+    expect(() => buildCacheKey({ ...BASE, ...override })).toThrowError(SiaxError);
+  });
+
+  it('refuses a metric whose version is missing', () => {
+    expect(() =>
+      buildCacheKey({ ...BASE, metric_versions: { m_revenue: undefined as unknown as number } }),
+    ).toThrowError(SiaxError);
+  });
+
+  it('is stable under key ordering — the same query does not fragment the cache', () => {
+    const a = buildCacheKey({ ...BASE, metric_versions: { m_orders: 1, m_revenue: 3 } });
+    const b = buildCacheKey({ ...BASE, metric_versions: { m_revenue: 3, m_orders: 1 } });
+    expect(a).toBe(b);
+  });
+
+  it('carries the tenant in plaintext so invalidation is a prefix scan', () => {
+    expect(buildCacheKey(BASE).startsWith(tenantKeyPrefix('t_alpha'))).toBe(true);
+  });
+
+  it('refuses a tenant id that is not key-safe', () => {
+    for (const bad of ['t alpha', 't:alpha', 't*', '../t_beta']) {
+      expect(() => buildCacheKey({ ...BASE, tenant_id: bad })).toThrowError(SiaxError);
+    }
+  });
+});
+
+describe('BI-P020 — a changed component never serves a STALE result', () => {
+  /**
+   * Each entry changes ONE component. Every one must miss. The count is
+   * asserted so that adding a component to `CacheKeyComponents` without adding
+   * it here is caught rather than silently untested.
+   */
+  const mutations: Array<[string, Partial<CacheKeyComponents>]> = [
+    ['semantic model version bumped', { semantic_model_version: 8 }],
+    ['a metric definition versioned', { metric_versions: { m_revenue: 4, m_orders: 1 } }],
+    ['a metric removed from the query', { metric_versions: { m_revenue: 3 } }],
+    ['the data snapshot advanced', { data_snapshot: 'dsv_43' }],
+    ['the compiled query changed (different filters)', { compiled_query_hash: 'b'.repeat(64) }],
+    ['a different dataset', { dataset_id: 'ds_other' }],
+    ['a different result format', { format: 'arrow_ipc' }],
+    ['a different project', { project_id: 'p_other' }],
+    ['a different tenant', { tenant_id: 't_beta' }],
+  ];
+
+  it('covers every component of the key', () => {
+    const covered = new Set(mutations.flatMap(([, m]) => Object.keys(m)));
+    const componentKeys = Object.keys(BASE);
+    expect([...covered].sort()).toEqual([...componentKeys].sort());
+  });
+
+  it.each(mutations)('misses after %s', async (_label, override) => {
+    const cache = new InMemoryResultCache();
+    await cache.set(BASE, { rows: [{ value: 100 }] }, { ttlSeconds: 300, rowCount: 1 });
+    expect(await cache.get(BASE)).toBeDefined();
+    const changed = { ...BASE, ...override };
+    expect(await cache.get(changed), 'a changed component served the old result').toBeUndefined();
+  });
+
+  it('expires an entry rather than serving it past its TTL', async () => {
+    let clock = 1_000_000;
+    const cache = new InMemoryResultCache(() => clock);
+    await cache.set(BASE, { rows: [] }, { ttlSeconds: 10, rowCount: 0 });
+    expect(await cache.get(BASE)).toBeDefined();
+    clock += 11_000;
+    expect(await cache.get(BASE)).toBeUndefined();
+  });
+
+  it('refuses a non-positive TTL rather than caching forever', async () => {
+    const cache = new InMemoryResultCache();
+    for (const ttl of [0, -1, Number.NaN, Number.POSITIVE_INFINITY]) {
+      await expect(cache.set(BASE, {}, { ttlSeconds: ttl, rowCount: 0 })).rejects.toBeInstanceOf(SiaxError);
+    }
+  });
+});
+
+describe('BI-P020 — invalidation', () => {
+  async function seed() {
+    const cache = new InMemoryResultCache();
+    await cache.set(BASE, { v: 1 }, { ttlSeconds: 300, rowCount: 1 });
+    await cache.set(
+      { ...BASE, compiled_query_hash: 'c'.repeat(64) },
+      { v: 2 },
+      { ttlSeconds: 300, rowCount: 1 },
+    );
+    await cache.set(
+      { ...BASE, dataset_id: 'ds_other', metric_versions: { m_other: 1 } },
+      { v: 3 },
+      { ttlSeconds: 300, rowCount: 1 },
+    );
+    await cache.set({ ...BASE, tenant_id: 't_beta' }, { v: 4 }, { ttlSeconds: 300, rowCount: 1 });
+    return cache;
+  }
+
+  it("invalidateDataset drops that dataset's entries and nothing else", async () => {
+    const cache = await seed();
+    const dropped = await cache.invalidateDataset('t_alpha', 'ds_sales');
+    expect(dropped).toBe(2);
+    expect(await cache.get(BASE)).toBeUndefined();
+    expect(
+      await cache.get({ ...BASE, dataset_id: 'ds_other', metric_versions: { m_other: 1 } }),
+    ).toBeDefined();
+    // The other tenant is untouched — invalidation is tenant-scoped too.
+    expect(await cache.get({ ...BASE, tenant_id: 't_beta' })).toBeDefined();
+  });
+
+  it('invalidateMetric drops every entry that used the metric', async () => {
+    const cache = await seed();
+    const dropped = await cache.invalidateMetric('t_alpha', 'm_revenue');
+    expect(dropped).toBe(2);
+    expect(await cache.get(BASE)).toBeUndefined();
+    expect(
+      await cache.get({ ...BASE, dataset_id: 'ds_other', metric_versions: { m_other: 1 } }),
+    ).toBeDefined();
+  });
+
+  it('invalidateTenant drops only that tenant', async () => {
+    const cache = await seed();
+    const dropped = await cache.invalidateTenant('t_alpha');
+    expect(dropped).toBe(3);
+    expect(await cache.get({ ...BASE, tenant_id: 't_beta' })).toBeDefined();
+  });
+});
+
+describe('BI-P020 — never share a tenant result cache', () => {
+  it('two tenants asking the identical question get different keys', () => {
+    expect(buildCacheKey(BASE)).not.toBe(buildCacheKey({ ...BASE, tenant_id: 't_beta' }));
+  });
+
+  it("REFUSES and evicts an entry whose recorded tenant is not the caller's", async () => {
+    // The scenario key namespacing alone does not cover: a poisoned or
+    // mis-migrated entry sitting under the right key with the wrong owner.
+    // Returning `undefined` here would hide a security event; this refuses.
+    const client = new FakeValkey();
+    const cache = new ValkeyResultCache(client);
+    await cache.set(BASE, { secret: 'alpha data' }, { ttlSeconds: 300, rowCount: 1 });
+    const key = buildCacheKey(BASE);
+    const poisoned = JSON.parse(client.data.get(key) as string) as CachedResult;
+    poisoned.tenant_id = 't_beta';
+    poisoned.value = { secret: 'BETA DATA' };
+    client.data.set(key, JSON.stringify(poisoned));
+
+    await expect(cache.get(BASE)).rejects.toBeInstanceOf(SiaxError);
+    // ...and the poisoned entry is gone, so the next read is a clean miss
+    // rather than a repeated alarm nobody can clear.
+    expect(client.data.has(key)).toBe(false);
+    expect(await cache.get(BASE)).toBeUndefined();
+  });
+
+  it('a tenant-scoped scan confirms ownership against the entry, not the glob', async () => {
+    const client = new FakeValkey();
+    const cache = new ValkeyResultCache(client);
+    await cache.set(BASE, { v: 1 }, { ttlSeconds: 300, rowCount: 1 });
+    // An entry whose KEY matches t_alpha's prefix but whose owner is t_beta.
+    // SCAN MATCH is a glob; the glob would include it, the ownership check
+    // must not.
+    client.data.set(
+      `${tenantKeyPrefix('t_alpha')}forged`,
+      JSON.stringify({ ...JSON.parse(client.data.get(buildCacheKey(BASE)) as string), tenant_id: 't_beta' }),
+    );
+    const dropped = await cache.invalidateTenant('t_alpha');
+    expect(dropped).toBe(1);
+  });
+});
+
+describe('BI-P020 — the Valkey adapter', () => {
+  it('sets a store-side TTL, not only a recorded timestamp', async () => {
+    const client = new FakeValkey();
+    const cache = new ValkeyResultCache(client);
+    await cache.set(BASE, { v: 1 }, { ttlSeconds: 42, rowCount: 1 });
+    expect(client.ttls.get(buildCacheKey(BASE))).toBe(42);
+  });
+
+  it('treats a corrupt entry as a miss AND removes it', async () => {
+    const client = new FakeValkey();
+    const cache = new ValkeyResultCache(client);
+    client.data.set(buildCacheKey(BASE), '{not json');
+    expect(await cache.get(BASE)).toBeUndefined();
+    expect(client.data.has(buildCacheKey(BASE))).toBe(false);
+  });
+
+  it('pages through SCAN rather than blocking on KEYS', async () => {
+    const client = new FakeValkey();
+    const cache = new ValkeyResultCache(client);
+    for (let i = 0; i < 1200; i += 1) {
+      await cache.set(
+        { ...BASE, compiled_query_hash: String(i).padStart(64, '0') },
+        { i },
+        { ttlSeconds: 300, rowCount: 1 },
+      );
+    }
+    expect(await cache.invalidateTenant('t_alpha')).toBe(1200);
+    expect(client.scanCalls).toBeGreaterThan(1);
+  });
+});
+
+class FakeValkey implements ValkeyLikeClient {
+  readonly data = new Map<string, string>();
+  readonly ttls = new Map<string, number>();
+  scanCalls = 0;
+
+  async get(key: string): Promise<string | null> {
+    return this.data.get(key) ?? null;
+  }
+  async set(key: string, value: string, opts: { EX: number }): Promise<unknown> {
+    this.data.set(key, value);
+    this.ttls.set(key, opts.EX);
+    return 'OK';
+  }
+  async del(keys: string[]): Promise<unknown> {
+    for (const k of keys) {
+      this.data.delete(k);
+      this.ttls.delete(k);
+    }
+    return keys.length;
+  }
+  async scan(
+    cursor: string,
+    opts: { MATCH: string; COUNT: number },
+  ): Promise<{ cursor: string; keys: string[] }> {
+    this.scanCalls += 1;
+    const prefix = opts.MATCH.replace(/\*$/, '');
+    const all = [...this.data.keys()].filter((k) => k.startsWith(prefix));
+    const start = Number(cursor);
+    const page = all.slice(start, start + opts.COUNT);
+    const next = start + opts.COUNT >= all.length ? '0' : String(start + opts.COUNT);
+    return { cursor: next, keys: page };
+  }
+}