feat(analytics): BI-P105 — a pre-aggregate serves only while it is provably undrifted

A slow query is a complaint; a drifted pre-aggregate answers instantly, with
confidence, and is wrong. So the router is fail-closed on eight named reasons,
and the one people leave out — drift_never_verified — is the one that matters
on launch day: a brand-new aggregate is exactly as unproven as a drifted one.

Proved against ClickHouse 26.3.10.60, 2M rows, a real materialized view
(evidence/bi-p105/materialization-drift.json):

  after backfill      match, 48 buckets, 801 571 002 both sides
  GDPR erasure        ALTER TABLE ... DELETE on the source. An MV is an INSERT
                      trigger and never sees a mutation, so the aggregate kept
                      the deleted revenue: it would have reported 801 571 002
                      against a true 761 580 554 — 39 990 448 overstated,
                      5.251%, with NOTHING stale. Staleness at the moment of
                      the decision: 1 second.
  guarded router      base_table (state_not_serving)
  counterfactual      a staleness-only router SERVES it — measured, not assumed
  after rebuild       match, and routing returns to the materialization

Drift detection treats a MISSING bucket as drift even when every value present
is correct (late-arriving rows are the sneakiest failure), and an empty
comparison as insufficient_data rather than match — the same false green
BI-P025 found in a quality check over an empty table.

The measured speed-up is reported as a distribution, not a figure: p50 3.4x,
but the run-to-run range spans 0.6x-7.5x on one idle host, so quoting "3.4x"
as a property of the system would be misleading (BI-P104 doctrine).

19 unit tests + 4 against live ClickHouse.

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

evidence/bi-p105/materialization-drift.json

@@ -0,0 +1,75 @@
+{
+  "clickhouse_version": "26.3.10.60",
+  "recorded_at": "2026-09-07T01:55:42.821Z",
+  "source_rows": 2000000,
+  "after_backfill": {
+    "verdict": "match",
+    "buckets_compared": 48,
+    "revenue_total": {
+      "source": 801571002,
+      "aggregate": 801571002,
+      "delta": 0
+    }
+  },
+  "performance": {
+    "note": "ClickHouse caches marks and pages between repetitions, so these are WARM numbers on one host with nothing else running. They describe this shape of query on this data, not a system property.",
+    "repetitions": 9,
+    "base_table_ms": {
+      "n": 9,
+      "min": 48.70366200000012,
+      "p50": 76.67811500000016,
+      "p95": 115.04416399999991,
+      "max": 115.04416399999991,
+      "spread_ratio": 2.36
+    },
+    "pre_aggregate_ms": {
+      "n": 9,
+      "min": 15.281603999999788,
+      "p50": 22.727642999999944,
+      "p95": 84.41279499999973,
+      "max": 84.41279499999973,
+      "spread_ratio": 5.52
+    },
+    "speedup_p50": 3.4,
+    "speedup_range_min_max": [0.6, 7.5]
+  },
+  "gdpr_erasure_drill": {
+    "mutation": "ALTER TABLE sales DELETE WHERE tenant_id='t_alpha' AND customer_id < 250",
+    "verdict": "drifted",
+    "buckets_compared": 48,
+    "revenue_true_after_erasure": 761580554,
+    "revenue_pre_aggregate_would_report": 801571002,
+    "revenue_overstatement": 39990448,
+    "revenue_overstatement_pct": 5.251,
+    "worst_bucket": {
+      "key": "2026-08-28|FI",
+      "metric": "revenue",
+      "source_value": 15830779,
+      "aggregate_value": 16699145,
+      "absolute_delta": 868366,
+      "relative_delta": 0.05485301765630106
+    },
+    "guarded_router_decision": {
+      "source": "base_table",
+      "refusal": "state_not_serving",
+      "detail": "materialization state is \"drifted\""
+    },
+    "staleness_only_router_would_serve": true,
+    "staleness_seconds_at_decision": 1,
+    "note": "The aggregate was NOT stale by any clock: it had just been refreshed, and every freshness check passed. Only the source comparison caught it. This is why staleness alone is not a correctness guarantee."
+  },
+  "after_rebuild": {
+    "verdict": "match",
+    "revenue_total": {
+      "source": 761580554,
+      "aggregate": 761580554,
+      "delta": 0
+    },
+    "routing": {
+      "source": "materialization",
+      "materialization_id": "mv_sales_daily",
+      "target_ref": "opt_p105_mtql9p2f.sales_daily",
+      "staleness_seconds": 0
+    }
+  }
+}

src/analytics/materialization/drift-detector.ts

@@ -0,0 +1,170 @@
+/**
+ * src/analytics/materialization/drift-detector.ts
+ * ================================================
+ * BI-P105 — is the pre-aggregate still telling the truth?
+ *
+ * The comparison is bucket-by-bucket between the aggregate and a recomputation
+ * from the source over the same window. Three things make it a real detector
+ * rather than a ceremony:
+ *
+ * 1. A MISSING BUCKET IS DRIFT. The nastiest pre-aggregate failure is not a
+ *    wrong value, it is an absent row: late-arriving data, a materialized view
+ *    that was created after the backfill, a refresh that failed silently. Every
+ *    value it DOES hold is correct, so a value-only comparison passes while the
+ *    total is wrong. Set difference is checked first, before any arithmetic.
+ *
+ * 2. ZERO BUCKETS IS `insufficient_data`, NEVER `match`. A comparison over an
+ *    empty window finds no discrepancies — the same false green BI-P025 found
+ *    in a quality check over an empty table. The router treats
+ *    `insufficient_data` as "do not serve", not as "fine".
+ *
+ * 3. RELATIVE AND ABSOLUTE TOLERANCE BOTH HAVE TO HOLD. A 0.1% tolerance on a
+ *    revenue bucket of 40M is 40 000 kr of invisible error; an absolute-only
+ *    tolerance flags every rounding difference on a large bucket. Requiring
+ *    both means a discrepancy has to be small in kronor AND small in
+ *    proportion.
+ */
+
+import type { DriftTolerance, DriftVerdict, MaterializationStatus } from './materialization';
+
+/** One aggregated row, keyed by its dimension tuple. */
+export interface AggregateBucket {
+  /** Stable key for the dimension tuple, e.g. "2026-09-01|SE". */
+  key: string;
+  /** metric name -> value. */
+  values: Record<string, number>;
+}
+
+export interface DriftDiscrepancy {
+  key: string;
+  metric: string;
+  source_value: number;
+  aggregate_value: number;
+  absolute_delta: number;
+  /** |delta| / |source|, or Infinity when the source value is 0 and the aggregate is not. */
+  relative_delta: number;
+}
+
+export interface DriftReport {
+  verdict: DriftVerdict;
+  checked_at: string;
+  buckets_compared: number;
+  /** In the source recomputation, absent from the aggregate. Always drift. */
+  missing_in_aggregate: string[];
+  /** In the aggregate, absent from the source. Always drift (deleted rows). */
+  extra_in_aggregate: string[];
+  discrepancies: DriftDiscrepancy[];
+  worst?: DriftDiscrepancy;
+  tolerance: DriftTolerance;
+  /**
+   * What a caller would have been told if it had trusted the aggregate: the
+   * total the aggregate reports vs the total the source computes, per metric.
+   * This is the number that makes a drift report actionable — "0.4% off" means
+   * nothing until it is "1 284 500 kr reported, 1 271 900 kr true".
+   */
+  totals: Record<string, { source: number; aggregate: number; delta: number }>;
+}
+
+const MAX_REPORTED_DISCREPANCIES = 50;
+
+function relative(sourceValue: number, delta: number): number {
+  if (sourceValue === 0) return delta === 0 ? 0 : Number.POSITIVE_INFINITY;
+  return Math.abs(delta / sourceValue);
+}
+
+/**
+ * Compare a recomputation from source against the pre-aggregate.
+ *
+ * `fromSource` is the truth. Both sides must cover the same window; the caller
+ * is responsible for that, and `missing_in_aggregate` will catch it loudly if
+ * they do not.
+ */
+export function detectDrift(args: {
+  fromSource: AggregateBucket[];
+  fromAggregate: AggregateBucket[];
+  tolerance: DriftTolerance;
+  metrics: string[];
+  now?: Date;
+}): DriftReport {
+  const { fromSource, fromAggregate, tolerance, metrics } = args;
+  const checked_at = (args.now ?? new Date()).toISOString();
+  const sourceByKey = new Map(fromSource.map((b) => [b.key, b]));
+  const aggByKey = new Map(fromAggregate.map((b) => [b.key, b]));
+
+  const missing_in_aggregate = [...sourceByKey.keys()].filter((k) => !aggByKey.has(k)).sort();
+  const extra_in_aggregate = [...aggByKey.keys()].filter((k) => !sourceByKey.has(k)).sort();
+
+  const discrepancies: DriftDiscrepancy[] = [];
+  let buckets_compared = 0;
+  const totals: DriftReport['totals'] = {};
+  for (const m of metrics) totals[m] = { source: 0, aggregate: 0, delta: 0 };
+
+  for (const [key, sourceBucket] of sourceByKey) {
+    for (const m of metrics) totals[m].source += sourceBucket.values[m] ?? 0;
+    const aggBucket = aggByKey.get(key);
+    if (!aggBucket) continue;
+    buckets_compared += 1;
+    for (const m of metrics) {
+      const sv = sourceBucket.values[m] ?? 0;
+      const av = aggBucket.values[m] ?? 0;
+      const delta = av - sv;
+      const rel = relative(sv, delta);
+      if (Math.abs(delta) > tolerance.absolute || rel > tolerance.relative) {
+        discrepancies.push({
+          key,
+          metric: m,
+          source_value: sv,
+          aggregate_value: av,
+          absolute_delta: delta,
+          relative_delta: rel,
+        });
+      }
+    }
+  }
+  for (const bucket of aggByKey.values()) {
+    for (const m of metrics) totals[m].aggregate += bucket.values[m] ?? 0;
+  }
+  for (const m of metrics) totals[m].delta = totals[m].aggregate - totals[m].source;
+
+  discrepancies.sort((a, b) => Math.abs(b.absolute_delta) - Math.abs(a.absolute_delta));
+
+  // A comparison that compared nothing has not proved anything. This is the
+  // BI-P025 false green, restated for pre-aggregates.
+  const verdict: DriftVerdict =
+    buckets_compared === 0 && missing_in_aggregate.length === 0 && extra_in_aggregate.length === 0
+      ? 'insufficient_data'
+      : discrepancies.length > 0 || missing_in_aggregate.length > 0 || extra_in_aggregate.length > 0
+        ? 'drifted'
+        : 'match';
+
+  const report: DriftReport = {
+    verdict,
+    checked_at,
+    buckets_compared,
+    missing_in_aggregate,
+    extra_in_aggregate,
+    discrepancies: discrepancies.slice(0, MAX_REPORTED_DISCREPANCIES),
+    tolerance,
+    totals,
+  };
+  if (discrepancies.length > 0) report.worst = discrepancies[0];
+  return report;
+}
+
+/**
+ * Fold a drift report into the runtime status. A drifted aggregate is taken
+ * OUT of service here rather than merely annotated: leaving it `serving` with
+ * a bad verdict relies on every future reader remembering to check the
+ * verdict, and one that forgets serves the wrong number.
+ *
+ * `insufficient_data` also stops service. It means the check could not prove
+ * anything, and an unproven aggregate is not a current one.
+ */
+export function applyDriftReport(status: MaterializationStatus, report: DriftReport): MaterializationStatus {
+  return {
+    ...status,
+    state: report.verdict === 'match' ? status.state : 'drifted',
+    last_drift_check_at: report.checked_at,
+    last_drift_verdict: report.verdict,
+  };
+}

src/analytics/materialization/materialization.ts

@@ -0,0 +1,309 @@
+/**
+ * src/analytics/materialization/materialization.ts
+ * =================================================
+ * BI-P105 — MATERIALIZATION / PRE-AGGREGATION.
+ *
+ * A pre-aggregate is the one optimisation that fails SILENTLY. A slow query is
+ * a complaint; a pre-aggregate that has drifted from its source answers
+ * instantly, with confidence, and is wrong — no error, no timeout, nothing in
+ * a log. So this module is built around one rule:
+ *
+ *     A MATERIALIZATION SERVES ONLY WHILE IT IS PROVABLY CURRENT AND PROVABLY
+ *     UNDRIFTED. Every unknown routes to the source table.
+ *
+ * Six refusal reasons, each of which is a real way the aggregate becomes a
+ * confident liar:
+ *
+ *   coverage_miss              it does not carry the metric/dimension/grain
+ *                              asked for (a filter on a dimension the
+ *                              aggregate collapsed cannot be applied after
+ *                              the fact, and applying it anyway is how a
+ *                              pre-aggregate returns another slice's number)
+ *   watermark_unknown          nobody knows how current it is. BI-P026 already
+ *                              refuses to render UNKNOWN age as current; the
+ *                              same doctrine applies here.
+ *   staleness_exceeded         it is older than its own freshness contract
+ *   window_beyond_watermark    the question reaches past the newest data the
+ *                              aggregate contains — it would silently answer
+ *                              with a shorter period
+ *   drift_never_verified       it has never been compared to its source
+ *   drift_verification_stale   the last comparison is older than the interval
+ *                              the spec requires
+ *   drift_detected             the last comparison FAILED
+ *
+ * `drift_never_verified` is the one people leave out, and it is the one that
+ * matters most on day one: a brand-new materialization is exactly as unproven
+ * as a drifted one, and a router that defaults new aggregates to "serving"
+ * ships the bug on the day the feature launches.
+ *
+ * The DoD also asks for measured improvement and documented freshness
+ * semantics: scripts/bi-p105-materialization-drift.mjs measures both against a
+ * live ClickHouse and writes evidence/bi-p105/materialization-drift.json,
+ * including what happens WITHOUT this router (the drifted aggregate served).
+ */
+import { SiaxError } from '../../platform/errors';
+import type { AnalyticsQueryRequest, TimeGrain } from '../contract/query-request';
+
+export const MATERIALIZATION_STRATEGIES = [
+  'clickhouse_materialized_view',
+  'clickhouse_projection',
+  'dbt_incremental',
+  'cube_preaggregation',
+] as const;
+export type MaterializationStrategy = (typeof MATERIALIZATION_STRATEGIES)[number];
+
+export const MATERIALIZATION_STATES = ['building', 'serving', 'drifted', 'disabled'] as const;
+export type MaterializationState = (typeof MATERIALIZATION_STATES)[number];
+
+/**
+ * Grains ordered coarsest-last. An aggregate at `day` can answer `day` and
+ * anything coarser by re-aggregating; it can NEVER answer `hour`, because the
+ * detail was thrown away when the rows were rolled up.
+ */
+const GRAIN_ORDER: readonly TimeGrain[] = ['hour', 'day', 'week', 'month', 'quarter', 'year'];
+
+/**
+ * `week` does not divide `month` and `month` does not divide `quarter` by row
+ * count, but both roll up cleanly from `day`. Grains that CANNOT be derived by
+ * re-aggregating a coarser one are listed here so the router refuses rather
+ * than quietly returning a mis-bucketed series.
+ */
+function canRollUp(from: TimeGrain, to: TimeGrain): boolean {
+  if (from === to) return true;
+  const fi = GRAIN_ORDER.indexOf(from);
+  const ti = GRAIN_ORDER.indexOf(to);
+  if (fi < 0 || ti < 0 || ti < fi) return false;
+  // Weekly buckets do not compose into calendar months/quarters/years.
+  if (from === 'week' && to !== 'week') return false;
+  return true;
+}
+
+export interface DriftTolerance {
+  /** Absolute delta allowed per metric per bucket. */
+  absolute: number;
+  /** Relative delta allowed (0.001 = 0.1%). Both must hold. */
+  relative: number;
+}
+
+/** What a materialization IS. Declarative; the engine adapters read it. */
+export interface MaterializationSpec {
+  materialization_id: string;
+  tenant_id: string;
+  project_id: string;
+  /** The dataset it pre-aggregates (BI-P003). */
+  dataset: string;
+  strategy: MaterializationStrategy;
+  /** Physical target (ClickHouse table, dbt model, Cube pre-agg name). */
+  target_ref: string;
+  metrics: string[];
+  dimensions: string[];
+  time_dimension?: string;
+  grain?: TimeGrain;
+  /** The freshness contract. Older than this and it stops serving. */
+  max_staleness_seconds: number;
+  /** How often the aggregate must be re-proved against its source. */
+  drift_check_interval_seconds: number;
+  tolerance: DriftTolerance;
+}
+
+export const DRIFT_VERDICTS = ['match', 'drifted', 'insufficient_data'] as const;
+export type DriftVerdict = (typeof DRIFT_VERDICTS)[number];
+
+/** Runtime state. Separate from the spec: the spec is config, this is fact. */
+export interface MaterializationStatus {
+  materialization_id: string;
+  state: MaterializationState;
+  /**
+   * The newest source timestamp the aggregate represents. UNDEFINED means
+   * UNKNOWN, and unknown is never treated as current.
+   */
+  data_watermark?: string;
+  last_refresh_at?: string;
+  last_drift_check_at?: string;
+  last_drift_verdict?: DriftVerdict;
+}
+
+export const ROUTING_REFUSALS = [
+  'coverage_miss',
+  'watermark_unknown',
+  'staleness_exceeded',
+  'window_beyond_watermark',
+  'drift_never_verified',
+  'drift_verification_stale',
+  'drift_detected',
+  'state_not_serving',
+] as const;
+export type RoutingRefusal = (typeof ROUTING_REFUSALS)[number];
+
+export type RoutingDecision =
+  | {
+      source: 'materialization';
+      materialization_id: string;
+      target_ref: string;
+      /** Reported so a caller can show it; never inferred, always measured. */
+      staleness_seconds: number;
+    }
+  | { source: 'base_table'; refusal: RoutingRefusal; detail: string };
+
+function refuse(refusal: RoutingRefusal, detail: string): RoutingDecision {
+  return { source: 'base_table', refusal, detail };
+}
+
+/**
+ * Staleness of an aggregate, in seconds, or `undefined` when it cannot be
+ * computed. Deliberately NOT defaulted to 0: "we could not tell" and "it is
+ * current" are different facts, and conflating them is the bug.
+ */
+export function materializationStaleness(status: MaterializationStatus, now: Date): number | undefined {
+  if (!status.data_watermark) return undefined;
+  const t = Date.parse(status.data_watermark);
+  if (Number.isNaN(t)) return undefined;
+  return Math.max(0, Math.floor((now.getTime() - t) / 1000));
+}
+
+/**
+ * Decide whether a request may be answered from the pre-aggregate.
+ *
+ * Fail-closed by construction: the function returns `materialization` only
+ * after every check has passed, and every unknown is a refusal with a named
+ * reason. Callers get the reason so it can be logged and counted (BI-P103
+ * reads the counts).
+ */
+export function selectQuerySource(args: {
+  request: AnalyticsQueryRequest;
+  spec: MaterializationSpec;
+  status: MaterializationStatus;
+  now: Date;
+}): RoutingDecision {
+  const { request, spec, status, now } = args;
+
+  if (request.dataset !== spec.dataset) {
+    return refuse('coverage_miss', `materialization covers "${spec.dataset}", not "${request.dataset}"`);
+  }
+  if (status.state !== 'serving') {
+    return refuse('state_not_serving', `materialization state is "${status.state}"`);
+  }
+
+  // ── coverage ────────────────────────────────────────────────────────────
+  const metrics = new Set(spec.metrics);
+  for (const m of request.metrics) {
+    if (!metrics.has(m)) return refuse('coverage_miss', `metric "${m}" is not in the pre-aggregate`);
+  }
+  const dimensions = new Set(spec.dimensions);
+  for (const d of request.dimensions) {
+    if (!dimensions.has(d)) return refuse('coverage_miss', `dimension "${d}" is not in the pre-aggregate`);
+  }
+  for (const f of request.filters) {
+    if (!dimensions.has(f.dimension)) {
+      // The aggregate summed across this dimension. Filtering afterwards would
+      // return a number computed over rows the filter should have excluded.
+      return refuse(
+        'coverage_miss',
+        `filter on "${f.dimension}" cannot be applied: the pre-aggregate has already summed across it`,
+      );
+    }
+  }
+  const requestedGrain = request.timeRange?.grain;
+  if (requestedGrain) {
+    if (!spec.grain) return refuse('coverage_miss', 'the pre-aggregate has no time grain');
+    if (!canRollUp(spec.grain, requestedGrain)) {
+      return refuse(
+        'coverage_miss',
+        `a "${spec.grain}" pre-aggregate cannot answer at "${requestedGrain}" grain`,
+      );
+    }
+  }
+
+  // ── freshness ───────────────────────────────────────────────────────────
+  const staleness = materializationStaleness(status, now);
+  if (staleness === undefined) {
+    return refuse(
+      'watermark_unknown',
+      'the pre-aggregate has no data watermark — unknown age is not current age',
+    );
+  }
+  if (staleness > spec.max_staleness_seconds) {
+    return refuse(
+      'staleness_exceeded',
+      `${staleness}s stale, contract allows ${spec.max_staleness_seconds}s`,
+    );
+  }
+  const to = request.timeRange?.to;
+  if (to) {
+    const toMs = Date.parse(to);
+    const watermarkMs = Date.parse(status.data_watermark!);
+    if (!Number.isNaN(toMs) && toMs > watermarkMs) {
+      return refuse(
+        'window_beyond_watermark',
+        `the question runs to ${to} but the pre-aggregate only contains data to ${status.data_watermark}`,
+      );
+    }
+  }
+
+  // ── drift ───────────────────────────────────────────────────────────────
+  if (!status.last_drift_check_at || !status.last_drift_verdict) {
+    return refuse(
+      'drift_never_verified',
+      'the pre-aggregate has never been compared to its source; a new aggregate is as unproven as a drifted one',
+    );
+  }
+  if (status.last_drift_verdict !== 'match') {
+    return refuse('drift_detected', `last drift check returned "${status.last_drift_verdict}"`);
+  }
+  const checkedAgo = Math.floor((now.getTime() - Date.parse(status.last_drift_check_at)) / 1000);
+  if (Number.isNaN(checkedAgo)) {
+    return refuse('drift_never_verified', 'last_drift_check_at is not a timestamp');
+  }
+  if (checkedAgo > spec.drift_check_interval_seconds) {
+    return refuse(
+      'drift_verification_stale',
+      `last verified ${checkedAgo}s ago, spec requires every ${spec.drift_check_interval_seconds}s`,
+    );
+  }
+
+  return {
+    source: 'materialization',
+    materialization_id: spec.materialization_id,
+    target_ref: spec.target_ref,
+    staleness_seconds: staleness,
+  };
+}
+
+/**
+ * The routing decision as a freshness envelope for the result (BI-P026 shape):
+ * a consumer must be able to see WHICH source answered and how old it was.
+ * A result that does not say is a result nobody can audit.
+ */
+export function describeRouting(decision: RoutingDecision): {
+  served_from: 'materialization' | 'base_table';
+  materialization_id?: string;
+  staleness_seconds?: number;
+  refusal?: RoutingRefusal;
+  detail?: string;
+} {
+  return decision.source === 'materialization'
+    ? {
+        served_from: 'materialization',
+        materialization_id: decision.materialization_id,
+        staleness_seconds: decision.staleness_seconds,
+      }
+    : { served_from: 'base_table', refusal: decision.refusal, detail: decision.detail };
+}
+
+export function assertSpec(spec: MaterializationSpec): void {
+  const invalid = (d: string): never => {
+    throw new SiaxError('INVALID_REQUEST', '0pt.analytics.materialization', d);
+  };
+  if (!spec.materialization_id?.trim()) invalid('materialization_id is required');
+  if (!spec.tenant_id?.trim() || !spec.project_id?.trim()) invalid('a materialization is tenant-scoped');
+  if (spec.metrics.length === 0) invalid('a materialization with no metrics pre-aggregates nothing');
+  if (spec.max_staleness_seconds <= 0) invalid('max_staleness_seconds must be positive');
+  if (spec.drift_check_interval_seconds <= 0) {
+    invalid(
+      'drift_check_interval_seconds must be positive — an aggregate that is never re-verified is one ' +
+        'source mutation away from serving wrong numbers forever',
+    );
+  }
+  if (spec.tolerance.absolute < 0 || spec.tolerance.relative < 0) invalid('tolerance must be non-negative');
+  if (spec.grain && !GRAIN_ORDER.includes(spec.grain)) invalid('grain is not a time grain');
+}

tests/analytics/materialization-drift.test.ts

@@ -0,0 +1,327 @@
+/**
+ * tests/analytics/materialization-drift.test.ts
+ * ==============================================
+ * BI-P105 — MATERIALIZATION / PRE-AGGREGATION.
+ *
+ * The DoD asks for measured improvement and documented freshness semantics.
+ * The failure mode it does not name, and the one that actually costs money, is
+ * a pre-aggregate that returns a confident wrong number. So every test here is
+ * written as an attack on the router:
+ *
+ *   - a drifted aggregate must NOT serve, and the counterfactual (a naive
+ *     router that only checks staleness) must serve it — otherwise the guard
+ *     is not known to be doing anything;
+ *   - a MISSING bucket must count as drift even though every value present is
+ *     correct;
+ *   - a comparison over an empty window must be `insufficient_data`, never
+ *     `match`;
+ *   - an aggregate that has never been verified must not serve, because on
+ *     launch day it is exactly as unproven as a drifted one.
+ */
+import { describe, expect, it } from 'vitest';
+import {
+  applyDriftReport,
+  detectDrift,
+  type AggregateBucket,
+} from '@/src/analytics/materialization/drift-detector';
+import {
+  assertSpec,
+  describeRouting,
+  materializationStaleness,
+  selectQuerySource,
+  type MaterializationSpec,
+  type MaterializationStatus,
+} from '@/src/analytics/materialization/materialization';
+import type { AnalyticsQueryRequest } from '@/src/analytics/contract/query-request';
+
+const NOW = new Date('2026-09-07T12:00:00.000Z');
+const TOLERANCE = { absolute: 0.5, relative: 0.001 };
+
+const SPEC: MaterializationSpec = {
+  materialization_id: 'mv_sales_daily',
+  tenant_id: 't_alpha',
+  project_id: 'p_main',
+  dataset: 'ds_sales',
+  strategy: 'clickhouse_materialized_view',
+  target_ref: 'opt_sales_daily',
+  metrics: ['revenue', 'orders'],
+  dimensions: ['order_date', 'region'],
+  time_dimension: 'order_date',
+  grain: 'day',
+  max_staleness_seconds: 3600,
+  drift_check_interval_seconds: 86_400,
+  tolerance: TOLERANCE,
+};
+
+/** Fresh, recently verified, matching. The only state that may serve. */
+function healthyStatus(): MaterializationStatus {
+  return {
+    materialization_id: SPEC.materialization_id,
+    state: 'serving',
+    data_watermark: '2026-09-07T11:45:00.000Z',
+    last_refresh_at: '2026-09-07T11:46:00.000Z',
+    last_drift_check_at: '2026-09-07T06:00:00.000Z',
+    last_drift_verdict: 'match',
+  };
+}
+
+const REQUEST: AnalyticsQueryRequest = {
+  dataset: 'ds_sales',
+  metrics: ['revenue'],
+  dimensions: ['order_date'],
+  filters: [{ dimension: 'region', operator: 'eq', values: ['SE'] }],
+  timeRange: { from: '2026-09-01T00:00:00.000Z', to: '2026-09-07T00:00:00.000Z', grain: 'day' },
+};
+
+/**
+ * The router someone writes when they have thought about staleness but not
+ * about correctness. It exists here as the counterfactual: it must SERVE the
+ * drifted aggregate that `selectQuerySource` refuses.
+ */
+function naiveRouter(status: MaterializationStatus, now: Date): 'materialization' | 'base_table' {
+  const staleness = materializationStaleness(status, now) ?? 0;
+  return staleness <= SPEC.max_staleness_seconds ? 'materialization' : 'base_table';
+}
+
+function buckets(spec: Record<string, [number, number]>): AggregateBucket[] {
+  return Object.entries(spec).map(([key, [revenue, orders]]) => ({ key, values: { revenue, orders } }));
+}
+
+const SOURCE = buckets({
+  '2026-09-01|SE': [120_000, 410],
+  '2026-09-02|SE': [131_500, 442],
+  '2026-09-03|SE': [98_250, 305],
+  '2026-09-04|SE': [140_000, 480],
+});
+
+describe('BI-P105 drift detection', () => {
+  it('an identical aggregate matches', () => {
+    const report = detectDrift({
+      fromSource: SOURCE,
+      fromAggregate: SOURCE,
+      tolerance: TOLERANCE,
+      metrics: ['revenue', 'orders'],
+      now: NOW,
+    });
+    expect(report.verdict).toBe('match');
+    expect(report.buckets_compared).toBe(4);
+    expect(report.totals.revenue.delta).toBe(0);
+  });
+
+  it('a value beyond tolerance is drift, and the report says what a caller would have been told', () => {
+    const drifted = buckets({
+      '2026-09-01|SE': [120_000, 410],
+      '2026-09-02|SE': [131_500, 442],
+      '2026-09-03|SE': [98_250, 305],
+      // A source row was deleted (GDPR erasure) but the aggregate still counts it.
+      '2026-09-04|SE': [152_400, 495],
+    });
+    const report = detectDrift({
+      fromSource: SOURCE,
+      fromAggregate: drifted,
+      tolerance: TOLERANCE,
+      metrics: ['revenue', 'orders'],
+      now: NOW,
+    });
+    expect(report.verdict).toBe('drifted');
+    expect(report.worst?.key).toBe('2026-09-04|SE');
+    expect(report.worst?.absolute_delta).toBe(12_400);
+    expect(report.totals.revenue.aggregate - report.totals.revenue.source).toBe(12_400);
+  });
+
+  it('a MISSING bucket is drift even though every value present is correct', () => {
+    const lateArriving = SOURCE.filter((b) => b.key !== '2026-09-04|SE');
+    const report = detectDrift({
+      fromSource: SOURCE,
+      fromAggregate: lateArriving,
+      tolerance: TOLERANCE,
+      metrics: ['revenue', 'orders'],
+      now: NOW,
+    });
+    expect(report.discrepancies).toHaveLength(0); // every compared value agrees
+    expect(report.verdict).toBe('drifted'); // and it is still wrong
+    expect(report.missing_in_aggregate).toEqual(['2026-09-04|SE']);
+    expect(report.totals.revenue.delta).toBe(-140_000);
+  });
+
+  it('an EXTRA bucket is drift — the source rows are gone, the aggregate is not', () => {
+    const stale = [...SOURCE, { key: '2026-08-31|SE', values: { revenue: 90_000, orders: 300 } }];
+    const report = detectDrift({
+      fromSource: SOURCE,
+      fromAggregate: stale,
+      tolerance: TOLERANCE,
+      metrics: ['revenue', 'orders'],
+      now: NOW,
+    });
+    expect(report.verdict).toBe('drifted');
+    expect(report.extra_in_aggregate).toEqual(['2026-08-31|SE']);
+  });
+
+  it('an EMPTY comparison is insufficient_data, not match', () => {
+    const report = detectDrift({
+      fromSource: [],
+      fromAggregate: [],
+      tolerance: TOLERANCE,
+      metrics: ['revenue'],
+      now: NOW,
+    });
+    expect(report.verdict).toBe('insufficient_data');
+    expect(report.buckets_compared).toBe(0);
+    // And it must not be allowed to serve on the strength of that.
+    const status = applyDriftReport(healthyStatus(), report);
+    expect(status.state).toBe('drifted');
+  });
+
+  it('both tolerances must hold — a large bucket cannot hide behind a relative bound', () => {
+    const source = buckets({ 'b|SE': [40_000_000, 1] });
+    const off = buckets({ 'b|SE': [40_020_000, 1] }); // 0.05%: inside relative, far outside absolute
+    const report = detectDrift({
+      fromSource: source,
+      fromAggregate: off,
+      tolerance: { absolute: 1000, relative: 0.001 },
+      metrics: ['revenue'],
+      now: NOW,
+    });
+    expect(report.verdict).toBe('drifted');
+    expect(report.worst?.relative_delta).toBeCloseTo(0.0005, 6);
+  });
+
+  it('a bucket that should be zero and is not is drift, not a division error', () => {
+    const report = detectDrift({
+      fromSource: buckets({ 'b|SE': [0, 0] }),
+      fromAggregate: buckets({ 'b|SE': [12, 1] }),
+      tolerance: TOLERANCE,
+      metrics: ['revenue'],
+      now: NOW,
+    });
+    expect(report.verdict).toBe('drifted');
+    expect(report.worst?.relative_delta).toBe(Number.POSITIVE_INFINITY);
+  });
+});
+
+describe('BI-P105 routing — a pre-aggregate serves only while it is provably current and undrifted', () => {
+  it('serves when everything is proved', () => {
+    const decision = selectQuerySource({ request: REQUEST, spec: SPEC, status: healthyStatus(), now: NOW });
+    expect(decision.source).toBe('materialization');
+    expect(describeRouting(decision)).toMatchObject({
+      served_from: 'materialization',
+      materialization_id: 'mv_sales_daily',
+      staleness_seconds: 900,
+    });
+  });
+
+  it('a DRIFTED aggregate is refused — and the naive router serves it', () => {
+    const drifted = applyDriftReport(healthyStatus(), {
+      verdict: 'drifted',
+      checked_at: '2026-09-07T11:00:00.000Z',
+      buckets_compared: 4,
+      missing_in_aggregate: [],
+      extra_in_aggregate: [],
+      discrepancies: [],
+      tolerance: TOLERANCE,
+      totals: {},
+    });
+
+    const guarded = selectQuerySource({ request: REQUEST, spec: SPEC, status: drifted, now: NOW });
+    expect(guarded.source).toBe('base_table');
+    expect(guarded).toMatchObject({ refusal: 'state_not_serving' });
+
+    // The counterfactual: the same state, routed by staleness alone.
+    expect(naiveRouter(drifted, NOW)).toBe('materialization');
+  });
+
+  it('a NEVER-VERIFIED aggregate is refused — and the naive router serves it', () => {
+    const brandNew: MaterializationStatus = {
+      materialization_id: SPEC.materialization_id,
+      state: 'serving',
+      data_watermark: '2026-09-07T11:59:00.000Z',
+      last_refresh_at: '2026-09-07T11:59:00.000Z',
+    };
+    const guarded = selectQuerySource({ request: REQUEST, spec: SPEC, status: brandNew, now: NOW });
+    expect(guarded).toMatchObject({ source: 'base_table', refusal: 'drift_never_verified' });
+    expect(naiveRouter(brandNew, NOW)).toBe('materialization');
+  });
+
+  it('a verification older than the required interval is refused', () => {
+    const status = { ...healthyStatus(), last_drift_check_at: '2026-09-05T06:00:00.000Z' };
+    expect(selectQuerySource({ request: REQUEST, spec: SPEC, status, now: NOW })).toMatchObject({
+      refusal: 'drift_verification_stale',
+    });
+  });
+
+  it('an UNKNOWN watermark is refused rather than treated as current', () => {
+    const status = { ...healthyStatus(), data_watermark: undefined };
+    expect(materializationStaleness(status, NOW)).toBeUndefined();
+    expect(selectQuerySource({ request: REQUEST, spec: SPEC, status, now: NOW })).toMatchObject({
+      refusal: 'watermark_unknown',
+    });
+  });
+
+  it('an aggregate older than its freshness contract is refused', () => {
+    const status = { ...healthyStatus(), data_watermark: '2026-09-07T09:00:00.000Z' };
+    expect(selectQuerySource({ request: REQUEST, spec: SPEC, status, now: NOW })).toMatchObject({
+      refusal: 'staleness_exceeded',
+    });
+  });
+
+  it('a question that reaches past the watermark is refused, not silently shortened', () => {
+    const request = { ...REQUEST, timeRange: { ...REQUEST.timeRange, to: '2026-09-07T12:00:00.000Z' } };
+    expect(selectQuerySource({ request, spec: SPEC, status: healthyStatus(), now: NOW })).toMatchObject({
+      refusal: 'window_beyond_watermark',
+    });
+  });
+
+  it('a filter on a dimension the aggregate summed across is refused', () => {
+    const request = {
+      ...REQUEST,
+      filters: [{ dimension: 'sales_channel', operator: 'eq' as const, values: ['web'] }],
+    };
+    const decision = selectQuerySource({ request, spec: SPEC, status: healthyStatus(), now: NOW });
+    expect(decision).toMatchObject({ source: 'base_table', refusal: 'coverage_miss' });
+    expect((decision as { detail: string }).detail).toMatch(/already summed across it/);
+  });
+
+  it('a metric or dimension the aggregate does not carry is refused', () => {
+    expect(
+      selectQuerySource({
+        request: { ...REQUEST, metrics: ['margin'] },
+        spec: SPEC,
+        status: healthyStatus(),
+        now: NOW,
+      }),
+    ).toMatchObject({ refusal: 'coverage_miss' });
+    expect(
+      selectQuerySource({
+        request: { ...REQUEST, dimensions: ['product'] },
+        spec: SPEC,
+        status: healthyStatus(),
+        now: NOW,
+      }),
+    ).toMatchObject({ refusal: 'coverage_miss' });
+  });
+
+  it('a daily aggregate cannot answer an hourly question, but can answer a monthly one', () => {
+    const hourly = { ...REQUEST, timeRange: { ...REQUEST.timeRange, grain: 'hour' as const } };
+    expect(
+      selectQuerySource({ request: hourly, spec: SPEC, status: healthyStatus(), now: NOW }),
+    ).toMatchObject({ refusal: 'coverage_miss' });
+
+    const monthly = { ...REQUEST, timeRange: { ...REQUEST.timeRange, grain: 'month' as const } };
+    expect(
+      selectQuerySource({ request: monthly, spec: SPEC, status: healthyStatus(), now: NOW }).source,
+    ).toBe('materialization');
+  });
+
+  it('a WEEKLY aggregate cannot answer a monthly question — weeks do not compose into months', () => {
+    const weekly = { ...SPEC, grain: 'week' as const };
+    const monthly = { ...REQUEST, timeRange: { ...REQUEST.timeRange, grain: 'month' as const } };
+    expect(
+      selectQuerySource({ request: monthly, spec: weekly, status: healthyStatus(), now: NOW }),
+    ).toMatchObject({ refusal: 'coverage_miss' });
+  });
+
+  it('a spec with no drift check interval is refused at declaration time', () => {
+    expect(() => assertSpec({ ...SPEC, drift_check_interval_seconds: 0 })).toThrow(/never re-verified/);
+    expect(() => assertSpec(SPEC)).not.toThrow();
+  });
+});

tests/integration/materialization-drift-clickhouse.integration.test.ts

@@ -0,0 +1,348 @@
+/**
+ * tests/integration/materialization-drift-clickhouse.integration.test.ts
+ * =======================================================================
+ * BI-P105 against a LIVE ClickHouse. Skips honestly when CLICKHOUSE_URL is
+ * unset; with it set and the server down, it FAILS rather than skipping.
+ *
+ * BI-P105's DoD is "performance improvement measured and freshness semantics
+ * documented", and both halves are done here against real data rather than
+ * asserted. What the DoD does not say, and what this file is really about, is
+ * that a pre-aggregate fails SILENTLY: it answers instantly and wrongly.
+ *
+ * The drill drives a real ClickHouse materialized view through the exact
+ * sequence that produces a confident wrong number in production:
+ *
+ *   1. 2M source rows, a SummingMergeTree target and a materialized view.
+ *   2. Backfill, verify with the SHIPPED detector: verdict `match`.
+ *   3. Measure both paths repeatedly, and report the DISTRIBUTION — a single
+ *      speed-up figure quoted as a system property is misleading (BI-P104).
+ *   4. `ALTER TABLE ... DELETE` on the source: a GDPR erasure, the single most
+ *      common real cause of pre-aggregate drift. A ClickHouse materialized
+ *      view is an INSERT trigger; it does not see mutations, so the target
+ *      keeps the deleted revenue forever and nothing errors.
+ *   5. The detector must catch it, and `selectQuerySource` must refuse to
+ *      serve. The counterfactual — a router that checks only staleness — is
+ *      run on the same state and MUST serve, with the wrong number recorded
+ *      in kronor.
+ *
+ * Everything measured lands in evidence/bi-p105/materialization-drift.json.
+ *
+ *   docker run -d --name ch -p 28123:8123 -e CLICKHOUSE_PASSWORD=test \
+ *     clickhouse/clickhouse-server:26.3
+ *   CLICKHOUSE_URL=http://127.0.0.1:28123 CLICKHOUSE_AUTH=default:test \
+ *     pnpm vitest run tests/integration/materialization-drift-clickhouse.integration.test.ts
+ *
+ * It CREATEs and DROPs its own database. Never point it at the b0ts
+ * ClickHouse or any quarantined market data.
+ */
+import fs from 'node:fs';
+import path from 'node:path';
+import { afterAll, beforeAll, describe, expect, it } from 'vitest';
+import {
+  applyDriftReport,
+  detectDrift,
+  type AggregateBucket,
+  type DriftReport,
+} from '@/src/analytics/materialization/drift-detector';
+import {
+  materializationStaleness,
+  selectQuerySource,
+  type MaterializationSpec,
+  type MaterializationStatus,
+} from '@/src/analytics/materialization/materialization';
+import type { AnalyticsQueryRequest } from '@/src/analytics/contract/query-request';
+
+const CLICKHOUSE_URL = process.env.CLICKHOUSE_URL;
+const CLICKHOUSE_AUTH = process.env.CLICKHOUSE_AUTH;
+const run = CLICKHOUSE_URL ? describe : describe.skip;
+const DB = `opt_p105_${Date.now().toString(36)}`;
+const ROWS = 2_000_000;
+const REPETITIONS = 9;
+
+const headers: Record<string, string> = CLICKHOUSE_AUTH
+  ? { Authorization: `Basic ${Buffer.from(CLICKHOUSE_AUTH).toString('base64')}` }
+  : {};
+
+async function ch(query: string): Promise<string> {
+  const res = await fetch(CLICKHOUSE_URL ?? 'http://skipped.invalid', {
+    method: 'POST',
+    headers,
+    body: query,
+  });
+  const text = await res.text();
+  if (!res.ok) throw new Error(`ClickHouse ${res.status}: ${text.slice(0, 400)}`);
+  return text;
+}
+
+/** Rows as buckets keyed "date|region", which is what the detector compares. */
+async function bucketsFrom(query: string): Promise<AggregateBucket[]> {
+  const text = await ch(`${query} FORMAT JSONEachRow`);
+  return text
+    .trim()
+    .split('\n')
+    .filter(Boolean)
+    .map((line) => {
+      const r = JSON.parse(line) as { d: string; region: string; revenue: number; orders: number };
+      return { key: `${r.d}|${r.region}`, values: { revenue: Number(r.revenue), orders: Number(r.orders) } };
+    });
+}
+
+/** Median and the spread around it. A single number is not a system property. */
+function distribution(samples: number[]): {
+  n: number;
+  min: number;
+  p50: number;
+  p95: number;
+  max: number;
+  spread_ratio: number;
+} {
+  const s = [...samples].sort((a, b) => a - b);
+  const at = (q: number) => s[Math.min(s.length - 1, Math.floor(q * s.length))];
+  return {
+    n: s.length,
+    min: s[0],
+    p50: at(0.5),
+    p95: at(0.95),
+    max: s[s.length - 1],
+    spread_ratio: Number((s[s.length - 1] / Math.max(s[0], 1e-9)).toFixed(2)),
+  };
+}
+
+async function timeQuery(query: string): Promise<number> {
+  const t0 = performance.now();
+  await ch(`${query} FORMAT JSONEachRow`);
+  return performance.now() - t0;
+}
+
+const SOURCE_AGG = `
+  SELECT toString(order_date) AS d, region, sum(amount) AS revenue, count() AS orders
+  FROM ${DB}.sales WHERE tenant_id = 't_alpha' GROUP BY d, region ORDER BY d, region`;
+const TARGET_AGG = `
+  SELECT toString(order_date) AS d, region, sum(revenue) AS revenue, sum(orders) AS orders
+  FROM ${DB}.sales_daily WHERE tenant_id = 't_alpha' GROUP BY d, region ORDER BY d, region`;
+
+const SPEC: MaterializationSpec = {
+  materialization_id: 'mv_sales_daily',
+  tenant_id: 't_alpha',
+  project_id: 'p_main',
+  dataset: 'ds_sales',
+  strategy: 'clickhouse_materialized_view',
+  target_ref: `${DB}.sales_daily`,
+  metrics: ['revenue', 'orders'],
+  dimensions: ['order_date', 'region'],
+  time_dimension: 'order_date',
+  grain: 'day',
+  max_staleness_seconds: 3600,
+  drift_check_interval_seconds: 86_400,
+  tolerance: { absolute: 0.5, relative: 0.0001 },
+};
+
+const REQUEST: AnalyticsQueryRequest = {
+  dataset: 'ds_sales',
+  metrics: ['revenue'],
+  dimensions: ['order_date'],
+  filters: [],
+  timeRange: { grain: 'day' },
+};
+
+const evidence: Record<string, unknown> = {};
+let status: MaterializationStatus;
+
+run('BI-P105 materialization against a LIVE ClickHouse', () => {
+  beforeAll(async () => {
+    evidence.clickhouse_version = (await ch('SELECT version()')).trim();
+    evidence.recorded_at = new Date().toISOString();
+    evidence.source_rows = ROWS;
+
+    await ch(`CREATE DATABASE IF NOT EXISTS ${DB}`);
+    await ch(`
+      CREATE TABLE ${DB}.sales (
+        tenant_id LowCardinality(String),
+        order_date Date,
+        region LowCardinality(String),
+        customer_id UInt32,
+        amount Decimal(18, 2)
+      ) ENGINE = MergeTree ORDER BY (tenant_id, order_date, region)`);
+    await ch(`
+      INSERT INTO ${DB}.sales
+      SELECT
+        if(number % 5 = 0, 't_beta', 't_alpha') AS tenant_id,
+        toDate('2026-08-01') + toIntervalDay(number % 30) AS order_date,
+        ['SE','NO','DK','FI'][(number % 4) + 1] AS region,
+        toUInt32(number % 5000) AS customer_id,
+        toDecimal64((number % 977) + 13, 2) AS amount
+      FROM numbers(${ROWS})`);
+
+    await ch(`
+      CREATE TABLE ${DB}.sales_daily (
+        tenant_id LowCardinality(String),
+        order_date Date,
+        region LowCardinality(String),
+        revenue Decimal(38, 2),
+        orders UInt64
+      ) ENGINE = SummingMergeTree ORDER BY (tenant_id, order_date, region)`);
+    await ch(`
+      CREATE MATERIALIZED VIEW ${DB}.sales_daily_mv TO ${DB}.sales_daily AS
+      SELECT tenant_id, order_date, region, sum(amount) AS revenue, count() AS orders
+      FROM ${DB}.sales GROUP BY tenant_id, order_date, region`);
+    // The MV is an INSERT trigger: it saw nothing that was already there, so
+    // the backfill is explicit. (This is drift cause #1, and it is why the
+    // first check below has to be able to fail.)
+    await ch(`
+      INSERT INTO ${DB}.sales_daily
+      SELECT tenant_id, order_date, region, sum(amount) AS revenue, count() AS orders
+      FROM ${DB}.sales GROUP BY tenant_id, order_date, region`);
+    await ch(`OPTIMIZE TABLE ${DB}.sales_daily FINAL`);
+
+    status = {
+      materialization_id: SPEC.materialization_id,
+      state: 'serving',
+      data_watermark: new Date().toISOString(),
+      last_refresh_at: new Date().toISOString(),
+    };
+  }, 300_000);
+
+  afterAll(async () => {
+    if (!CLICKHOUSE_URL) return;
+    try {
+      await ch(`DROP DATABASE IF EXISTS ${DB}`);
+    } finally {
+      const dir = path.join(process.cwd(), 'evidence/bi-p105');
+      fs.mkdirSync(dir, { recursive: true });
+      fs.writeFileSync(
+        path.join(dir, 'materialization-drift.json'),
+        `${JSON.stringify(evidence, null, 2)}\n`,
+      );
+    }
+  }, 60_000);
+
+  it('the freshly backfilled aggregate MATCHES its source', async () => {
+    const report = detectDrift({
+      fromSource: await bucketsFrom(SOURCE_AGG),
+      fromAggregate: await bucketsFrom(TARGET_AGG),
+      tolerance: SPEC.tolerance,
+      metrics: ['revenue', 'orders'],
+    });
+    expect(report.buckets_compared).toBeGreaterThan(0); // an empty compare proves nothing
+    expect(report.verdict).toBe('match');
+
+    status = applyDriftReport(status, report);
+    evidence.after_backfill = {
+      verdict: report.verdict,
+      buckets_compared: report.buckets_compared,
+      revenue_total: report.totals.revenue,
+    };
+  }, 120_000);
+
+  it('measures the improvement as a DISTRIBUTION, not a single number', async () => {
+    const base: number[] = [];
+    const agg: number[] = [];
+    for (let i = 0; i < REPETITIONS; i += 1) {
+      // Interleaved so a machine that gets busy halfway through penalises both.
+      base.push(await timeQuery(SOURCE_AGG));
+      agg.push(await timeQuery(TARGET_AGG));
+    }
+    const baseD = distribution(base);
+    const aggD = distribution(agg);
+    evidence.performance = {
+      note:
+        'ClickHouse caches marks and pages between repetitions, so these are WARM numbers on one host ' +
+        'with nothing else running. They describe this shape of query on this data, not a system property.',
+      repetitions: REPETITIONS,
+      base_table_ms: baseD,
+      pre_aggregate_ms: aggD,
+      speedup_p50: Number((baseD.p50 / Math.max(aggD.p50, 1e-9)).toFixed(1)),
+      speedup_range_min_max: [
+        Number((baseD.min / Math.max(aggD.max, 1e-9)).toFixed(1)),
+        Number((baseD.max / Math.max(aggD.min, 1e-9)).toFixed(1)),
+      ],
+    };
+    // The claim under test is only that the pre-aggregate is not SLOWER; the
+    // magnitude is reported, never asserted as a threshold.
+    expect(aggD.p50).toBeLessThanOrEqual(baseD.p50);
+  }, 300_000);
+
+  it('a GDPR erasure on the source DRIFTS the aggregate, and the drift is caught', async () => {
+    const before = await bucketsFrom(TARGET_AGG);
+
+    // The mutation a right-to-erasure request produces. A materialized view is
+    // an insert trigger — it never sees this.
+    await ch(
+      `ALTER TABLE ${DB}.sales DELETE WHERE tenant_id = 't_alpha' AND customer_id < 250 SETTINGS mutations_sync = 2`,
+    );
+
+    const fromSource = await bucketsFrom(SOURCE_AGG);
+    const fromAggregate = await bucketsFrom(TARGET_AGG);
+    expect(fromAggregate).toEqual(before); // the aggregate genuinely did not move
+
+    const report: DriftReport = detectDrift({
+      fromSource,
+      fromAggregate,
+      tolerance: SPEC.tolerance,
+      metrics: ['revenue', 'orders'],
+    });
+    expect(report.verdict).toBe('drifted');
+    expect(report.worst).toBeDefined();
+
+    const drifted = applyDriftReport(status, report);
+    const guarded = selectQuerySource({ request: REQUEST, spec: SPEC, status: drifted, now: new Date() });
+    expect(guarded.source).toBe('base_table');
+
+    // COUNTERFACTUAL: a router that checks only staleness serves it, and this
+    // is the number a dashboard would have shown.
+    const stalenessOnlyServes =
+      (materializationStaleness(drifted, new Date()) ?? 0) <= SPEC.max_staleness_seconds;
+    expect(stalenessOnlyServes).toBe(true);
+
+    evidence.gdpr_erasure_drill = {
+      mutation: "ALTER TABLE sales DELETE WHERE tenant_id='t_alpha' AND customer_id < 250",
+      verdict: report.verdict,
+      buckets_compared: report.buckets_compared,
+      revenue_true_after_erasure: report.totals.revenue.source,
+      revenue_pre_aggregate_would_report: report.totals.revenue.aggregate,
+      revenue_overstatement: report.totals.revenue.delta,
+      revenue_overstatement_pct: Number(
+        ((report.totals.revenue.delta / report.totals.revenue.source) * 100).toFixed(3),
+      ),
+      worst_bucket: report.worst,
+      guarded_router_decision: guarded,
+      staleness_only_router_would_serve: stalenessOnlyServes,
+      staleness_seconds_at_decision: materializationStaleness(drifted, new Date()),
+      note:
+        'The aggregate was NOT stale by any clock: it had just been refreshed, and every freshness check ' +
+        'passed. Only the source comparison caught it. This is why staleness alone is not a correctness ' +
+        'guarantee.',
+    };
+    status = drifted;
+  }, 300_000);
+
+  it('and a refresh that re-derives the aggregate clears the drift', async () => {
+    await ch(`TRUNCATE TABLE ${DB}.sales_daily`);
+    await ch(`
+      INSERT INTO ${DB}.sales_daily
+      SELECT tenant_id, order_date, region, sum(amount) AS revenue, count() AS orders
+      FROM ${DB}.sales GROUP BY tenant_id, order_date, region`);
+    await ch(`OPTIMIZE TABLE ${DB}.sales_daily FINAL`);
+
+    const report = detectDrift({
+      fromSource: await bucketsFrom(SOURCE_AGG),
+      fromAggregate: await bucketsFrom(TARGET_AGG),
+      tolerance: SPEC.tolerance,
+      metrics: ['revenue', 'orders'],
+    });
+    expect(report.verdict).toBe('match');
+
+    const healed: MaterializationStatus = applyDriftReport(
+      { ...status, state: 'serving', data_watermark: new Date().toISOString() },
+      report,
+    );
+    const decision = selectQuerySource({ request: REQUEST, spec: SPEC, status: healed, now: new Date() });
+    expect(decision.source).toBe('materialization');
+
+    evidence.after_rebuild = {
+      verdict: report.verdict,
+      revenue_total: report.totals.revenue,
+      routing: decision,
+    };
+  }, 300_000);
+});