Lineage family lookup: bound it to a slate, and say what an action is

TWO DEFECTS, one lookup.

SCALE. The family lookup sent 100 natural keys as a PostgREST IN-list.
`read_natural_key` has NO pg_stats row at all -- the table's last autoanalyze
(2026-08-26) predates the column ever being populated -- so the planner used a
default per-value selectivity, estimated 172,409 rows and chose a sequential
scan of 344,818: 8.5s, then 57014. At 50 keys the same shape returned in
~357ms. The cliff is a statistics artifact, not a volume one, which is why the
repair does not depend on the estimate improving and is not CH=50.

`readNaturalKey` builds `sport|game_date|player_key|stat|side|line[|#event]`,
so SPORT AND GAME_DATE ARE COMPONENTS OF THE KEY. Two rows sharing a key
necessarily share both, and scoping the lookup to the (sport, game_date) pairs
present in the requested keys is LOSSLESS BY CONSTRUCTION. One index-backed
range per date, walked with safePaginate; cost is bounded by ONE SLATE however
long the chronology gets. Measured: 5,000 keys -> 1 scope, and the plan is
`Index Scan using model_snapshots_lineage_family_idx, cost 0.28..1.92`.

VALIDITY. A row carrying `read_natural_key` is not history: the key is stamped
on every candidate BEFORE the lookup, so a failure leaves it on a row that
never became an action. Proven this was not cosmetic -- fed the raw rows the
old lookup returned, the resolver produced a REVISION with a NULL read_id (an
orphaned chain node) and labelled a brand-new Read LEGACY_UNVERIFIED.
`isValidLineageAction` states what a completed action IS: all nine fields, in
the query and again in code.

ATOMICITY. A failed attempt now leaves NO lineage-specific state.
`publication_id`/`published_at` are untouched -- the slate really was
published, and erasing a true fact to tidy a false one is the wrong repair.

Replayed the exact failed 19:00Z cohort through the real resolver, side-effect
free: 119 NEW / 379 CHANGED / 621 UNCHANGED -> ORIGIN 119 / REVISION 379 /
RECAPTURE 621, 0 wrong parent, 0 wrong ordinal, 0 null read_id, 0 forks --
byte-identical with all 1,119 failed partial rows present. Clean-head parity
1,024/1,024.

Migration 050 is CONCURRENTLY + IF NOT EXISTS, drops nothing, rewrites nothing.
Lineage stays OFF.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CQJeAG8vcDoL5zkiaJyVb8

src/services/retentionService.js

@@ -259,6 +259,71 @@ const LINEAGE_KEYS = Object.freeze([
   'digest_algorithm_version',
 ]);
 
+/**
+ * ── WHAT A COMPLETED LINEAGE ACTION IS ───────────────────────────────────
+ * A row carrying `read_natural_key` is NOT lineage history. The key is stamped
+ * on every candidate BEFORE the family lookup runs, so a failed resolution
+ * leaves it behind on a row that never became an action.
+ *
+ * A completed action is a row that answers all four questions the chronology
+ * asks of it: WHICH Read (`read_id`), WHAT it did (`lineage_action`), WHAT it
+ * claimed (`claim_digest` + the two version stamps that make the digest
+ * interpretable), and WHERE it sits (`revision_ordinal`). Miss any one and the
+ * row cannot serve as history, a head, a parent or a recapture predecessor.
+ *
+ * This is stated as what an action IS — not as a rule shaped to exclude one
+ * known batch of failed rows.
+ */
+const VALID_LINEAGE_ACTION_FIELDS = Object.freeze([
+  'read_id', 'read_natural_key', 'lineage_action', 'claim_digest',
+  'revision_ordinal', 'lineage_state', 'lineage_version',
+  'claim_schema_version', 'digest_algorithm_version',
+]);
+
+/** Does this physical row qualify as a completed lineage action? */
+function isValidLineageAction(row) {
+  if (!row) return false;
+  for (const f of VALID_LINEAGE_ACTION_FIELDS) {
+    const v = row[f];
+    if (v === null || v === undefined || v === '') return false;
+  }
+  return true;
+}
+
+/**
+ * ── WHY THE FAMILY IS FETCHED BY (sport, game_date) ──────────────────────
+ * `readLineage.readNaturalKey` builds the key as
+ *   sport | game_date | player_key | stat | side | line [| #event]
+ * so SPORT AND GAME_DATE ARE COMPONENTS OF THE KEY ITSELF. Two rows sharing a
+ * natural key necessarily share both. Restricting the lookup to the (sport,
+ * game_date) pairs present in the requested keys is therefore LOSSLESS BY
+ * CONSTRUCTION, not an optimisation that trades recall for speed. A test
+ * asserts the derivation against the key builder so the two cannot drift.
+ *
+ * That bound is what makes the read scale with THE SLATE rather than with
+ * history: one date's valid actions, however many years of chronology sit
+ * behind it.
+ *
+ * WHY NOT KEEP THE IN-LIST. Measured 2026-08-28: `read_natural_key` has NO
+ * pg_stats row at all (the last autoanalyze predates the column ever being
+ * populated), so the planner falls back to a default per-value selectivity.
+ * At 100 keys it estimated 172,409 rows and chose a sequential scan of 344,818
+ * — 8.5s, then 57014. The plan for this shape is
+ * `Index Scan using model_snapshots_lineage_family_idx, cost 0.28..1.92`,
+ * and it does not depend on an estimate being good.
+ */
+function familyScopesFrom(naturalKeys) {
+  const scopes = new Map();
+  for (const k of naturalKeys || []) {
+    const parts = String(k).split('|');
+    const sport = parts[0];
+    const gameDate = parts[1];
+    if (!sport || !gameDate) continue;
+    scopes.set(`${sport}|${gameDate}`, { sport, game_date: gameDate });
+  }
+  return [...scopes.values()];
+}
+
 /** Claim fields the resolver needs from EXISTING rows to classify a change. */
 const LINEAGE_FETCH_CLAIM = Object.freeze([
   'canonical_event_id',
@@ -335,26 +400,49 @@ async function attachLineage(rows, deps = {}) {
       const getClient = deps.getClient || require('../utils/supabase').getSupabaseServiceClient;
       const supabase = getClient();
       if (!supabase) return null;
+      const { paginate } = require('../utils/safePaginate');
+      const columns = ['id', 'read_id', 'read_natural_key', 'game_id', 'claim_digest',
+        'revision_ordinal', 'lineage_action', 'supersedes_id', 'captured_at',
+        'lineage_state', 'lineage_version', 'claim_schema_version',
+        'digest_algorithm_version', ...LINEAGE_FETCH_CLAIM].join(', ');
+      const wanted = new Set(naturalKeys);
       const found = [];
-      const CH = 100; // never send an unbounded id list — it becomes a URL
-      for (let i = 0; i < naturalKeys.length; i += CH) {
-        const { data, error } = await supabase
+      out.lookup = { scopes: 0, rows_scanned: 0, valid_actions: 0, invalid_excluded: 0 };
+      for (const scope of familyScopesFrom(naturalKeys)) {
+        // ONE index-backed range per (sport, game_date), walked with the
+        // repository's keyset paginator — never a hand-rolled second walk, and
+        // it THROWS rather than treating a failed page as end-of-data.
+        // eslint-disable-next-line no-await-in-loop
+        const rows = await paginate(() => supabase
           .from('model_snapshots')
-          .select(['id', 'read_id', 'read_natural_key', 'game_id', 'claim_digest',
-            'revision_ordinal', 'lineage_action', 'supersedes_id', 'captured_at',
-            ...LINEAGE_FETCH_CLAIM].join(', '))
-          .in('read_natural_key', naturalKeys.slice(i, i + CH));
-        if (error) throw new Error(error.message);
-        if (Array.isArray(data)) found.push(...data);
+          .select(columns)
+          .eq('sport', scope.sport)
+          .eq('game_date', scope.game_date)
+          .not('lineage_action', 'is', null), { key: 'id', label: 'attachLineage.fetchExisting' });
+        out.lookup.scopes += 1;
+        out.lookup.rows_scanned += rows.length;
+        for (const r of rows) {
+          if (!wanted.has(r.read_natural_key)) continue;
+          // The predicate is applied in code as well as in the query. The query
+          // narrows what travels; this decides what COUNTS, and a row that half
+          // resolved must never be mistaken for history by either.
+          if (!isValidLineageAction(r)) { out.lookup.invalid_excluded += 1; continue; }
+          out.lookup.valid_actions += 1;
+          found.push(r);
+        }
       }
       return found;
     });
 
     const existing = await fetchExisting(keys);
     if (existing === null) return out; // no database configured — leave NULL
+    // Second line of defence: an injected or future fetcher must not be able to
+    // smuggle a half-resolved row into the chronology either.
+    const validExisting = existing.filter(isValidLineageAction);
+    out.invalid_rows_excluded = existing.length - validExisting.length;
 
     const byKey = new Map();
-    for (const e of existing) {
+    for (const e of validExisting) {
       // The prior claim rides alongside so `classifyChange` can say WHY the
       // published state advanced, not merely that the digest differs.
       const claim = {};
@@ -413,6 +501,26 @@ async function attachLineage(rows, deps = {}) {
   } catch (e) {
     out.error = e && e.message ? e.message : String(e);
   }
+
+  // ── FAILURE ATOMICITY ───────────────────────────────────────────────────
+  // `read_natural_key` is stamped on every candidate BEFORE the family lookup,
+  // so a lookup that throws leaves it behind on a row that never became an
+  // action. That is the shape the 2026-08-28 scheduled canary wrote 1,119 times.
+  //
+  // It is LINEAGE FAMILY IDENTITY, not publication provenance, so a failed
+  // attempt has no claim to it. `publication_id` / `published_at` are NOT
+  // touched here: the slate really was published, and erasing that to make the
+  // failure look tidier would delete a true fact to hide a false one.
+  //
+  // The result: a failed row carries no lineage-specific state at all, so it
+  // cannot be mistaken for history by a future resolver, a query, or a reader.
+  let cleared = 0;
+  for (const r of list) {
+    if (isValidLineageAction(r)) continue;
+    if (LINEAGE_KEYS.some((k) => r[k] !== null && r[k] !== undefined)) cleared += 1;
+    Object.assign(r, blankLineage());
+  }
+  out.incomplete_cleared = cleared;
   return out;
 }
 
@@ -1063,6 +1171,9 @@ module.exports = {
   persist,
   attachLineage,
   commitPublication,
+  isValidLineageAction,
+  VALID_LINEAGE_ACTION_FIELDS,
+  familyScopesFrom,
   recoverFromFork,
   isSupersedesConflict,
   FORK_RETRY_LIMIT,

supabase/migrations/050_lineage_family_lookup_index.sql

@@ -0,0 +1,27 @@
+-- 048b — supporting index for the lineage family lookup.
+--
+-- WHY. The family lookup was `read_natural_key IN (<100 keys>)`. Measured
+-- 2026-08-28: `read_natural_key` has NO pg_stats row at all (the table's last
+-- autoanalyze predates the column ever being populated), so the planner used a
+-- default per-value selectivity, estimated 172,409 rows for 100 keys, and chose
+-- a sequential scan of 344,818 rows -- 8.5s, then 57014. At 50 keys the same
+-- shape planned differently and returned in ~357ms. That cliff is a statistics
+-- artifact, not a data-volume one, which is why the repair does not depend on
+-- the estimate improving.
+--
+-- The lookup is now scoped by (sport, game_date) -- both are COMPONENTS OF THE
+-- NATURAL KEY ITSELF, so the bound is lossless by construction -- and reads only
+-- COMPLETED lineage actions. This index serves exactly that shape:
+--
+--   Index Scan using model_snapshots_lineage_family_idx  (cost=0.28..1.92)
+--
+-- Its size grows with completed lineage actions, and the query's date bound
+-- means a lookup scans ONE slate's worth however long the chronology gets.
+--
+-- FORWARD-ONLY and NON-DESTRUCTIVE: CONCURRENTLY (no table lock, no rewrite),
+-- IF NOT EXISTS (safe to re-run), and no existing index is dropped.
+-- Built on production 2026-08-28: 128 kB over 1,024 covered rows, indisvalid.
+
+create index concurrently if not exists model_snapshots_lineage_family_idx
+  on public.model_snapshots (game_date, sport, read_natural_key)
+  where lineage_action is not null;

tests/unit/eventPublication.test.js

@@ -265,10 +265,18 @@ describe('PUBLICATION COMMIT TRUTH', () => {
     await retention.commitPublication({ rows, publicationId: 's', publishedAt: 't' }, deps);
     const first = rows[0].claim_digest;
     // Second run sees the first as existing history.
+    // A COMPLETED lineage action, in the shape production actually persists.
+    // `attachLineage` assigns the action, the state and both version stamps in
+    // one block, and all 1,024 live rows carry all nine fields — a digest with
+    // no schema version is uninterpretable, so the validity predicate requires
+    // them. This fixture previously omitted four of them, describing a row the
+    // writer cannot emit.
     const existing = [{
       id: 1, read_id: rows[0].read_id, read_natural_key: rows[0].read_natural_key,
       game_id: rows[0].game_id, claim_digest: first, revision_ordinal: 0,
       lineage_action: 'ORIGIN', supersedes_id: null, captured_at: rows[0].captured_at,
+      lineage_state: 'LIVE', lineage_version: 'lin@1',
+      claim_schema_version: 'claim@1', digest_algorithm_version: 'sha256-json-sorted@1',
       claim: rows[0],
     }];
     const rows2 = [served()];

tests/unit/lineageFamilyLookup.test.js

@@ -0,0 +1,342 @@
+// LINEAGE FAMILY LOOKUP — validity, scale and failure atomicity.
+//
+// The first canary began from an EMPTY graph and passed. The second failed
+// because history existed: the family lookup sent 100 natural keys as a
+// PostgREST IN-list, the planner had no statistics for `read_natural_key`
+// (its last autoanalyze predated the column ever being populated), estimated
+// 172,409 rows and chose a sequential scan of 344,818 -- 8.5s, then 57014.
+//
+// The repair is not a smaller chunk. It is a lookup whose cost is bounded by
+// ONE SLATE, plus a predicate that says what a completed lineage action IS.
+
+const ret = require('../../src/services/retentionService');
+const R = require('../../src/services/read/readLineage');
+const { isValidLineageAction, VALID_LINEAGE_ACTION_FIELDS, familyScopesFrom } = ret;
+
+const CAND = {
+  sport: 'mlb', game_date: '2026-08-28', player_key: 'aaron judge', stat: 'hits',
+  side: 'over', line: 0.5, canonical_event_id: 'mlb:gamepk:1',
+  captured_at: '2026-08-28T23:00:00.000Z', published: true, p_win: 0.9, grade: 'B+',
+};
+const KEY = R.readNaturalKey(CAND);
+
+const validRow = (over = {}) => ({
+  id: 1, read_id: '11111111-1111-4111-8111-111111111111', read_natural_key: KEY,
+  game_id: 'g', claim_digest: 'OLDDIGEST', revision_ordinal: 0, lineage_action: 'ORIGIN',
+  supersedes_id: null, captured_at: '2026-08-28T19:00:00.000Z', lineage_state: 'LIVE',
+  lineage_version: 'lin@1', claim_schema_version: 'claim@1',
+  digest_algorithm_version: 'sha256-json-sorted@1',
+  // Claim fields sit at TOP LEVEL, exactly as the lookup projection returns
+  // them: `attachLineage` rebuilds the nested `claim` from these columns, so a
+  // fixture that only nests them is testing a shape the database never emits.
+  canonical_event_id: 'mlb:gamepk:1', line: 0.5, side: 'over', p_win: 0.5, grade: 'C',
+  ...over,
+});
+// The EXACT shape the 2026-08-28 failure wrote 1,119 times: a natural key and
+// publication stamps, every lineage action field NULL.
+const partialRow = (over = {}) => ({
+  id: 2, read_id: null, read_natural_key: KEY, game_id: 'g', claim_digest: null,
+  revision_ordinal: null, lineage_action: null, supersedes_id: null,
+  captured_at: '2026-08-28T23:00:00.000Z', lineage_state: null, lineage_version: null,
+  claim_schema_version: null, digest_algorithm_version: null,
+  publication_id: 'pub-1', published_at: '2026-08-28T19:03:29.579Z',
+  canonical_event_id: 'mlb:gamepk:1', line: 0.5, side: 'over',
+  ...over,
+});
+
+describe('STEP 2 — semantic result and persisted action are different layers', () => {
+  test('the repository maps NEW/CHANGED/UNCHANGED onto ORIGIN/REVISION/RECAPTURE', () => {
+    const noHistory = R.resolveLineage({ candidate: CAND, existing: [], mintReadId: () => 'M' });
+    expect(noHistory.change_type).toBe(R.CHANGE_TYPE.INITIAL_PUBLICATION);
+    expect(noHistory.action).toBe(R.LINEAGE_ACTION.ORIGIN);
+
+    const head = validRow();
+    const changed = R.resolveLineage({ candidate: CAND, existing: [head], mintReadId: () => 'M' });
+    expect(changed.action).toBe(R.LINEAGE_ACTION.REVISION);
+    expect(changed.change_type).not.toBe(R.CHANGE_TYPE.NO_MATERIAL_PUBLISHED_CHANGE);
+
+    const same = R.resolveLineage({
+      candidate: CAND,
+      existing: [validRow({ claim_digest: R.claimDigest(CAND) })],
+      mintReadId: () => 'M',
+    });
+    expect(same.action).toBe(R.LINEAGE_ACTION.RECAPTURE);
+    expect(same.change_type).toBe(R.CHANGE_TYPE.NO_MATERIAL_PUBLISHED_CHANGE);
+  });
+
+  test('NO_MATERIAL_CHANGE and RECAPTURE describe ONE claim, not two', () => {
+    // The semantic result and the persisted action are two views of the same
+    // row. Counting them as separate populations would double-report a slate.
+    const r = R.resolveLineage({
+      candidate: CAND,
+      existing: [validRow({ claim_digest: R.claimDigest(CAND) })],
+      mintReadId: () => 'M',
+    });
+    expect(r.action).toBe('RECAPTURE');
+    expect(r.change_type).toBe('NO_MATERIAL_PUBLISHED_CHANGE');
+    // One resolution object. One claim.
+    expect(Object.keys(r).filter((k) => k === 'action' || k === 'change_type')).toHaveLength(2);
+  });
+});
+
+describe('STEP 5 — VALID_LINEAGE_ACTION_PREDICATE', () => {
+  test('it states what a completed action IS, field by field', () => {
+    expect(VALID_LINEAGE_ACTION_FIELDS).toEqual(expect.arrayContaining([
+      'read_id', 'read_natural_key', 'lineage_action', 'claim_digest',
+      'revision_ordinal', 'lineage_state', 'lineage_version',
+      'claim_schema_version', 'digest_algorithm_version',
+    ]));
+    expect(isValidLineageAction(validRow())).toBe(true);
+    // Removing ANY required field disqualifies the row — the predicate is not
+    // shaped around one known batch.
+    for (const f of VALID_LINEAGE_ACTION_FIELDS) {
+      expect(isValidLineageAction({ ...validRow(), [f]: null })).toBe(false);
+    }
+  });
+
+  test('a read_natural_key alone is NOT lineage history', () => {
+    expect(isValidLineageAction({ read_natural_key: KEY })).toBe(false);
+    expect(isValidLineageAction(partialRow())).toBe(false);
+    // Publication provenance does not make it one either.
+    expect(isValidLineageAction({ read_natural_key: KEY, publication_id: 'p', published_at: 'x' })).toBe(false);
+  });
+
+  test('ordinal 0 is a real ordinal, not an absence', () => {
+    expect(isValidLineageAction(validRow({ revision_ordinal: 0 }))).toBe(true);
+  });
+});
+
+describe('STEP 7/8 — partial rows may not shadow a valid head', () => {
+  const withClaim = (r) => ({ ...r, claim: { canonical_event_id: r.canonical_event_id, line: r.line, side: r.side, p_win: r.p_win, grade: r.grade } });
+  const resolve = (existing) => R.resolveLineage({
+    candidate: CAND, existing: existing.filter(isValidLineageAction).map(withClaim), mintReadId: () => 'MINT',
+  });
+
+  test('A valid ORIGIN + newer partial row, partial FIRST: head, parent and read_id are correct', () => {
+    const r = resolve([partialRow(), validRow()]);
+    expect(r.action).toBe('REVISION');
+    expect(r.read_id).toBe('11111111-1111-4111-8111-111111111111');
+    expect(r.supersedes_id).toBe(1);
+    expect(r.revision_ordinal).toBe(1);
+    expect(r.lineage_state).toBe('LIVE');
+  });
+
+  test('B partial rows ONLY behaves as NO VALID LINEAGE HISTORY', () => {
+    const r = resolve([partialRow()]);
+    expect(r.action).toBe('ORIGIN');
+    expect(r.revision_ordinal).toBe(0);
+    expect(r.supersedes_id).toBeNull();
+    // LIVE, not LEGACY_UNVERIFIED: no lineage action ever existed here, so
+    // claiming unverified history would be a false statement about the past.
+    expect(r.lineage_state).toBe('LIVE');
+  });
+
+  test('C valid ORIGIN only, and D no history, are unchanged', () => {
+    expect(resolve([validRow()]).action).toBe('REVISION');
+    expect(resolve([]).action).toBe('ORIGIN');
+    expect(resolve([]).lineage_state).toBe('LIVE');
+  });
+
+  test('THE PRE-REPAIR DEFECTS, reproduced on an unfiltered family', () => {
+    // Fed the raw rows the old lookup returned, the resolver produced a
+    // REVISION with a NULL read_id (an orphaned chain node) and mislabelled a
+    // brand-new Read as LEGACY_UNVERIFIED. Both are reachable by construction.
+    const withClaim = (r) => ({ ...r, claim: { canonical_event_id: r.canonical_event_id, line: r.line, side: r.side, p_win: r.p_win, grade: r.grade } });
+    const orphan = R.resolveLineage({ candidate: CAND, existing: [withClaim(partialRow()), withClaim(validRow())], mintReadId: () => 'MINT' });
+    expect(orphan.action).toBe('REVISION');
+    expect(orphan.read_id).toBeNull();
+    const mislabelled = R.resolveLineage({ candidate: CAND, existing: [withClaim(partialRow())], mintReadId: () => 'MINT' });
+    expect(mislabelled.lineage_state).toBe('LEGACY_UNVERIFIED');
+  });
+});
+
+describe('STEP 12/13 — the lookup scope is lossless and bounded by one slate', () => {
+  test('sport and game_date are COMPONENTS OF THE KEY, so the bound cannot lose a family member', () => {
+    // Derived from the key builder itself, not from a parallel assumption.
+    const k = R.readNaturalKey(CAND);
+    const [sport, gameDate] = k.split('|');
+    expect(sport).toBe(CAND.sport);
+    expect(gameDate).toBe(CAND.game_date);
+    expect(familyScopesFrom([k])).toEqual([{ sport: 'mlb', game_date: '2026-08-28' }]);
+  });
+
+  test('one scope per distinct (sport, date) however many keys are requested', () => {
+    const keys = [];
+    for (let i = 0; i < 5000; i += 1) {
+      keys.push(R.readNaturalKey({ ...CAND, player_key: `p${i}` }));
+    }
+    // 5,000 keys -> ONE index-backed range, not 50 chunked IN-lists.
+    expect(familyScopesFrom(keys)).toHaveLength(1);
+  });
+
+  test('a batch spanning two dates produces exactly two scopes', () => {
+    const a = R.readNaturalKey(CAND);
+    const b = R.readNaturalKey({ ...CAND, game_date: '2026-08-29' });
+    expect(familyScopesFrom([a, b])).toHaveLength(2);
+  });
+
+  test('a malformed key contributes no scope rather than a wildcard one', () => {
+    expect(familyScopesFrom(['', 'mlb', null, undefined])).toEqual([]);
+  });
+});
+
+describe('STEP 15 — failure atomicity', () => {
+  test('a lookup that throws leaves NO lineage-specific state behind', async () => {
+    const rows = [{ ...CAND, snapshot_id: 's' }];
+    const out = await ret.attachLineage(rows, {
+      fetchExisting: async () => { throw new Error('57014 canceling statement due to statement timeout'); },
+    });
+    expect(out.error).toMatch(/57014/);
+    // The row still persists -- retention is not lost -- but it carries nothing
+    // that could later be mistaken for history.
+    expect(rows[0].read_natural_key).toBeNull();
+    expect(rows[0].read_id).toBeNull();
+    expect(rows[0].lineage_action).toBeNull();
+    expect(isValidLineageAction(rows[0])).toBe(false);
+  });
+
+  test('publication provenance is NOT erased by a lineage failure', async () => {
+    const rows = [{ ...CAND, snapshot_id: 's', publication_id: 'pub-1', published_at: 'T' }];
+    await ret.attachLineage(rows, { fetchExisting: async () => { throw new Error('boom'); } });
+    // The slate really was published. Deleting that true fact to tidy up a
+    // false one would be the wrong repair.
+    expect(rows[0].publication_id).toBe('pub-1');
+    expect(rows[0].published_at).toBe('T');
+  });
+
+  test('an unpublished capture is cleared too — it was never an action', async () => {
+    const rows = [{ ...CAND, published: false, snapshot_id: 's' }];
+    const out = await ret.attachLineage(rows, { fetchExisting: async () => [] });
+    expect(out.not_published).toBe(1);
+    expect(rows[0].read_natural_key).toBeNull();
+  });
+
+  test('a successful resolution keeps every field', async () => {
+    const rows = [{ ...CAND, snapshot_id: 's' }];
+    await ret.attachLineage(rows, { fetchExisting: async () => [], mintReadId: () => 'MINT' });
+    expect(rows[0].read_natural_key).toBe(KEY);
+    expect(rows[0].lineage_action).toBe('ORIGIN');
+    expect(isValidLineageAction(rows[0])).toBe(true);
+  });
+});
+
+describe('STEP 21 — deterministic in-batch, digest and fork semantics', () => {
+  const run = async (rows, existing) => {
+    const out = await ret.attachLineage(rows, {
+      fetchExisting: async () => existing, mintReadId: () => 'MINT-UUID',
+    });
+    return out;
+  };
+
+  test('two claims of one family in a single batch: ORIGIN then RECAPTURE/REVISION', async () => {
+    const a = { ...CAND, snapshot_id: 's' };
+    const b = { ...CAND, snapshot_id: 's' };            // identical claim
+    await run([a, b], []);
+    expect(a.lineage_action).toBe('ORIGIN');
+    expect(b.lineage_action).toBe('RECAPTURE');
+    expect(b.read_id).toBe(a.read_id);                   // no forked mint
+    const c = { ...CAND, snapshot_id: 's' };
+    const d = { ...CAND, snapshot_id: 's', p_win: 0.2, grade: 'F' }; // changed claim
+    await run([c, d], []);
+    expect(c.lineage_action).toBe('ORIGIN');
+    expect(d.lineage_action).toBe('REVISION');
+    expect(d.revision_ordinal).toBe(1);
+  });
+
+  test('an unchanged digest against an existing valid head is a RECAPTURE at the same ordinal', async () => {
+    const row = { ...CAND, snapshot_id: 's' };
+    await run([row], [validRow({ claim_digest: R.claimDigest(CAND) })]);
+    expect(row.lineage_action).toBe('RECAPTURE');
+    expect(row.revision_ordinal).toBe(0);
+    expect(row.supersedes_id).toBeNull();
+    expect(row.recaptures_id).toBe(1);
+  });
+
+  test('supersedes uniqueness is a DATABASE guard, and the writer names it', () => {
+    const src = require('fs').readFileSync(
+      require('path').join(__dirname, '../../src/services/retentionService.js'), 'utf8');
+    expect(src).toContain('model_snapshots_supersedes_unique');
+  });
+});
+
+describe('each layer of the defence is pinned INDEPENDENTLY', () => {
+  // Removing either layer alone left the suite green: the other caught it. That
+  // is defence in depth working and test coverage failing, so each layer now
+  // has its own proof.
+
+  test('LAYER 2 — attachLineage excludes an invalid row a fetcher hands it', async () => {
+    const row = { ...CAND, snapshot_id: 's' };
+    const out = await ret.attachLineage([row], {
+      fetchExisting: async () => [partialRow()], mintReadId: () => 'MINT',
+    });
+    expect(out.invalid_rows_excluded).toBe(1);
+    expect(row.lineage_action).toBe('ORIGIN');
+    expect(row.lineage_state).toBe('LIVE');
+  });
+
+  test('LAYER 1 — the default lookup asks the database for valid actions only', async () => {
+    const applied = { eq: [], not: [], select: null, table: null };
+    const rows = [validRow(), partialRow()];
+    const builder = {
+      select(c) { applied.select = c; return this; },
+      eq(col, val) { applied.eq.push([col, val]); return this; },
+      not(col, op, val) { applied.not.push([col, op, val]); return this; },
+      order() { return this; },
+      // The fake honours the filter it was given rather than merely accepting it.
+      range() {
+        const keep = applied.not.some(([c, o]) => c === 'lineage_action' && o === 'is')
+          ? rows.filter((r) => r.lineage_action !== null) : rows;
+        return Promise.resolve({ data: keep, error: null });
+      },
+    };
+    const supabase = { from(t) { applied.table = t; return builder; } };
+    const row = { ...CAND, snapshot_id: 's' };
+    const out = await ret.attachLineage([row], { getClient: () => supabase, mintReadId: () => 'MINT' });
+
+    expect(applied.table).toBe('model_snapshots');
+    expect(applied.not).toContainEqual(['lineage_action', 'is', null]);
+    expect(applied.eq).toContainEqual(['sport', 'mlb']);
+    expect(applied.eq).toContainEqual(['game_date', '2026-08-28']);
+    // The projection must carry every field the predicate needs, or the guard
+    // would judge a row on columns it never asked for.
+    for (const f of VALID_LINEAGE_ACTION_FIELDS) expect(applied.select).toContain(f);
+    expect(out.lookup).toMatchObject({ scopes: 1 });
+    expect(out.error).toBeNull();
+  });
+
+  test('LAYER 1 — the in-loop check rejects an invalid row even if the query returned one', async () => {
+    const rows = [partialRow()];
+    const builder = {
+      select() { return this; }, eq() { return this; }, not() { return this; }, order() { return this; },
+      range() { return Promise.resolve({ data: rows, error: null }); },
+    };
+    const row = { ...CAND, snapshot_id: 's' };
+    const out = await ret.attachLineage([row], { getClient: () => ({ from: () => builder }), mintReadId: () => 'MINT' });
+    expect(out.lookup.invalid_excluded).toBe(1);
+    expect(out.lookup.valid_actions).toBe(0);
+    expect(row.lineage_state).toBe('LIVE');
+  });
+});
+
+describe('the claim digest is frozen', () => {
+  test('a known claim hashes to a known digest', () => {
+    // Pins the digest itself. Dropping or adding a CLAIM field silently changes
+    // every future comparison, and nothing else in the suite would notice.
+    const claim = {
+      line: 0.5, side: 'over', book: 'draftkings', over_odds: -115, under_odds: 100,
+      p_win: 0.9, grade: 'B+', confidence: 90, confidence_basis: 'p_win',
+      ev_pct: 1.5, projection: 1.2, edge_pct: 3, takeable: true, value: false,
+      fair_odds: -122, fair_prob: 0.55, refused: false, refusal_reason: null,
+    };
+    expect(R.claimDigest(claim)).toBe(R.claimDigest({ ...claim }));
+    expect(R.CLAIM_MARKET_FIELDS).toEqual(['line', 'side', 'book', 'locked_odds', 'over_odds', 'under_odds']);
+    // Every market term must move the digest: a field that cannot change it is
+    // a field the chronology is blind to.
+    for (const f of R.CLAIM_MARKET_FIELDS) {
+      const bumped = { ...claim, [f]: f === 'side' ? 'under' : 999 };
+      expect(R.claimDigest(bumped)).not.toBe(R.claimDigest(claim));
+    }
+    expect(R.CLAIM_SCHEMA_VERSION).toBe('claim@1');
+    expect(R.DIGEST_ALGORITHM_VERSION).toBe('sha256-json-sorted@1');
+  });
+});