vyndr/src/services/snapshotService.js

/**
 * snapshotService — scheduled grade pipeline (Session 45).
 *
 * Orchestrates ONE snapshot cycle for one sport. The on-demand "Read" model is
 * retired: a snapshot pre-grades the full slate, LOCKS each grade to the line at
 * snapshot time (`gradedAt`), classifies each player's archetype, computes line
 * deltas vs the previous snapshot, and emits ticker events.
 *
 * Everything is orchestration of EXISTING services (oddsService,
 * gradeSlateService, archetypeService, playerIntelService). All I/O is injectable
 * so the whole cycle is unit-testable with zero network.
 *
 * Redis keys written:
 *   snapshot:{sport}:latest    — current locked snapshot { sport, updated_at, grades, deltas }
 *   snapshot:{sport}:previous  — prior snapshot (for the next delta computation)
 *   grades:{sport}             — { grades, updated_at, source } (GameCard / Explore / leaders)
 *   ticker:items               — capped array of ticker events (newest first)
 */

// Session 60 (AUTONOMY P0) — 24h, NOT 6h. The overnight cron gap is 11 hours
// (03:00 UTC → 14:00 UTC); with a 6h TTL the morning settle pass read an
// EXPIRED snapshot and the accuracy loop silently settled nothing. The
// snapshot must survive until the morning-after settle reads it.
const SNAP_TTL = 24 * 3600;
const TICKER_TTL = 24 * 3600;
const TICKER_CAP = 50;
const DELTA_NOISE = 0.5;        // ignore movements smaller than this
const DELTA_MOVE = 1.0;         // ticker MOVE threshold
const STATS_CONCURRENCY = 5;

const { nameKey, normalizeName } = require('../utils/playerName');
// Session 46 — group/dedupe by the normalized name key so "A.J. Ewing" and
// "AJ Ewing" (or "Jazz Chisholm" / "Jazz Chisholm Jr.") collapse to one player.
const norm = (s) => nameKey(s);
const lastName = (full) => {
  const parts = String(full || '').trim().split(/\s+/);
  return parts.length > 1 ? parts[parts.length - 1] : (parts[0] || '');
};
const sideChar = (dir) => (String(dir || 'over').toLowerCase() === 'under' ? 'U' : 'O');
const propKey = (g) => `${norm(g.player || g.player_name)}|${String(g.stat_type || g.stat || '').toLowerCase()}|${String(g.direction || '').toLowerCase()}`;

async function mapLimit(items, concurrency, fn) {
  const out = new Array(items.length);
  let i = 0;
  async function worker() {
    while (i < items.length) {
      const idx = i++;
      out[idx] = await fn(items[idx], idx);
    }
  }
  await Promise.all(Array.from({ length: Math.min(concurrency, items.length || 1) }, worker));
  return out;
}

/** Index original odds props by propKey-ish (player|stat) for odds lookup.
 *  Session 61 — prefer a book row with BOTH sides priced: the first-seen row
 *  sometimes carried only one side (e.g. betmgm SB unders with no juice),
 *  which locked a NULL odds even though another book priced it. Still a
 *  REAL book row, never synthesized. */
/**
 * Index the odds rows the LOCKED PRICE is read from.
 *
 * TAKEABLE-ONLY (2026-08-02). This used to index the FULL props list, which
 * became the display-widened list on 2026-08-01 — so `gradedAt.odds`, the price
 * a grade is locked at, could be a DFS pick'em or exchange price. Measured: the
 * ledger's non-takeable share went 0% -> 47.9% overnight.
 *
 * The lock price must be one a bettor could ACTUALLY have taken, so this gates
 * on TAKEABLE_BOOKS — not MODEL_BOOKS. `pinnacle` is model-eligible and
 * deliberately NOT takeable, so gating on MODEL would re-break this the moment
 * pinnacle's feed recovers.
 *
 * No takeable quote → the key is ABSENT and the price stays null. An honest
 * missing price beats a price from a book you cannot bet.
 */
function indexOdds(props) {
  const { isTakeableBook } = require('../config/bookRoles');
  const map = {};
  const bothSides = (p) => p && p.over_odds != null && p.under_odds != null;
  for (const p of props || []) {
    if (!isTakeableBook(p && p.book)) continue;
    const k = `${norm(p.player)}|${String(p.stat_type || '').toLowerCase()}`;
    if (!map[k] || (!bothSides(map[k]) && bothSides(p))) map[k] = p;
  }
  return map;
}

function gradedAtFor(g, oddsByKey, ts) {
  const k = `${norm(g.player || g.player_name)}|${String(g.stat_type || g.stat || '').toLowerCase()}`;
  const o = oddsByKey[k];
  let odds = null;
  if (o) {
    odds = String(g.direction || '').toLowerCase() === 'under'
      ? (o.under_odds ?? o.under ?? o.odds ?? null)
      : (o.over_odds ?? o.over ?? o.odds ?? null);
  }
  return { line: g.line, odds, timestamp: ts };
}

/**
 * Compare current grades to the previous snapshot's locked lines. A delta is
 * emitted only when |movement| >= DELTA_NOISE. `direction`:
 *   'toward' = market moving in the direction of our graded side (confirming)
 *   'away'   = market moving against it.
 * For an OVER, a rising line confirms (toward); for an UNDER, a falling line.
 */
function computeLineDeltas(current, previous) {
  const prevMap = {};
  for (const p of previous || []) prevMap[propKey(p)] = p;
  // Session 52 — opt-in verification log (SNAPSHOT_DEBUG=1). Confirms the delta
  // pipeline has a previous snapshot to diff against; off by default (hot path).
  if (process.env.SNAPSHOT_DEBUG === '1') {
    console.log(`[deltas] diffing ${(current || []).length} current vs ${(previous || []).length} previous locked lines`);
  }
  const out = [];
  for (const c of current || []) {
    const prev = prevMap[propKey(c)];
    if (!prev) continue;
    const gradedLine = prev.gradedAt ? prev.gradedAt.line : prev.line;
    const currentLine = c.line;
    if (gradedLine == null || currentLine == null) continue;
    const delta = +(Number(currentLine) - Number(gradedLine)).toFixed(2);
    if (Math.abs(delta) < DELTA_NOISE) continue;
    const side = String(c.direction || 'over').toLowerCase();
    const toward = side === 'over' ? delta > 0 : delta < 0;
    out.push({
      player: c.player || c.player_name,
      stat: c.stat_type || c.stat,
      side: sideChar(side),
      gradedLine: Number(gradedLine),
      currentLine: Number(currentLine),
      delta,
      direction: toward ? 'toward' : 'away',
      grade: c.grade,
    });
  }
  return out;
}

const isTopGrade = (g) => g === 'A+' || g === 'A';

/**
 * Build ticker events from a snapshot: a SCAN summary, GRADE events for the top
 * grades, and MOVE events for significant deltas. Newest-relevant first.
 */
function generateTickerEvents(sport, grades, deltas, ts) {
  const events = [];
  events.push({
    // VERB LAW — the verb is READ, never SCAN. One READ event per sport (deduped).
    tag: 'READ', color: 'var(--g-a)', ts, sport,
    text: `${sport.toUpperCase()} slate read · ${grades.length} props graded`,
  });
  for (const g of grades.filter((x) => isTopGrade(x.grade)).slice(0, 6)) {
    const arch = g.archetype ? `${g.archetype} ` : '';
    events.push({
      tag: g.grade, color: g.grade === 'A+' ? 'var(--g-ap)' : 'var(--g-a)', ts,
      text: `${arch}${lastName(g.player || g.player_name)} ${g.stat_type || g.stat} ${sideChar(g.direction)}${g.line} graded ${g.grade}`,
    });
  }
  for (const d of deltas.filter((x) => Math.abs(x.delta) >= DELTA_MOVE).slice(0, 6)) {
    const arrow = d.delta > 0 ? '▲' : '▼';
    const s = d.side === 'U' ? 'u' : 'o';
    events.push({
      tag: 'MOVE', color: 'var(--amber)', ts,
      text: `${lastName(d.player)} ${s}${d.gradedLine} → ${s}${d.currentLine} ${arrow}${d.delta > 0 ? '+' : ''}${d.delta}`,
    });
  }
  return events;
}

// Session 47 — a READ event's sport, from the event field or its text prefix
// (defends ticker items written before the `sport` field existed). Accepts the
// legacy 'SCAN'/'slate scanned' shape too so cached items dedupe cleanly through
// the verb-law rollover (the ticker regenerates as READ on the next snapshot).
function readSportOf(e) {
  if (e.tag !== 'READ' && e.tag !== 'SCAN') return null;
  if (e.sport) return String(e.sport).toLowerCase();
  const m = String(e.text || '').match(/^([a-z]+)\s+slate (?:read|scanned)/i);
  return m ? m[1].toLowerCase() : null;
}

async function pushTickerItems(events, deps) {
  if (!events || events.length === 0) return;
  const existing = await deps.cacheGet('ticker:items');
  const arr = Array.isArray(existing) ? existing : [];
  // Keep only the LATEST READ per sport: drop existing READ events for any sport
  // that has a fresh READ in this batch. MOVE/GRADE events are time-specific and
  // preserved.
  const freshReadSports = new Set(events.map(readSportOf).filter(Boolean));
  const pruned = arr.filter((e) => {
    const sp = readSportOf(e);
    return !(sp && freshReadSports.has(sp));
  });
  const merged = [...events, ...pruned].slice(0, TICKER_CAP);
  await deps.cacheSet('ticker:items', merged, TICKER_TTL);
}

// Session 60 (night2/B) — accumulate slate players' game logs into the
// roster blob the streaks/hot-list engines read. Merge by nameKey (newer
// entry wins), cap the blob, 72h TTL (a player off the slate for 3 days
// ages out — honest churn, not a leak).
const ROSTERLOGS_TTL = 72 * 3600;
const ROSTERLOGS_CAP = 300;

async function mergeRosterLogs(sport, entries, deps) {
  if (!entries || entries.length === 0) return;
  try {
    const key = `rosterlogs:${sport}`;
    const existing = await deps.cacheGet(key);
    const byKey = new Map();
    for (const e of Array.isArray(existing) ? existing : []) byKey.set(nameKey(e.name), e);
    for (const e of entries) byKey.set(nameKey(e.name), e); // fresh resolve wins
    const merged = [...byKey.values()].slice(-ROSTERLOGS_CAP);
    await deps.cacheSet(key, merged, ROSTERLOGS_TTL);
  } catch (e) {
    console.warn(`[snapshot] rosterlogs merge failed for ${sport}:`, e.message);
  }
}

const ACTIVE_SPORTS = ['mlb', 'nba', 'wnba', 'soccer'];

/**
 * Run one snapshot cycle for `sport`. Returns a summary; never throws.
 * opts (all injectable): getOdds, gradeAndCacheSlate, resolveStats, classify,
 * cacheGet, cacheSet, now, nowMs.
 */
/**
 * Statcast aggregates for the season, indexed by our player key. One read per
 * snapshot run (~1,350 rows / 5 MB), reused for every grade — the alternative
 * is a per-prop lookup inside a tight grading loop.
 */
async function loadStatcastRows(sport) {
  try {
    const sb = require('../utils/supabase').getSupabaseServiceClient();
    if (!sb) return null;
    const { data, error } = await sb.from('statcast_aggregates')
      .select('*').eq('sport', sport).limit(5000);
    if (error || !data) return null;
    const map = new Map();
    for (const r of data) {
      if (!r.player_key) continue;
      // A two-way player has two rows; the one with the larger sample is the
      // profile his props are about far more often than not.
      const prev = map.get(r.player_key);
      const size = Number(r.sample_pa || r.sample_ip || 0);
      const prevSize = prev ? Number(prev.sample_pa || prev.sample_ip || 0) : -1;
      if (!prev || size > prevSize) map.set(r.player_key, r);
    }
    return map;
  } catch { return null; }
}

/**
 * proj-v1 — opposing-pitcher arsenals by source_id (MLBAM id), classified once
 * per snapshot. Read-only; a failure returns an empty map (proj-v1 then has no
 * matchup contribution, honest-absent). Separate from loadStatcastRows (which
 * keys by player name) because the opposing pitcher is resolved by ID.
 */
async function loadPitcherArsenals(sport) {
  const out = new Map();
  try {
    if (String(sport).toLowerCase() !== 'mlb') return out;
    const sb = require('../utils/supabase').getSupabaseServiceClient();
    if (!sb) return out;
    const matchupRead = require('./projection/matchupRead');
    const { data, error } = await sb.from('statcast_aggregates')
      .select('source_id,pitch_mix').eq('sport', sport).eq('role', 'pitcher').limit(5000);
    if (error || !data) return out;
    for (const r of data) {
      if (r.source_id == null || !r.pitch_mix) continue;
      const arsenal = matchupRead.classifyArsenal(r.pitch_mix);
      if (arsenal) out.set(Number(r.source_id), arsenal);
    }
    return out;
  } catch { return out; }
}

/**
 * Stats whose calibration passed leave-one-date-out and may serve a calibrated
 * number. PROVISIONAL: auto-demoted the first time the held-out interval stops
 * excluding zero or the favourite over-prediction flips sign.
 *
 * hits / rbi / runs are deliberately ABSENT — each fails LODO. See
 * specs/lodo-provisional-calibration.md.
 */
const CALIBRATION_DEPLOYED = Object.freeze(['total_bases']);

async function runSnapshot(sport, opts = {}) {
  const sp = String(sport || '').toLowerCase();
  const deps = {
    getOdds: opts.getOdds || require('./oddsService').getOdds,
    gradeAndCacheSlate: opts.gradeAndCacheSlate || require('./gradeSlateService').gradeAndCacheSlate,
    resolveStats: opts.resolveStats || require('./playerIntelService').resolvePlayerStats,
    classify: opts.classify || require('./archetypeService').classify,
    cacheGet: opts.cacheGet || require('../utils/redis').cacheGet,
    cacheSet: opts.cacheSet || require('../utils/redis').cacheSet,
    now: opts.now || (() => new Date().toISOString()),
    nowMs: opts.nowMs || (() => Date.now()),
    // Session 56 — ops alerting (ntfy) + retry-once on a hard odds failure.
    notify: opts.notify || require('../utils/opsNotify').notify,
    sleep: opts.sleep || ((ms) => new Promise((r) => setTimeout(r, ms))),
    retryDelayMs: opts.retryDelayMs != null ? opts.retryDelayMs : 60_000,
    // Session 58 — Phase 1 truth infrastructure. ledger no-ops without
    // SUPABASE env, so tests / local dev never touch a database.
    ledger: opts.ledger || require('./ledgerService'),
    // Wave 2B — reliable ESPN athlete id + DIRECT headshot href from feeds the
    // pipeline already calls (schedule + summary). Fills the NBA/WNBA espnId gap
    // when the stats-resolve fallback misses. Returns {} for MLB / errors.
    buildEspnIndex: opts.buildEspnIndex || require('./espnAthleteIndex').buildEspnAthleteIndex,
    // Session 63 — the opponent-rank feed. Injectable so tests never hit ESPN;
    // under NODE_ENV=test it defaults to a no-op (the opsNotify precedent) so a
    // suite that doesn't know about this dep can never make a live ESPN call.
    // Session 64 — model-snapshot retention. Injectable; null disables it.
    retention: opts.retention !== undefined ? opts.retention : require('./retentionService'),
    // Book Comparison order (Phase 1) — DISPLAY-ONLY per-book price capture.
    // Fenced: written to its own `bookprices:{sport}` key, read by nothing on
    // the grade path. Injectable; a failure never touches grading.
    captureBookPrices: opts.captureBookPrices || require('./bookPriceStore').captureBookPrices,
    // Lock-line persistence (measurement-only). Persists multi-book lines to the DB at
    // the lock moment so a future audit can run the currently-BLOCKED staleness check.
    // Fenced: writes its own `lock_lines` table, read by nothing on the grade path.
    lockLineCapture: opts.lockLineCapture || require('./lockLineCapture'),
    refreshTeamStats: opts.refreshTeamStats
      || (process.env.NODE_ENV === 'test'
        ? async () => null
        : require('./intelligence/teamStatsCache').refreshTeamStats),
  };
  const start = deps.nowMs();
  const ts = deps.now();

  // Session 56 — retry ONCE on a hard failure (thrown error / null response = a
  // transient PropLine/network blip). A successful-but-empty slate is NOT a
  // failure (off-hours), so it is not retried — that would waste quota + latency.
  let odds;
  try {
    odds = await deps.getOdds(sp);
    if (odds == null) throw new Error('null odds response');
  } catch (e) {
    try {
      await deps.sleep(deps.retryDelayMs);
      odds = await deps.getOdds(sp);
      if (odds == null) throw new Error('null odds response (retry)');
    } catch (e2) {
      await deps.notify(`❌ ${sp.toUpperCase()} snapshot FAILED: ${e2.message}`, { title: 'VYNDR pipeline', priority: 'high', tags: ['x'] });
      return { sport: sp, status: 'error', reason: e2.message, gradeCount: 0 };
    }
  }
  const props = (odds && Array.isArray(odds.props)) ? odds.props : [];
  if (props.length === 0) {
    await deps.notify(`⚠️ ${sp.toUpperCase()} snapshot: 0 props (odds unavailable)`, { title: 'VYNDR pipeline', priority: 'low', tags: ['warning'] });
    return { sport: sp, status: 'skipped', reason: 'no props', gradeCount: 0 };
  }

  // Session 64 (Order 1.5) — BIND EVERY PROP TO ITS REAL GAME before anything
  // downstream dates it. PropLine emits no commence_time, so ledgerService's
  // `dateET(prop.game_time) || dateET(gradedTs)` always fell through to the
  // GRADE timestamp — and a 01:00/03:00 UTC snapshot is the previous ET day,
  // so props for tonight were filed under yesterday. Everything downstream
  // (ledger, retention, settlement) reads prop.game_time, so fixing it here
  // fixes all of them at once.
  try {
    const binder = deps.gameBinder || require('./gameBinder');
    const b = await binder.attachGameTimes(sp, props, { gradedAt: ts });
    console.log(`[snapshot] game binding ${sp}: ${b.bound} bound, ${b.alreadyHad} already had times, ${b.unresolved} UNRESOLVED, ${b.ambiguous} ambiguous(doubleheader)`);
    if (b.unresolved > 0 && b.bound === 0 && b.alreadyHad === 0) {
      await deps.notify(`Game binding produced NOTHING for ${sp.toUpperCase()} — ${b.unresolved} props could not be tied to a scheduled game. Their rows will be skipped rather than mis-dated.`, {
        title: 'VYNDR pipeline', priority: 'high', tags: ['rotating_light'],
      });
    }
  } catch (e) {
    console.warn(`[snapshot] game binding failed for ${sp} (rows without a real game time will be skipped):`, e.message);
  }

  // Book Comparison order (Phase 1) — CAPTURE PER-BOOK PRICES BEFORE DEDUP.
  // `props` still carries one row per book here (dedupeProps runs downstream
  // inside gradeAndCacheSlate). The capture only READS props and writes its own
  // `bookprices:{sport}` key at SNAP_TTL — long enough to outlive the cron gap
  // (the 1h odds cache would blank between runs). Best-effort and structurally
  // fenced: nothing on the grade path reads this key, and the graded slate is
  // byte-identical whether or not this runs (bookPriceStore.test.js locks it).
  try {
    const bookStore = deps.captureBookPrices(props, { now: deps.now });
    await deps.cacheSet(`bookprices:${sp}`, bookStore, SNAP_TTL);
    console.log(`[snapshot] book prices ${sp}: ${bookStore.count} props with book rows captured`);
  } catch (e) {
    console.warn(`[snapshot] book-price capture failed for ${sp} (display-only, grading continues):`, e.message);
  }

  // Session 63 — REFRESH TEAM STATS BEFORE GRADING.
  // `refreshTeamStats` is the ONLY writer of `team_stats:{sport}:{abbr}`, which
  // is the ONLY source of `opp_rank_stat` — and it had zero production callers,
  // so that feature was permanently null and engine1's ±1.0 opponent-defense
  // factor could never fire. It is 24h-cached and rate-limited, so this is one
  // cheap ESPN pass per snapshot. Best-effort: a failure here must never break
  // the snapshot — the features simply stay absent, as before.
  try {
    const summary = await deps.refreshTeamStats(sp);
    if (summary && summary.captured != null) {
      console.log(`[snapshot] team stats refreshed for ${sp}: ${summary.captured} captured, ${summary.errored ?? 0} errored`);
    }
  } catch (e) {
    console.warn(`[snapshot] team stats refresh failed for ${sp} (grading continues):`, e.message);
  }

  // Session 64 — RETENTION (Phase 2, priority zero). Collect one row per graded
  // prop per SIDE — graded AND refused — with the feature vector that produced
  // it, so history compounds from tonight and a future model can be replayed
  // against the exact conditions this one faced. Refusals are included on
  // purpose: the ledger drops them, so a gate refusing props that would have
  // won is otherwise invisible.
  const retention = deps.retention;
  // Reuse the LEDGER's date + game-id helpers so retention rows share the
  // ledger's natural key exactly — otherwise the settle pass could never join
  // outcomes onto them.
  const ledgerInternals = (deps.ledger && deps.ledger.__internals) || require('./ledgerService').__internals;
  const retentionGameDate = ledgerInternals.dateET(ts) || ledgerInternals.dateET(new Date().toISOString());
  const retentionCtx = {
    snapshotId: retention ? retention.newSnapshotId() : null,
    capturedAt: ts,
    cycleHourUtc: new Date(ts).getUTCHours(),
    sport: sp,
    gameDate: retentionGameDate,
    gameIdFor: (base) => ledgerInternals.gameIdFor(sp, base, retentionGameDate),
  };
  const collector = retention ? retention.createCollector(retentionCtx) : null;

  // Grade the slate via the existing service; capture the envelope instead of
  // letting it write (we re-write an ENRICHED version below).
  let envelope = null;
  await deps.gradeAndCacheSlate(sp, props, {
    // Bisect hook (2026-08-01): lets the internal trigger run a bounded slate
    // without a prod env change, so a cap regression can be isolated by
    // measurement instead of guessed at. Omitted => gradeSlateService's own
    // DEFAULT_LIMIT (the real production value).
    ...(Number(opts.limit) > 0 ? { limit: Number(opts.limit) } : {}),
    source: (odds && odds.provider) || 'odds-api',
    now: deps.now,
    cacheSet: async (_k, v) => { envelope = v; },
    onGraded: collector ? collector.onGraded : undefined,
  });

  // Retention is COLLECTED here (grade time — features must be exactly what the
  // model saw) but PERSISTED after enrichment below, so archetype/team/opponent
  // are filled in. `persistRetention` is called on BOTH exits, including the
  // empty-slate early return: a slate that graded nothing but refused
  // everything is exactly the case worth recording.
  let retentionRows = 0;
  const persistRetention = async (enrichedGrades) => {
    if (!retention || !collector || !collector.rows.length) return;
    try {
      const rows = retention.mergeEnrichment
        ? retention.mergeEnrichment(collector.rows, enrichedGrades || [])
        : collector.rows;
      const r = await retention.persist(rows);
      retentionRows = r.written || 0;
      console.log(`[snapshot] retention ${sp}: ${r.written}/${r.attempted} rows${r.skipped ? ' (skipped — no supabase env)' : ''}${r.error ? ` ERROR: ${r.error}` : ''}`);
    } catch (e) {
      console.warn(`[snapshot] retention write failed for ${sp} (snapshot continues):`, e.message);
    }
  };
  const rawGraded = (envelope && Array.isArray(envelope.grades)) ? envelope.grades : [];
  if (rawGraded.length === 0) {
    await persistRetention([]);   // refusal-only slate is still history
    return { sport: sp, status: 'skipped', reason: 'no grades', gradeCount: 0 };
  }

  // Session 48 — normalize player display names + dedupe variant grades at the
  // SOURCE so every consumer (GameCard, Explore, leaders, profile) gets clean,
  // merged names. PropLine sends "Matt"/"Matthew", "A.J."/"AJ", "(STL)" tags as
  // separate players; collapse to ONE grade per normalized player + stat (keep
  // the highest-confidence; rawGraded is already confidence-desc).
  // Session 54 — also keep the RICHEST display per player (prefer the accented
  // variant: "José" over "Jose", then the longer string) so the prop rows match
  // the accented pitcher line. The GRADE picked is still the highest-confidence.
  const hasAccent = (s) => [...String(s)].some((c) => c.charCodeAt(0) > 127);
  const richerDisplay = (a, b) => {
    if (!b) return a;
    if (hasAccent(a) !== hasAccent(b)) return hasAccent(a) ? a : b;
    return a.length >= b.length ? a : b;
  };
  const dedup = new Map();
  const bestDisplay = new Map();
  for (const g of rawGraded) {
    const disp = normalizeName(g.player || g.player_name).display || g.player || g.player_name || '';
    const pk = nameKey(disp);
    bestDisplay.set(pk, richerDisplay(disp, bestDisplay.get(pk)));
    const k = `${pk}|${String(g.stat_type || g.stat || '').toLowerCase()}`;
    const cur = { ...g, player: disp, player_name: disp };
    const prev = dedup.get(k);
    if (!prev || (Number(g.confidence) || 0) > (Number(prev.confidence) || 0)) dedup.set(k, cur);
  }
  const graded = [...dedup.values()].map((g) => {
    const disp = bestDisplay.get(nameKey(g.player)) || g.player;
    return { ...g, player: disp, player_name: disp };
  });

  // Archetype per unique player (pure math once we have stats). Best-effort —
  // a missing stat line → no badge (not a fallback archetype).
  const oddsByKey = indexOdds(props);
  const players = [...new Set(graded.map((g) => g.player || g.player_name).filter(Boolean))];
  const archByPlayer = {};
  // Session 59 — capture the player's REAL team from the same stats resolve
  // (statsapi/ESPN), never guessed. Feeds the ledger team/opponent columns
  // and the slate join guard (a prop only attaches to its own game).
  const teamByPlayer = {};
  // Wave 2A — the REAL athlete id from the SAME stats resolve, keyed by player.
  // MLB → MLBAM id (mlbstatic headshot CDN); NBA/WNBA → ESPN athlete id
  // (a.espncdn headshot CDN). Stored on the enriched grade so it flows free to
  // grades:{sport} → slate strips → PlayerAvatar. Zero new I/O. Absent → the
  // component falls to a team-colored monogram (never a fabricated face).
  const playerIdByPlayer = {};
  const espnIdByPlayer = {};
  // Wave 1 (trust bug) — the prop's game participants become the resolve's
  // teamHint: it disambiguates namesake collisions (two "James Wood") and, when
  // the resolved player's real team isn't in the prop's game, the resolver drops
  // the team rather than tag a foreign one (the streaks/rosterlogs JOIN
  // INVARIANT, mirroring the S59 slate guard). Keyed by the normalized name.
  const teamHintByPlayer = {};
  for (const p of props || []) {
    const k = norm(p.player);
    if (!k || teamHintByPlayer[k]) continue;
    const hint = [p.home_team, p.away_team].filter(Boolean);
    if (hint.length) teamHintByPlayer[k] = hint;
  }
  // Session 60 (night2/B) — THE STREAKS PRODUCER. The aggregator (streaks +
  // hot lists) starved because its data producers were all external and
  // unarmed (tank01-prefetch via n8n, the offline Python grading flow).
  // The stats resolve above already fetched each slate player's game log —
  // accumulate it into the `rosterlogs:{sport}` blob rosterLogs.loadRosterLogs
  // reads FIRST. Zero extra API calls; the pipeline now feeds its own
  // free layer.
  const logEntries = [];
  await mapLimit(players, STATS_CONCURRENCY, async (player) => {
    try {
      const teamHint = teamHintByPlayer[norm(player)] || null;
      const stats = await deps.resolveStats(player, sp, teamHint ? { teamHint } : {});
      if (stats && stats.found) {
        const c = deps.classify(sp, stats.classifierInput || {});
        archByPlayer[player] = c.primary ? c.primary.name : null;
        if (stats.team) teamByPlayer[player] = stats.team;
        // Wave 2A — capture the resolved athlete id (headshot thread).
        if (stats.playerId != null) playerIdByPlayer[player] = stats.playerId;
        if (stats.espnId != null) espnIdByPlayer[player] = stats.espnId;
        if (Array.isArray(stats.rawLog) && stats.rawLog.length > 0) {
          logEntries.push({
            name: normalizeName(player).display || player,
            playerId: stats.playerId ?? null,
            team: stats.team || null,
            group: stats.group || null,
            seasonRaw: stats.seasonRaw || null,
            games: stats.rawLog,
          });
        }
      }
    } catch { /* graceful — no badge */ }
  });
  await mergeRosterLogs(sp, logEntries, deps);

  // Wave 2B — the RELIABLE espnId/headshot source. The stats-resolve espnId
  // above comes only from espnStatsAdapter (the offline-Python fallback), which
  // is flaky in prod. ESPN's own schedule→summary feeds (already free, already
  // called elsewhere) carry each athlete's id AND often a DIRECT headshot href.
  // Build the index once per snapshot (MLB → {} so its MLBAM path is untouched)
  // and fill any player the primary resolve left without an id. A direct href is
  // preferred — it's the exact URL, so it never 404s on a constructed path.
  let espnIndex = {};
  try {
    espnIndex = (await deps.buildEspnIndex(sp, { cacheGet: deps.cacheGet, cacheSet: deps.cacheSet })) || {};
  } catch { espnIndex = {}; /* graceful — every player falls to a monogram */ }
  const headshotUrlByPlayer = {};
  for (const player of players) {
    const entry = espnIndex[nameKey(player)];
    if (!entry) continue;
    if (espnIdByPlayer[player] == null && entry.espnId != null) espnIdByPlayer[player] = entry.espnId;
    // A direct ESPN href wins over any constructed URL (most reliable; the only
    // honest route for soccer, where we never construct an id-based URL).
    if (entry.headshotHref) headshotUrlByPlayer[player] = entry.headshotHref;
  }

  const enriched = graded.map((g) => {
    const pn = g.player || g.player_name;
    return {
      ...g,
      gradedAt: gradedAtFor(g, oddsByKey, ts),
      archetype: archByPlayer[pn] || null,
      team: teamByPlayer[pn] || g.team || null,
      // Wave 2A — real headshot id (MLBAM for MLB, ESPN for NBA/WNBA), threaded
      // from the stats resolve above. Absent → PlayerAvatar renders a monogram.
      playerId: playerIdByPlayer[pn] ?? g.playerId ?? null,
      espnId: espnIdByPlayer[pn] ?? g.espnId ?? null,
      // Wave 2B — a RESOLVED absolute headshot URL from ESPN (preferred over the
      // constructed (sport,id) URL). Absent → the id/monogram path stands.
      headshotUrl: headshotUrlByPlayer[pn] ?? g.headshotUrl ?? null,
    };
  });

  // LOCK-LINE PERSISTENCE (measurement-only). At THIS moment the grades have locked to
  // their lines (`gradedAt` uses `ts`), so `props` is the multi-book snapshot AS IT
  // EXISTED AT LOCK. Persist each graded prop's per-book lines to `lock_lines`,
  // timestamped `ts`, so a future audit can check whether our locked line was stale-high
  // vs consensus at lock. Built from the in-memory `props` (no Redis re-read → no TTL
  // race). Best-effort + structurally fenced: nothing on the grade path reads lock_lines,
  // and the graded slate is byte-identical whether or not this runs.
  try {
    const gradedKeys = new Set(enriched.map((g) => `${norm(g.player || g.player_name)}|${String(g.stat_type || g.stat || '').toLowerCase()}`));
    const lockRows = deps.lockLineCapture.buildLockRows(sp, props, gradedKeys, { lockedAt: ts });
    const lr = await deps.lockLineCapture.persist(lockRows);
    console.log(`[lock-lines] ${sp}: ${lr.written}/${lr.attempted} multi-book rows persisted at lock${lr.skipped ? ' (skipped — no supabase env)' : ''}${lr.error ? ` ERROR: ${lr.error}` : ''}`);
  } catch (e) {
    console.warn(`[lock-lines] ${sp} persist failed (measurement-only, snapshot continues):`, e.message);
  }

  // Session 64 — retention persists HERE, after enrichment, so archetype/team/
  // opponent are populated. Feature values were captured at grade time and are
  // NOT touched by the merge (mergeEnrichment only fills the three null fields).
  // Session 71 — CHAMPION / CHALLENGER. The challenger is computed here, where
  // the archetype resolve already happened, so grade-time I/O stays at zero.
  // `enriched` (champion p_win) is READ, never written: the serving projection
  // is untouched, and the challenger rides alongside it to the ledger.
  let withChallenger = enriched;
  try {
    const challenger = deps.challenger || require('./challengerProjection');
    const axes = deps.archetypeAxes || require('./archetypeAxes');
    const rowsByKey = await (deps.loadStatcast || loadStatcastRows)(sp);
    if (rowsByKey && rowsByKey.size) {
      const classifyFor = (playerName) => {
        const row = rowsByKey.get(nameKey(playerName || ''));
        return row ? axes.classifyPlayer(row) : null;
      };
      // Session 77 — attach environment (park × weather) + matchup (platoon)
      // per grade. The batter hand rides on the statcast row; the rest is
      // fetched once here. Best-effort: a context failure leaves archetype-only.
      let contextFor = null;
      let ctxInternals = null; // hoisted for proj-v1 (opposing pitcher resolution)
      try {
        const envCtx = deps.environmentContext || require('./environmentContext');
        const ctx = await envCtx.buildContext(sp, {
          origin: process.env.BACKEND_SELF_ORIGIN || 'http://localhost:3000',
        });
        ctxInternals = ctx._internals || null;
        // Enrich each grade with the hitter hand the platoon estimate needs
        // (statcast_aggregates.bats, already loaded above).
        const handOf = (name) => {
          const row = rowsByKey.get(nameKey(name || ''));
          return row ? row.bats : null;
        };
        contextFor = (g) => ctx.contextFor({ ...g, bats: g.bats || handOf(g.player || g.player_name) });
        console.log(`[env] ${sp} — ${ctx.stats.games || 0} games, ${ctx.stats.venues_with_weather || 0} weather, ${ctx.stats.opp_declared || 0} opp-SP declared`);
      } catch (e) { console.warn(`[env] ${sp} context skipped:`, e.message); }

      withChallenger = await challenger.attachChallenger(enriched, classifyFor, contextFor);
      const moved = withChallenger.filter((g) => g.challenger_delta).length;
      const envMoved = withChallenger.filter((g) => g.env_multiplier != null).length;
      const platoonMoved = withChallenger.filter((g) => (g.challenger_adjustments || []).some((a) => a.axis === 'matchup')).length;
      console.log(`[challenger] ${sp} — ${moved}/${withChallenger.length} adjusted (env ${envMoved}, platoon ${platoonMoved})`);

      // Phase A #2 — SECOND challenger: SEASON contact quality (contact-v1),
      // tagged separately from arch-v1 so each is measured independently. Reuses
      // the statcast rows already loaded; champion + arch-v1 fields untouched.
      // Its own try — a second challenger must never break the pipeline either.
      try {
        const contact = deps.contactChallenger || require('./contactChallenger');
        const refs = contact.buildRefs([...rowsByKey.values()]);
        const rowForContact = (name) => rowsByKey.get(nameKey(name || ''));
        withChallenger = await contact.attachContactChallenger(withChallenger, rowForContact, refs);
        const cNudged = withChallenger.filter((g) => g.contact_delta).length;
        const cAbstain = withChallenger.filter((g) => g.p_win_contact == null).length;
        console.log(`[contact-challenger] ${sp} — ${cNudged} nudged, ${cAbstain} abstained / ${withChallenger.length}`);
      } catch (e) {
        console.warn(`[contact-challenger] ${sp} skipped:`, e.message);
      }

      // proj-v1 — ABSOLUTE matchup projection (MLB batting). Own try; flag-gated;
      // reuses arch-v1's already-computed park/weather/platoon on each grade so
      // it adds no duplicate env I/O. NEVER abstains — thin → wide distribution.
      if (String(process.env.PROJ_V1_ENABLED || '1') !== '0') {
        try {
          const projection = deps.projectionChallenger || require('./projectionChallenger');
          const { abbrOf } = require('./environmentContext');
          const arsenalById = await (deps.loadArsenals || loadPitcherArsenals)(sp);
          const oppByTeam = (ctxInternals && ctxInternals.oppPitcherByTeam) || new Map();
          const mlbAdapter = deps.mlbAdapter || require('./adapters/mlbStatsAdapter');
          withChallenger = await projection.attachProjection(withChallenger, {
            batterRowFor: (g) => rowsByKey.get(nameKey(g.player || g.player_name || '')),
            parkFor: (g) => (g.env_park_base != null ? Number(g.env_park_base) : null),
            weatherFor: (g) => (g.env_weather_mod != null ? Number(g.env_weather_mod) : null),
            platoonFor: (g) => {
              const a = (g.challenger_adjustments || []).find((x) => x.axis === 'matchup');
              return a ? a.multiplier : null;
            },
            arsenalFor: (g) => {
              const pid = oppByTeam.get(abbrOf(g.team));
              return pid != null ? (arsenalById.get(Number(pid)) || null) : null;
            },
            gameLogFor: async (g) => {
              if (!g.playerId) return [];
              try { return (await mlbAdapter.getPlayerGameLog(g.playerId)) || []; } catch { return []; }
            },
          });
          const projected = withChallenger.filter((g) => g.proj_point != null).length;
          const projMatchup = withChallenger.filter((g) => (g.proj_factors && g.proj_factors.breakdown || []).some((f) => f.label === 'matchup' && f.present)).length;
          console.log(`[proj-v1] ${sp} — ${projected}/${withChallenger.length} projected (${projMatchup} with matchup)`);
        } catch (e) {
          console.warn(`[proj-v1] ${sp} skipped:`, e.message);
        }
      }
    }
  } catch (e) {
    // The challenger must NEVER break the pipeline it is measured inside.
    console.warn(`[challenger] ${sp} skipped:`, e.message);
  }

  // ── FORWARD CALIBRATION (LODO-gated, per stat) ────────────────────────
  // Fitted on games that are OVER, applied to tonight's props. `p_win` is NOT
  // touched — the counter stays byte-identical and the calibrated value rides
  // beside it, because a calibration map is a correction TO a forecast, not a
  // different forecast.
  //
  // WHICH STATS SERVE IS MEASURED, NOT ASSUMED. The deploy bar is leave-one-
  // date-out stability: refit dropping each settled date in turn, and the
  // improvement must never reverse. That is the right instrument for a monotone
  // shrink-to-observed layer — the factor gate's >=40 date-cluster interval
  // floor was built for a CAUSAL claim and does not bind here.
  //
  // Measured 2026-08-07: total_bases passes at every held-size threshold. hits
  // FAILS (reverses on 2026-07-22 and 2026-07-26), so it is no longer served
  // calibrated even though it was — a stat that cannot survive dropping one day
  // was never calibrated, it was fitted to that day. rbi and runs also fail.
  //
  // `calibrated` is true only inside a band certified out-of-sample, and it is
  // what `chain.chainAcross` requires before it will compound anything. Removing
  // hits here makes hits props unstackable again, which is the honest
  // consequence of the measurement rather than a regression to work around.
  if (sp === 'mlb') {
    for (const stat of CALIBRATION_DEPLOYED) {
      try {
        const calSvc = deps.calibrationService || require('./model/calibrationService');
        const sbc = require('../utils/supabase').getSupabaseServiceClient();
        const calibrator = sbc ? await calSvc.fromLedger(sbc, { sport: 'mlb', stat }) : null;
        if (calibrator) {
          let marked = 0;
          for (const g of enriched) {
            if (String(g.stat_type || g.stat || '').toLowerCase() !== stat) continue;
            const out = calibrator.calibrate(g.p_win);
            g.p_win_calibrated = out.p_calibrated;
            g.calibrated = out.calibrated;
            g.calibration_reason = out.reason;
            g.calibration_status = 'provisional';
            if (out.calibrated) marked += 1;
          }
          console.log(`[calibration] ${sp} ${stat} (PROVISIONAL) — ${marked} stackable; fit n=${calibrator.fit_n} through ${calibrator.fitted_through}`);
        } else {
          console.log(`[calibration] ${sp} ${stat} — no calibrator (thin history); nothing is stackable`);
        }
      } catch (e) {
        console.warn(`[calibration] ${stat} skipped:`, e.message);
      }
    }
  }

  // LINEUP + BASERUNNER CONTEXT — the input RBI and runs have always needed.
  // Best-effort and dated: a context failure must never break a snapshot, and a
  // lineup is a PRE-GAME fact that changes by the hour, so what we knew at grade
  // time must never be overwritten by what turned out to be true.
  if (sp === 'mlb') {
    try {
      const ctx = deps.lineupContext || require('./lineupContextService');
      const sbc = require('../utils/supabase').getSupabaseServiceClient();
      if (sbc) {
        const res = await ctx.refreshContext({ supabase: sbc, sport: 'mlb' });
        console.log(`[lineup-context] ${sp} — lineups ${res.lineups}/${res.lineup_rows ?? 0} (${res.games_with_lineups ?? 0} games), opportunity ${res.opportunity}/${res.opportunity_rows ?? 0}${res.reason ? ` reason: ${res.reason}` : ''}`);
      }
    } catch (e) {
      console.warn('[lineup-context] skipped:', e.message);
    }
  }

  await persistRetention(enriched);

  // Line deltas vs the previous snapshot's locked lines.
  const prev = await deps.cacheGet(`snapshot:${sp}:latest`);
  const deltas = computeLineDeltas(enriched, prev && prev.grades);

  // Lock: previous = old latest, latest = new, grades = enriched.
  if (prev) await deps.cacheSet(`snapshot:${sp}:previous`, prev, SNAP_TTL);
  // `updated_at` = grade-LOCK time (advances only on a full snapshot, 5×/day —
  // grades never change in-game, so this is intentionally stable). `refreshed_at`
  // = the freshness heartbeat: seeded here at lock time, then bumped every
  // intraday refresh (intradayRefreshService). The SYNC badge keys off
  // refreshed_at — measuring the 20-min cadence against the grade-lock field is
  // what produced the "SIGNAL LIVE vs STALE 8h" contradiction.
  const snapshot = { sport: sp, updated_at: ts, refreshed_at: ts, grades: enriched, deltas, gradeCount: enriched.length };
  await deps.cacheSet(`snapshot:${sp}:latest`, snapshot, SNAP_TTL);
  // Session 59 — grades:{sport} must outlive the gap between cron runs (up to
  // 5h) or team rosters / Explore / leaders go dark mid-day. SNAP_TTL (6h),
  // NOT the legacy 2h gradeSlateService TTL — that gap was why /team showed
  // "No active props" for players who were on the slate (audit 2.4).
  await deps.cacheSet(`grades:${sp}`, { grades: enriched, updated_at: ts, source: (odds && odds.provider) || 'odds-api' }, SNAP_TTL);

  // Ticker exhaust.
  const events = generateTickerEvents(sp, enriched, deltas, ts);
  await pushTickerItems(events, deps);

  // Session 58 — Phase 1 truth infrastructure. (a) Upsert the public model
  // record (user_id NULL, idempotent — re-runs never duplicate or overwrite
  // the original lock). (b) Overwrite today's closing_line/odds from the
  // CURRENT feed — the last capture before game start IS the closing line.
  // Best-effort: the ledger must never break the snapshot.
  let ledgerWritten = 0;
  try {
    const rec = await deps.ledger.recordPipelineGrades(sp, withChallenger, props, { now: deps.now });
    ledgerWritten = rec.written || 0;
    await deps.ledger.captureClosing(sp, props);
  } catch (e) {
    console.warn(`[snapshot] ledger write failed for ${sp}:`, e.message);
  }

  // Session 56 — success alert, enriched with the rolling accuracy (if settled).
  let accPart = '';
  try {
    const acc = await deps.cacheGet(`accuracy:${sp}`);
    const pct = acc && acc.overall && acc.overall.pct;
    if (pct != null) accPart = `, ${pct}% accuracy (30d)`;
  } catch { /* accuracy is best-effort in the alert */ }
  await deps.notify(
    `✅ ${sp.toUpperCase()} snapshot: ${enriched.length} props graded, ${deltas.length} deltas${accPart}`,
    { title: 'VYNDR pipeline', tags: ['white_check_mark'] },
  );

  return {
    sport: sp,
    status: 'ok',
    gradeCount: enriched.length,
    ledgerWritten,
    // Session 64 — surfaced so the scheduler can page when retention silently
    // writes nothing. Retention is best-effort by design, which makes a broken
    // write invisible without this.
    retentionRows,
    topGrades: enriched.filter((g) => isTopGrade(g.grade)).slice(0, 5).map((g) => ({
      player: g.player || g.player_name, stat: g.stat_type || g.stat, grade: g.grade, archetype: g.archetype,
    })),
    deltas: deltas.length,
    duration: deps.nowMs() - start,
  };
}

/** Run snapshots for every active sport sequentially (cron entrypoint). */
// Job 1 — `opts.sports` scopes the run to a subset (the scheduler passes the
// hour's per-sport cadence). Absent → every ACTIVE sport (the on-demand
// /api/internal/snapshot/all behaviour is unchanged). runSnapshot ignores the
// extra key.
async function runAllSnapshots(opts = {}) {
  // An EXPLICIT array is honored verbatim (even empty = run nothing); only an
  // ABSENT `sports` key means "every active sport" (on-demand /all).
  const sports = Array.isArray(opts.sports) ? opts.sports : ACTIVE_SPORTS;
  const results = [];
  for (const sp of sports) {
    results.push(await runSnapshot(sp, opts));
  }
  return results;
}

module.exports = {
  runSnapshot,
  runAllSnapshots,
  computeLineDeltas,
  generateTickerEvents,
  pushTickerItems,
  ACTIVE_SPORTS,
  CALIBRATION_DEPLOYED,
  __internals: { propKey, gradedAtFor, indexOdds, lastName, isTopGrade, DELTA_NOISE, DELTA_MOVE, TICKER_CAP },
};