builtbykev/vyndr
1
0
Fork 0
Code Issues 1 Pull Requests Actions 5 Packages Projects Releases Wiki Activity

MLB canonical event identity, impossible-binding refusal, event-aware dedupe, publication commit

Browse Source 
Release-isolated slice built from 41ba38e. Ships ONLY the event-integrity +
publication + lineage-canary closure; the 90-path development tree stays
undeployed.

- canonical MLB event identity from statsapi gamePk (mlb:gamepk:<pk>), with
  event_identity_source/method/version recorded. The id is canonical; the
  binding is derived and says so.
- IMPOSSIBLE-BINDING REFUSAL. Verified in prod 2026-08-26: Joe Mack (Marlins)
  was bound to Dodgers@Braves and Yandy Diaz (Rays) to Rangers@WhiteSox, both
  from one book in the 01:00/03:01 UTC cycles after their own games began. Root
  cause is source market data, not the binder. A prop whose player's team is not
  an event participant now refuses; unknown team preserves uncertainty.
- event-aware dedupe: books still collapse, events no longer do. An unresolved
  MLB event fractures rather than falling back to the collision-prone
  date+teams key.
- publication commit moved AFTER the authoritative Redis slate write, with
  exact parity-gap identity when the product publishes and the record does not.
- lineage dual-write behind LINEAGE_CANARY_SPORTS, DISABLED for this deploy.

Excluded deliberately: WNBA feed/chain, market ontology, PerformanceDistribution,
calibration certification, truth diagnostics, applyRevision Phase-1, and the
analyzeViaEngine1 confidence-rounding change (a served field).

Suite 380/5,040/0 from this worktree; web tsc exit 0; champion output identical
to production.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CQJeAG8vcDoL5zkiaJyVb8
This commit is contained in:
Kev
2026-08-27 16:27:36 -04:00
parent 41ba38ed4e
commit 352016790a
16 changed files with 4217 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/services/retentionService.js
+436 -1
View File
@@ -88,6 +88,10 @@ function boolOrNull(v) {
*/
function rowsFromSides(base, sides, ctx = {}) {
const rows = [];
// `published` is declared on every row and defaults FALSE. A capture is not a
// publication until the winner signal says so; defaulting true would assert
// that 65% of these rows were shown to a user.
const list = Array.isArray(sides) ? sides : [];
for (const s of list) {
if (!s) continue;
@@ -143,6 +147,17 @@ function rowsFromSides(base, sides, ctx = {}) {
archetype: s.archetype || null,
refused,
// Set TRUE only by the collector's publication signal (see createCollector).
published: false,
// CANONICAL EVENT IDENTITY. Declared on every row; null when the sport
// has no resolver or the event could not be resolved. `game_id` below
// stays as the LEGACY DERIVED label — useful for diagnostics and
// compatibility, never authoritative for identity.
canonical_event_id: (s.canonical_event_id ?? base.canonical_event_id) ?? null,
event_identity_source: (s.event_identity_source ?? base.event_identity_source) ?? null,
event_identity_method: (s.event_identity_method ?? base.event_identity_method) ?? null,
event_identity_version: (s.event_identity_version ?? base.event_identity_version) ?? null,
event_occurrence: (s.event_occurrence ?? base.event_occurrence) ?? null,
refusal_reason: refused
? (s.suppressed_reason || (s.insufficient_data ? 'insufficient_data' : 'no_grade'))
: null,
@@ -171,16 +186,397 @@ function rowsFromSides(base, sides, ctx = {}) {
/** A collector to hand to gradeAndCacheSlate's `onGraded` hook. */
function createCollector(ctx) {
const rows = [];
// Index by (player|stat|line|side) so the publication signal can find the row
// it already collected without re-deriving the winner rule.
const index = new Map();
const keyOf = (r) => [r.player_key, r.stat, r.line, r.side].join('|');
return {
rows,
onGraded(base, sides) {
try {
rows.push(...rowsFromSides(base, sides, ctx));
const made = rowsFromSides(base, sides, ctx);
for (const r of made) index.set(keyOf(r), r);
rows.push(...made);
} catch { /* collection never affects grading */ }
},
/**
* PUBLICATION SIGNAL — fired only for the side that actually reaches the
* slate.
*
* ── WHY THIS EXISTS ──────────────────────────────────────────────────
* `onGraded` fires with BOTH sides, graded AND refused, BEFORE any
* filtering. Only the higher-confidence graded side becomes the served
* Read. MEASURED on mlb 2026-08-26: of 13,012 collected rows, 3,859 were
* refusals and only 4,569 matched a prop that reached the ledger — 8,443
* (64.9%) describe a state the user was never shown.
*
* Nothing on the row said which was which, so treating every capture as a
* published claim would have built "publication history" for model activity
* that was never published. This marks the difference at the one place that
* knows it: the moment the winner is chosen.
*/
onPublished(base, winner) {
try {
if (!winner) return;
const made = rowsFromSides(base, [winner], ctx);
for (const r of made) {
const hit = index.get(keyOf(r));
if (hit) { hit.published = true; hit.published_side = true; }
}
} catch { /* signalling never affects grading */ }
},
};
}
/**
* DUAL-WRITE: attach append-only Read lineage to rows about to be persisted.
*
* ── WHY THIS IS BEST-EFFORT AND WHY IT MUST DECLARE EVERY KEY ────────────
* Retention is already best-effort relative to the ledger; lineage sits one
* layer further out, so a lineage failure must never cost a retention row and
* never reach the grade. Every failure path below leaves the row persistable
* with NULL lineage, which reads as "unknown" — the honest value.
*
* Lineage keys are declared on EVERY row even when unresolved. PostgREST builds
* a bulk insert from the FIRST row's shape, so a key present on only some rows
* is dropped for the whole batch — the same defect that silently killed
* retention for three days when `chain_shadow` was missing from the schema.
*
* NOT AUTHORITATIVE. Nothing reads these columns to serve a product surface;
* this records truth in parallel so it can be compared against the live path
* before any authority changes.
*/
const LINEAGE_KEYS = Object.freeze([
'read_id', 'read_natural_key', 'claim_digest', 'revision_ordinal',
'supersedes_id', 'recaptures_id', 'lineage_action', 'lineage_state',
'lineage_version', 'change_type', 'claim_schema_version',
'digest_algorithm_version',
]);
/** Claim fields the resolver needs from EXISTING rows to classify a change. */
const LINEAGE_FETCH_CLAIM = Object.freeze([
'canonical_event_id',
'line', 'side', 'book', 'over_odds', 'under_odds',
'p_win', 'grade', 'confidence', 'projection', 'edge_pct', 'ev_pct',
'takeable', 'value', 'fair_odds', 'fair_prob', 'confidence_basis',
'refused', 'refusal_reason',
]);
function blankLineage() {
const o = {};
for (const k of LINEAGE_KEYS) o[k] = null;
return o;
}
/**
* ── DUAL-WRITE FAILURE CONTRACT ──────────────────────────────────────────
* Lineage lives on the SAME ROW as the retention record and is attached before
* the single upsert that writes it. That is not a convenience; it is what makes
* two of the four failure quadrants structurally impossible:
*
* legacy OK + lineage OK one insert carries both.
* legacy OK + lineage FAILS the row still persists with NULL lineage. The
* claim is retained; only its position in the
* chronology is unknown, and `out.lineage.error`
* plus the not_published/refused counters make
* the gap measurable rather than silent.
* legacy FAILS + lineage OK IMPOSSIBLE. There is no separate lineage
* write, so a failed retention insert cannot
* leave behind a lineage record claiming a Read
* was published. This is the quadrant that would
* manufacture false publication history, and the
* write ordering removes it rather than guarding
* against it.
* legacy FAILS + lineage n/a nothing is written at all.
*
* Ordering matters and is deliberate: resolve first, then one write. A separate
* lineage table written after the fact would reintroduce the impossible
* quadrant as a real one.
*/
async function attachLineage(rows, deps = {}) {
const out = {
attempted: Array.isArray(rows) ? rows.length : 0,
origins: 0, revisions: 0, recaptures: 0, forks: 0, refused: 0, unresolved: 0,
// A capture that was never the served Read is NOT a lineage failure. It is
// counted separately so the parity gap stays legible: lumping it into
// `refused` would make a healthy slate look broken.
not_published: 0,
change_types: {},
error: null,
};
const list = Array.isArray(rows) ? rows : [];
// Declare the keys first, unconditionally. Even a total failure leaves a
// persistable batch shape.
for (const r of list) Object.assign(r, blankLineage());
if (list.length === 0) return out;
try {
const lineage = deps.lineage || require('./read/readLineage');
const mintReadId = deps.mintReadId || (() => require('crypto').randomUUID());
const keyed = [];
for (const r of list) {
const key = lineage.readNaturalKey(r);
if (!key) { out.unresolved += 1; continue; }
r.read_natural_key = key;
keyed.push(r);
}
if (keyed.length === 0) return out;
// ONE bounded read of the existing families for these natural keys.
const keys = [...new Set(keyed.map((r) => r.read_natural_key))];
const fetchExisting = deps.fetchExisting || (async (naturalKeys) => {
const getClient = deps.getClient || require('../utils/supabase').getSupabaseServiceClient;
const supabase = getClient();
if (!supabase) return null;
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
.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);
}
return found;
});
const existing = await fetchExisting(keys);
if (existing === null) return out; // no database configured — leave NULL
const byKey = new Map();
for (const e of existing) {
// The prior claim rides alongside so `classifyChange` can say WHY the
// published state advanced, not merely that the digest differs.
const claim = {};
for (const f of LINEAGE_FETCH_CLAIM) claim[f] = e[f];
const entry = { ...e, claim };
if (!byKey.has(e.read_natural_key)) byKey.set(e.read_natural_key, []);
byKey.get(e.read_natural_key).push(entry);
}
for (const r of keyed) {
const res = lineage.resolveLineage({
candidate: r,
existing: byKey.get(r.read_natural_key) || [],
mintReadId,
});
if (!res.ok) {
if (res.refused === 'not_published') out.not_published += 1;
else if (res.action === lineage.LINEAGE_ACTION.FORK_DETECTED) out.forks += 1;
else out.refused += 1;
// A refusal leaves NULL lineage on a row that is still persisted. The
// claim is retained; only its position in the chronology is unknown.
continue;
}
r.read_id = res.read_id;
r.claim_digest = res.claim_digest;
r.revision_ordinal = res.revision_ordinal;
r.supersedes_id = res.supersedes_id ?? null;
r.recaptures_id = res.recaptures_id ?? null;
r.lineage_action = res.action;
r.lineage_state = res.lineage_state;
r.lineage_version = res.lineage_version;
r.change_type = res.change_type || null;
r.claim_schema_version = res.claim_schema_version || null;
r.digest_algorithm_version = res.digest_algorithm_version || null;
if (res.change_type) {
out.change_types[res.change_type] = (out.change_types[res.change_type] || 0) + 1;
}
if (res.action === lineage.LINEAGE_ACTION.ORIGIN) out.origins += 1;
else if (res.action === lineage.LINEAGE_ACTION.REVISION) out.revisions += 1;
else if (res.action === lineage.LINEAGE_ACTION.RECAPTURE) out.recaptures += 1;
// Newly minted read_ids must be visible to later rows in the SAME batch,
// or two rows of one Read would each mint an id and fork it immediately.
const fam = byKey.get(r.read_natural_key) || [];
const claim = {};
for (const f of LINEAGE_FETCH_CLAIM) claim[f] = r[f];
fam.push({
id: null, read_id: r.read_id, read_natural_key: r.read_natural_key,
game_id: r.game_id, claim_digest: r.claim_digest,
revision_ordinal: r.revision_ordinal, lineage_action: r.lineage_action,
supersedes_id: r.supersedes_id, captured_at: r.captured_at, claim,
});
byKey.set(r.read_natural_key, fam);
}
} catch (e) {
out.error = e && e.message ? e.message : String(e);
}
return out;
}
/** The unique index that prevents a forked chain, named so the guard is legible. */
const SUPERSEDES_CONSTRAINT = 'model_snapshots_supersedes_unique';
const FORK_RETRY_LIMIT = 2;
function isSupersedesConflict(error) {
if (!error) return false;
const m = `${error.message || ''} ${error.details || ''}`;
return m.includes(SUPERSEDES_CONSTRAINT) || (error && error.code === '23505' && m.includes('supersedes'));
}
/**
* Recover a batch that lost a supersedes race.
*
* Bounded on purpose: two attempts, then an explicit failure. An unbounded loop
* against a writer that keeps winning would spin forever, and the correct answer
* after a couple of misses is to report the gap rather than keep trying.
*
* Sequence per attempt: re-resolve the batch against the CURRENT head, then
* write. Re-resolution is what makes this safe -- if the rival already published
* an identical claim, `attachLineage` returns RECAPTURE and the row lands
* idempotently; if our claim is still materially different it appends after the
* new head instead of trying to supersede a parent that is no longer the head.
*/
async function recoverFromFork(chunk, deps = {}) {
const out = { recovered: false, written: 0, attempts: 0, error: null };
const supabase = deps.supabase;
for (let attempt = 1; attempt <= FORK_RETRY_LIMIT; attempt += 1) {
out.attempts = attempt;
try {
// Clear stale lineage so re-resolution sees the row as a fresh candidate.
for (const r of chunk) Object.assign(r, blankLineage());
// eslint-disable-next-line no-await-in-loop
await attachLineage(chunk, deps);
// eslint-disable-next-line no-await-in-loop
const { error } = await supabase
.from('model_snapshots')
.upsert(chunk, { onConflict: 'snapshot_id,player_key,stat,line,side' });
if (!error) { out.recovered = true; out.written = chunk.length; return out; }
if (!isSupersedesConflict(error)) { out.error = error.message; return out; }
out.error = error.message;
} catch (e) {
out.error = e && e.message ? e.message : String(e);
return out;
}
}
return out;
}
/**
* COMMIT PUBLICATION - the only place a claim becomes "published".
*
* -- WHY THIS IS SEPARATE FROM THE CAPTURE WRITE -------------------------
* The authoritative served slate is ONE atomic Redis write,
* `cacheSet('snapshot:{sport}:latest', snapshot)`. Retention persists BEFORE
* that write (measured: persistRetention at snapshotService:1137, the slate
* write at :1152), and the winner is selected earlier still. So neither capture
* nor winner-selection can mean "published":
*
* winner selected -> intended for the slate
* retention persisted -> the capture is on the record
* REDIS WRITE SUCCEEDS -> the claim was actually made available <-- HERE
*
* Setting the marker any earlier lets lineage assert a publication that a failed
* Redis write never made. That is the one thing lineage must never do.
*
* -- PUBLICATION IS SLATE-LEVEL, BECAUSE THE WRITE IS --------------------
* The whole envelope commits or none of it does, so every served row shares one
* `publication_id` and one `published_at`. Modelling per-row publication would
* claim a granularity the transport does not have.
*
* -- FAILURE DIRECTION IS DELIBERATE ------------------------------------
* If this update fails after a successful Redis write, rows stay
* `published: false` with NULL lineage. The record then UNDERSTATES what was
* served, which is recoverable and observable. The opposite error - claiming a
* publication that did not happen - is not.
*/
async function commitPublication(spec, deps = {}) {
const s = spec || {};
const out = {
publication_id: s.publicationId || null,
published_at: s.publishedAt || null,
candidates: 0, published: 0, skipped: false, error: null, lineage: null,
};
const served = Array.isArray(s.rows) ? s.rows.filter((r) => r && r.published === true) : [];
out.candidates = served.length;
if (served.length === 0) return out;
try {
const getClient = deps.getClient || require('../utils/supabase').getSupabaseServiceClient;
const supabase = getClient();
if (!supabase) { out.skipped = true; return out; }
for (const r of served) {
r.published_at = out.published_at;
r.publication_id = out.publication_id;
}
// Lineage resolves ONLY over confirmed-published rows.
if (deps.lineage !== false) {
out.lineage = await attachLineage(served, { ...deps, getClient });
}
const CHUNK = 250;
for (let i = 0; i < served.length; i += CHUNK) {
const chunk = served.slice(i, i + CHUNK);
// eslint-disable-next-line no-await-in-loop
const { error } = await supabase
.from('model_snapshots')
.upsert(chunk, { onConflict: 'snapshot_id,player_key,stat,line,side' });
if (!error) { out.published += chunk.length; continue; }
// ── CONCURRENCY RECOVERY ────────────────────────────────────────────
// `model_snapshots_supersedes_unique` rejects a second row superseding the
// same parent. That constraint is the authority against a forked history,
// and hitting it means another writer advanced the chain between our
// resolve and our write. The loser must not silently drop a materially new
// published state, and must not loop.
if (!isSupersedesConflict(error)) { out.error = error.message; break; }
// eslint-disable-next-line no-await-in-loop
const rec = await recoverFromFork(chunk, { ...deps, getClient, supabase });
out.conflicts = (out.conflicts || 0) + 1;
if (rec.recovered) {
out.recovered = (out.recovered || 0) + 1;
out.published += rec.written;
} else {
out.recovery_failed = (out.recovery_failed || 0) + 1;
out.error = rec.error || 'fork recovery exhausted';
break;
}
}
} catch (e) {
out.error = e && e.message ? e.message : String(e);
}
// ── EXACT PARITY-GAP IDENTITY ──────────────────────────────────────────
// "lineage failed" is not an answer. When the product published successfully
// and the record did not, the system must be able to say WHICH EXACT SERVED
// CLAIM is missing, without re-deriving it from player/stat/line/date.
//
// Everything needed to recover it deterministically is captured here from the
// in-memory rows that were actually served.
if (out.error || (out.lineage && out.lineage.error) || out.published < out.candidates) {
out.parity_gap = Object.freeze({
sport: served[0] ? served[0].sport : null,
publication_id: out.publication_id,
published_at: out.published_at,
attempted_at: (deps.now || (() => new Date().toISOString()))(),
reason: out.error || (out.lineage && out.lineage.error) || 'partial_commit',
missing_count: out.candidates - out.published,
// The exact rows, by the identity the retention table itself keys on.
missing: Object.freeze(served.slice(out.published).map((r) => Object.freeze({
snapshot_id: r.snapshot_id || null,
player_key: r.player_key || null,
stat: r.stat || null,
line: r.line === undefined ? null : r.line,
side: r.side || null,
canonical_event_id: r.canonical_event_id || null,
read_id: r.read_id || null,
read_natural_key: r.read_natural_key || null,
claim_digest: r.claim_digest || null,
captured_at: r.captured_at || null,
}))),
});
}
return out;
}
/**
* Persist rows. Chunked, upsert-on-conflict-ignore so a retried cycle can never
* duplicate. No-ops (successfully) without Supabase env, so tests and local dev
@@ -193,6 +589,15 @@ async function persist(rows, deps = {}) {
const getClient = deps.getClient || require('../utils/supabase').getSupabaseServiceClient;
const supabase = getClient();
if (!supabase) { out.skipped = true; return out; }
// NOTE: lineage is NOT resolved here any more.
//
// Capture happens BEFORE the authoritative Redis slate write
// (`snapshot:{sport}:latest`), so resolving lineage at this point would
// record a published claim for a slate that may never be served. Publication
// is committed by `commitPublication()` after that write succeeds.
//
// Rows are persisted with `published: false` and NULL lineage, which is the
// truthful state of a capture that has not yet been served.
const CHUNK = 250;
for (let i = 0; i < rows.length; i += CHUNK) {
const chunk = rows.slice(i, i + CHUNK);
@@ -253,6 +658,29 @@ function mergeEnrichment(rows, enrichedGrades) {
});
}
/**
* Attach the SHADOW CHAIN read to collected rows — the (chain_p, counter_p,
* outcome) triple's first two thirds.
*
* Like `mergeEnrichment` this fills ONE field and touches nothing else. It runs
* later than grade time for the same structural reason the archetype does: the
* chain needs the statcast rows and park factors the enrichment pass loaded, and
* pulling that forward into the grader would put per-prop I/O on the serving
* path.
*
* THE SIDE ALIGNMENT IS THE LOAD-BEARING PART. The chain computes P(over the
* line); `p_win` on the row is expressed for the graded SIDE. Storing the raw
* over-probability against an under row's `p_win` would invert every comparison
* made from it afterwards, silently — so `alignToSide` is given the row's own
* side and the row's own counter probability, and both halves of the triple end
* up pointing the same way. `outcome` is written by the ordinary settle pass and
* is already side-aligned, which completes it.
*
* A row the chain could not read is left NULL. It is not a zero probability and
* not an average hitter — the chain refusing to read someone is a fact worth
* keeping, and a fabricated third of a triple would poison the adjudication this
* column exists to enable.
*/
/**
* Attach the SHADOW CHAIN read to collected rows — the (chain_p, counter_p,
* outcome) triple's first two thirds.
@@ -303,6 +731,13 @@ module.exports = {
mergeEnrichment,
mergeChainShadow,
persist,
attachLineage,
commitPublication,
recoverFromFork,
isSupersedesConflict,
FORK_RETRY_LIMIT,
LINEAGE_KEYS,
LINEAGE_FETCH_CLAIM,
newSnapshotId,
__internals: { numOrNull, intOrNull, boolOrNull },
};