#!/usr/bin/env node
// Phase 2 driver. Resumable: every row is appended to disk as soon as it exists, so a session
// that dies mid-measurement is restarted by running this file again.
import { appendFileSync, existsSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs';
import { createHash, randomBytes } from 'node:crypto';
import { dirname, resolve } from 'node:path';
import { fileURLToPath } from 'node:url';
import { parseBuffer, scanTail } from './parse-head.mjs';

const RUN = process.argv[2] || '2026-09-10-viallo-metadata-survival';
const BUDGET_MS = Number(process.env.BUDGET_MS || 420000);
const STARTED = Date.now();

const UA = 'article-research/1.0 (viallo metadata survival run; contact info@zava-solutions.com)';
const HEAD_WINDOW = 131072;
const FULL_CEILING = 20 * 1024 * 1024;
const TARGET_FILES = 300;
const MAX_PAGES_S = 10;
const PAGE_LIMIT_S = 40;
const SUBSAMPLE_EVERY = 10;

// Anchored to the script, so the driver behaves the same whatever directory it is run from.
const RUN_DIR = resolve(dirname(fileURLToPath(import.meta.url)), '..');
const DIR = resolve(RUN_DIR, 'data');
mkdirSync(DIR, { recursive: true });
const P = {
  frames: `${DIR}/frames.json`,
  files: `${DIR}/files.jsonl`,
  units: `${DIR}/units.jsonl`,
  rejects: `${DIR}/sampling-rejections.jsonl`,
  validation: `${DIR}/validation.json`,
  progress: `${DIR}/progress.json`,
};

// A second driver would re-fetch units the first has already written, because each builds its
// done-set at start. Refuse to start when the lock is held.
const LOCK = `${DIR}/.driver.lock`;
if (process.env.DRIVER_CHILD !== '1') {
  if (existsSync(LOCK)) {
    // A released lock is an empty file, and pid 0 signals the whole process group, so it
    // must be excluded before the liveness probe.
    const held = Number(readFileSync(LOCK, 'utf8').trim());
    let alive = false;
    if (Number.isInteger(held) && held > 1) {
      try { process.kill(held, 0); alive = true; } catch { alive = false; }
    }
    if (alive) {
      console.error(`another driver is running (pid ${held}); refusing to start`);
      process.exit(3);
    }
  }
  writeFileSync(LOCK, String(process.pid));
  process.on('exit', () => { try { if (readFileSync(LOCK, 'utf8').trim() === String(process.pid)) writeFileSync(LOCK, ''); } catch {} });
}

const readJsonl = (p) => existsSync(p)
  ? readFileSync(p, 'utf8').split('\n').filter(Boolean).map((l) => JSON.parse(l))
  : [];
const append = (p, obj) => appendFileSync(p, JSON.stringify(obj) + '\n');
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const sha = (s) => createHash('sha256').update(s).digest('hex');

function frames() {
  if (!existsSync(P.frames)) {
    writeFileSync(P.frames, JSON.stringify({ salt: randomBytes(16).toString('hex') }, null, 2));
  }
  return JSON.parse(readFileSync(P.frames, 'utf8'));
}
function saveFrames(f) { writeFileSync(P.frames, JSON.stringify(f, null, 2)); }
const salted = (s) => sha(frames().salt + s).slice(0, 16);

// The imageinfo response carries no `archived` flag. Superseded versions are the entries after
// the current one and are served from an /archive/ path; the oldest is the last of those.
function archivedOf(imageinfo) {
  const arch = (imageinfo || []).slice(1).filter((x) => x.url && x.url.includes('/archive/'));
  return arch.length ? arch[arch.length - 1].url : null;
}

// One request per second per host, at most one in flight per host.
const hosts = new Map();
function hostFetch(url, opts = {}) {
  const host = new URL(url).host;
  const st = hosts.get(host) || { last: 0, chain: Promise.resolve(), calls: 0 };
  hosts.set(host, st);
  const run = st.chain.then(async () => {
    const wait = st.last + 1000 - Date.now();
    if (wait > 0) await sleep(wait);
    st.last = Date.now();
    st.calls++;
    return rawFetch(url, opts);
  });
  st.chain = run.then(() => {}, () => {});
  return run;
}

async function rawFetch(url, opts) {
  return fetch(url, { ...opts, headers: { 'User-Agent': UA, ...(opts.headers || {}) } });
}

// Retry classes are kept apart: only the second is a fact about the file.
async function fetchWithRetry(url, opts) {
  let throttled = false;
  for (let i = 0; i < 4; i++) {
    let res;
    try {
      res = await hostFetch(url, opts);
    } catch (e) {
      if (i === 0) { await sleep(30000); continue; }
      return { failure: 'fetch_failed', reason: e.name || 'network_error', throttled };
    }
    if (res.status === 429 || res.status === 503) {
      throttled = true;
      if (i === 3) return { failure: 'throttled', reason: String(res.status), throttled };
      const ra = Number(res.headers.get('retry-after'));
      await sleep(Number.isFinite(ra) && ra > 0 ? ra * 1000 : 2000 * (i + 1));
      continue;
    }
    if (res.status === 404 || res.status === 403 || res.status === 410) {
      if (i === 0) { await sleep(30000); continue; }
      return { failure: 'fetch_failed', reason: String(res.status), throttled };
    }
    return { res, throttled };
  }
  return { failure: 'fetch_failed', reason: 'exhausted', throttled };
}

async function readBody(res, cap) {
  const buf = Buffer.from(await res.arrayBuffer());
  return { buf, bytesOnWire: buf.length, capped: cap !== undefined && buf.length > cap };
}

// ---------------------------------------------------------------- stage 1: frames

async function init() {
  const f = frames();
  if (f.w_generator && f.s_version) return;
  if (!f.w_generator) {
    const r = await fetchWithRetry(
      'https://commons.wikimedia.org/w/api.php?action=query&meta=siteinfo&siprop=general&format=json&formatversion=2');
    if (r.res) {
      const j = await r.res.json();
      f.w_generator = j.query.general.generator;
      f.w_captured_at = new Date().toISOString();
    }
  }
  if (!f.s_version) {
    const chain = [];
    // The pre-registration's instance table was probed before this run; re-probing the refused
    // instance here keeps the field a fact of this run rather than a carried-over claim.
    try {
      const rr = await hostFetch('https://mastodon.social/api/v1/timelines/public?only_media=true&limit=1');
      if (rr.status === 422) chain.push({ host: 'mastodon.social', status: 422 });
    } catch { /* unreachable instance is recorded by its absence from the chain */ }
    for (const host of ['mstdn.social', 'fosstodon.org', 'mas.to']) {
      const r = await fetchWithRetry(`https://${host}/api/v1/instance`);
      if (r.res && r.res.ok) {
        const j = await r.res.json();
        f.s_instance = host;
        f.s_version = j.version;
        f.s_captured_at = new Date().toISOString();
        break;
      }
      chain.push({ host, status: r.res ? r.res.status : r.failure });
    }
    f.instances_refused = chain;
  }
  saveFrames(f);
}

// ---------------------------------------------------------------- stage 2: frame W

async function sampleW() {
  const f = frames();
  if (f.w_sampled) return;
  const accepted = [];
  const seen = new Set();
  let pages = 0;
  while (accepted.length < TARGET_FILES && pages < 40) {
    pages++;
    const r = await fetchWithRetry('https://commons.wikimedia.org/w/api.php?action=query&list=random'
      + '&rnnamespace=6&rnlimit=50&format=json&formatversion=2');
    if (!r.res || !r.res.ok) continue;
    const titles = (await r.res.json()).query.random.map((x) => x.title)
      .filter((t, i, a) => a.indexOf(t) === i);
    for (let i = 0; i < titles.length; i += 50) {
      if (accepted.length >= TARGET_FILES) break;
      const batch = titles.slice(i, i + 50);
      const q = new URLSearchParams({
        action: 'query', titles: batch.join('|'), prop: 'imageinfo|coordinates',
        iiprop: 'url|size|mime|sha1|timestamp|extmetadata', iilimit: 'max',
        iiurlwidth: '800', format: 'json', formatversion: '2',
      });
      const ir = await fetchWithRetry(`https://commons.wikimedia.org/w/api.php?${q}`);
      if (!ir.res || !ir.res.ok) continue;
      const j = await ir.res.json();
      for (const page of j.query.pages || []) {
        if (accepted.length >= TARGET_FILES) break;
        if (seen.has(page.pageid)) continue;
        const ii = (page.imageinfo || [])[0];
        if (!ii) { append(P.rejects, { frame: 'W', title: page.title, why: 'no_imageinfo' }); continue; }
        if (!ii.mime || !ii.mime.startsWith('image/')) {
          append(P.rejects, { frame: 'W', title: page.title, why: 'mime_not_image', mime: ii.mime });
          continue;
        }
        seen.add(page.pageid);
        const archived = archivedOf(page.imageinfo);
        accepted.push({
          file_id: `W:${page.pageid}`,
          frame: 'W',
          pageid: page.pageid,
          title: page.title,
          mime: ii.mime,
          api_size: ii.size,
          sha1_prefix: (ii.sha1 || '').slice(0, 8),
          timestamp: ii.timestamp,
          url_original: ii.url,
          url_display: ii.thumburl || null,
          display_reported_width: ii.thumbwidth ?? null,
          page_coordinates_present: Array.isArray(page.coordinates) && page.coordinates.length > 0,
          archived_url: archived.length ? archived[archived.length - 1].url : null,
          accepted_order: accepted.length,
          subsample: accepted.length % SUBSAMPLE_EVERY === 0,
        });
      }
    }
  }
  for (const rec of accepted) append(P.files, rec);
  f.w_sampled = true;
  f.w_accepted = accepted.length;
  f.w_random_pages = pages;
  saveFrames(f);
}

// The width pair needs its own API calls; constructing thumb URLs by hand is not reliable.
async function widenSubsample() {
  const f = frames();
  if (f.w_widened) return;
  const files = readJsonl(P.files).filter((x) => x.frame === 'W' && x.subsample);
  const patch = new Map();
  for (const rec of files) {
    for (const [key, w] of [['url_display_w200', 200], ['url_display_w1200', 1200]]) {
      const q = new URLSearchParams({
        action: 'query', titles: rec.title, prop: 'imageinfo', iiprop: 'url|size',
        iiurlwidth: String(w), format: 'json', formatversion: '2',
      });
      const r = await fetchWithRetry(`https://commons.wikimedia.org/w/api.php?${q}`);
      if (!r.res || !r.res.ok) continue;
      const j = await r.res.json();
      const ii = ((j.query.pages || [])[0] || {}).imageinfo;
      if (ii && ii[0]) patch.set(`${rec.file_id}|${key}`, ii[0].thumburl || null);
    }
  }
  const all = readJsonl(P.files).map((rec) => {
    if (rec.frame !== 'W') return rec;
    return {
      ...rec,
      url_display_w200: patch.get(`${rec.file_id}|url_display_w200`) ?? null,
      url_display_w1200: patch.get(`${rec.file_id}|url_display_w1200`) ?? null,
    };
  });
  writeFileSync(P.files, all.map((x) => JSON.stringify(x)).join('\n') + '\n');
  f.w_widened = true;
  saveFrames(f);
}

// ---------------------------------------------------------------- stage 3: frame S

async function sampleS() {
  const f = frames();
  if (f.s_sampled) return;
  const inst = f.s_instance || 'mstdn.social';
  const accepted = [];
  let maxId = null;
  let pages = 0;
  while (accepted.length < TARGET_FILES && pages < MAX_PAGES_S) {
    pages++;
    let url = `https://${inst}/api/v1/timelines/public?only_media=true&limit=${PAGE_LIMIT_S}`;
    if (maxId) url += `&max_id=${maxId}`;
    const r = await fetchWithRetry(url);
    if (!r.res || !r.res.ok) break;
    const statuses = await r.res.json();
    if (!Array.isArray(statuses) || statuses.length === 0) break;
    maxId = statuses[statuses.length - 1].id;
    for (const st of statuses) {
      for (const att of st.media_attachments || []) {
        if (accepted.length >= TARGET_FILES) break;
        if (att.type !== 'image') {
          append(P.rejects, { frame: 'S', why: 'attachment_type_not_image', type: att.type });
          continue;
        }
        if (!att.url) { append(P.rejects, { frame: 'S', why: 'no_url' }); continue; }
        accepted.push({
          file_id: `S:${salted(att.url)}`,
          frame: 'S',
          account_hash: salted(String(st.account && st.account.id)),
          status_hash: salted(String(st.id)),
          accepted_order: accepted.length,
          subsample: accepted.length % SUBSAMPLE_EVERY === 0,
          attachment_url_hash: salted(att.url),
          url_full: att.url,
          url_thumb: att.preview_url || null,
          url_remote: att.remote_url || null,
          api_size: null,
        });
      }
    }
  }
  for (const rec of accepted) append(P.files, rec);
  f.s_sampled = true;
  f.s_accepted = accepted.length;
  f.s_pages = pages;
  saveFrames(f);
}

async function backfillArchive() {
  const f = frames();
  if (f.w_archive_backfilled) return;
  const files = readJsonl(P.files);
  const w = files.filter((x) => x.frame === 'W');
  const found = new Map();
  for (let i = 0; i < w.length; i += 50) {
    const batch = w.slice(i, i + 50);
    const q = new URLSearchParams({
      action: 'query', titles: batch.map((x) => x.title).join('|'), prop: 'imageinfo',
      iiprop: 'url|size|mime|timestamp', iilimit: 'max', format: 'json', formatversion: '2',
    });
    const r = await fetchWithRetry(`https://commons.wikimedia.org/w/api.php?${q}`);
    if (!r.res || !r.res.ok) continue;
    const j = await r.res.json();
    for (const page of j.query.pages || []) {
      if (page.title) found.set(page.title, archivedOf(page.imageinfo));
    }
  }
  writeFileSync(P.files, files.map((rec) => JSON.stringify(
    rec.frame === 'W' ? { ...rec, archived_url: found.get(rec.title) ?? null } : rec)).join('\n') + '\n');
  f.w_archive_backfilled = true;
  f.w_with_archive = [...found.values()].filter(Boolean).length;
  saveFrames(f);
}

// ---------------------------------------------------------------- stage 4: fetch

function planUnits(rec) {
  const units = [];
  const arm = (name, url) => { if (url) units.push({ arm: name, url }); };
  if (rec.frame === 'W') {
    arm('original', rec.url_original);
    arm('display', rec.url_display);
    if (rec.subsample) {
      arm('archive_oldest', rec.archived_url);
      arm('display_w200', rec.url_display_w200);
      arm('display_w1200', rec.url_display_w1200);
    }
  } else {
    arm('s_full', rec.url_full);
    arm('s_thumb', rec.url_thumb);
    arm('s_remote', rec.url_remote);
  }
  return units;
}

const fullUnit = (rec) => ({ arm: 'full', url: rec.url_original || rec.url_full });

// The query string carries dots of its own, so it has to be stripped before the extension is read.
function extensionOf(u) {
  const path = u.split('?')[0].split('#')[0];
  const dot = path.lastIndexOf('.');
  return dot === -1 ? null : path.slice(dot + 1).toLowerCase();
}

function samePath(a, b) {
  if (!a || !b) return null;
  try { return new URL(a).pathname === new URL(b).pathname; } catch { return null; }
}

function containerOf(px, scan) {
  return px.container === 'tail' ? (scan.tail_container || 'tail') : px.container;
}

function rowFor(rec, arm, kind, probe) {
  const { px = null, scan = null, full = null } = probe;
  const src = full || px || {};
  return {
    unit_key: `${rec.file_id}|${arm}|${kind}`,
    file_id: rec.file_id,
    frame: rec.frame,
    arm,
    kind,
    url: rec.frame === 'W' ? probe.url : undefined,
    url_hash: rec.frame === 'S' ? salted(probe.url) : undefined,
    status: probe.status,
    range_honoured: probe.rangeHonoured,
    bytes_on_wire: probe.bytesOnWire,
    content_range_total: probe.contentRangeTotal,
    throttled: probe.throttled || false,
    fetch_failed: probe.fetchFailed || false,
    failure_reason: probe.failureReason || null,
    display_is_original: arm === 'display' ? samePath(rec.url_original, rec.url_display) : undefined,
    container_original: rec.mime ? rec.mime.split('/')[1] : undefined,
    container_display: arm.startsWith('display') && probe.url
      ? (extensionOf(probe.url)) : undefined,
    display_page_number: arm.startsWith('display') && probe.url && /page(\d+)/.test(probe.url)
      ? Number(probe.url.match(/page(\d+)/)[1]) : (arm.startsWith('display') ? null : undefined),
    has_exif: src.has_exif,
    gps_ifd_present: src.gps_ifd_present,
    gps_coord_tag_present: src.gps_coord_tag_present,
    gps_coord_all_zero: src.gps_coord_all_zero,
    gps_tag_count: src.gps_tag_count,
    make_present: src.make_present,
    make_is_consumer_device: src.make_is_consumer_device,
    model_present: src.model_present,
    serial_tag_present: src.serial_tag_present,
    datetime_original_present: src.datetime_original_present,
    software_present: src.software_present,
    icc_present: src.icc_present,
    icc_desc_is_stock: src.icc_desc_is_stock,
    xmp_present: src.xmp_present,
    mpf_index_present: src.mpf_index_present,
    jpeg_dqt_present: src.jpeg_dqt_present,
    width: src.width,
    height: src.height,
    container: full ? 'full' : containerOf(px || {}, scan || {}),
    parse_error: src.parse_error,
    metadata_region_complete: src.metadata_region_complete,
    tail_exif_present: scan ? scan.tail_exif_present : undefined,
    tail_gps_ifd_present: scan ? scan.tail_gps_ifd_present : undefined,
    tail_mpf_present: scan ? scan.tail_mpf_present : undefined,
    tail_xmp_present: scan ? scan.tail_xmp_present : undefined,
    fetched_at: new Date().toISOString(),
  };
}

async function doUnit(rec, unit, kind) {
  const headers = {};
  if (kind === 'head') headers.Range = `bytes=0-${HEAD_WINDOW - 1}`;
  if (kind === 'tail') headers.Range = `bytes=-${HEAD_WINDOW}`;
  const r = await fetchWithRetry(unit.url, { headers, signal: AbortSignal.timeout(30000) });
  const base = { url: unit.url, throttled: r.throttled };
  if (r.failure) {
    return { ...base, status: null, fetchFailed: true, failureReason: `${r.failure}:${r.reason}`, bytesOnWire: 0 };
  }
  const res = r.res;
  let buf, bytesOnWire;
  try {
    ({ buf, bytesOnWire } = await readBody(res, kind === 'full' ? FULL_CEILING : undefined));
  } catch (e) {
    // The request's abort signal also aborts the body stream, so a slow body surfaces here
    // rather than at the headers.
    return { ...base, status: res.status, fetchFailed: true,
             failureReason: `fetch_failed:${e.name || 'body_error'}`, bytesOnWire: 0 };
  }
  const cr = res.headers.get('content-range');
  const contentRangeTotal = cr && cr.includes('/') ? Number(cr.split('/')[1]) : null;
  const probe = {
    ...base, status: res.status, rangeHonoured: res.status === 206,
    bytesOnWire, contentRangeTotal,
  };
  if (kind === 'head') {
    probe.px = parseBuffer(buf.subarray(0, Math.min(buf.length, HEAD_WINDOW)),
      { truncated: buf.length > HEAD_WINDOW || (contentRangeTotal || 0) > HEAD_WINDOW });
  } else if (kind === 'tail') {
    // A host that ignores Range answers 200 with the whole file. Truncating to the last window
    // keeps the tail arm's input identical to the head arm's, and stops a marker in the middle
    // of the file from being read as a tail finding.
    probe.scan = scanTail(buf.subarray(Math.max(0, buf.length - HEAD_WINDOW)));
  } else {
    probe.full = parseBuffer(buf, { truncated: false });
    probe.px = null;
  }
  return probe;
}

// The tail is warranted when the file is larger than the window, which the head response states
// either in Content-Range or by having filled the window.
const needsTail = (headRow) =>
  (headRow.content_range_total && headRow.content_range_total > HEAD_WINDOW)
  || (headRow.bytes_on_wire || 0) >= HEAD_WINDOW;

async function fetchStage() {
  const files = readJsonl(P.files);
  const unitRows = readJsonl(P.units);
  const done = new Set(unitRows.map((u) => u.unit_key));
  const headByKey = new Map(unitRows.filter((u) => u.kind === 'head').map((u) => [u.unit_key, u]));
  const queue = [];
  for (const rec of files) {
    const units = planUnits(rec);
    for (const u of units) {
      const headKey = `${rec.file_id}|${u.arm}|head`;
      const headRow = headByKey.get(headKey);
      if (!headRow) { queue.push({ rec, unit: u, kind: 'head' }); continue; }
      const tailKey = `${rec.file_id}|${u.arm}|tail`;
      if (!done.has(tailKey) && needsTail(headRow)) queue.push({ rec, unit: u, kind: 'tail' });
    }
    if (rec.subsample) {
      const fullKey = `${rec.file_id}|full|full`;
      if (!done.has(fullKey)) queue.push({ rec, unit: fullUnit(rec), kind: 'full' });
    }
  }

  // Round-robin by host, so one frame's host does not starve the other's while queued behind it.
  const byHost = new Map();
  for (const item of queue) {
    const h = new URL(item.unit.url).host;
    if (!byHost.has(h)) byHost.set(h, []);
    byHost.get(h).push(item);
  }
  const lanes = [...byHost.values()];
  const ordered = [];
  for (let i = 0; ordered.length < queue.length; i++) {
    for (const lane of lanes) if (lane[i]) ordered.push(lane[i]);
  }

  let cursor = 0;
  const worker = async () => {
    while (cursor < ordered.length) {
      if (Date.now() - STARTED > BUDGET_MS) return;
      const item = ordered[cursor++];
      let probe;
      try {
        probe = await doUnit(item.rec, item.unit, item.kind);
      } catch (e) {
        probe = { url: item.unit.url, status: null, fetchFailed: true,
                  failureReason: `driver_failed:${e && e.message ? e.message : 'unknown'}`,
                  bytesOnWire: 0 };
      }
      append(P.units, rowFor(item.rec, item.unit.arm, item.kind, probe));
      // The tail is planned from what the head actually returned, not from a guessed size.
      if (item.kind === 'head'
          && needsTail({ content_range_total: probe.contentRangeTotal, bytes_on_wire: probe.bytesOnWire })) {
        const tailProbe = await doUnit(item.rec, item.unit, 'tail');
        append(P.units, rowFor(item.rec, item.unit.arm, 'tail', tailProbe));
      }
    }
  };
  await Promise.all([worker(), worker(), worker(), worker()]);
}

// ---------------------------------------------------------------- stage 5: validation

function validate() {
  const units = readJsonl(P.units);
  // The full-file arm belongs to the file, not to one of its URL arms, so it is paired with the
  // file's primary arm: the original for frame W, the full-size attachment for frame S.
  const PRIMARY = { W: 'original', S: 's_full' };
  const byKey = new Map();
  for (const u of units) {
    if (u.kind === 'head') byKey.set(`${u.file_id}|${u.arm}`, { head: u });
  }
  const byFile = new Map();
  for (const [k, v] of byKey) {
    const id = k.split('|')[0] + '|' + k.split('|')[1];
    if (v.head.arm === PRIMARY[v.head.frame]) byFile.set(v.head.file_id, v);
  }
  for (const u of units) {
    if (u.kind === 'tail') {
      const e = byKey.get(`${u.file_id}|${u.arm}`);
      if (e) e.tail = u;
    }
    if (u.kind === 'full') {
      const f = byFile.get(u.file_id);
      if (f) f.full = u;
    }
  }
  // The head window can only lose metadata, never invent it. So the direction that carries
  // information is "head did not find it, the wider arm did". A head that found a GPS IFD and a
  // tail that did not is the expected shape of a file whose metadata sits at the front, and
  // counting it as a disagreement would force a lower-bound caveat for no reason. Both counts are
  // reported, and the strict symmetric one is kept so a replicator can recompute it either way.
  let tailPairs = 0, tailSym = 0, tailInf = 0, superseded = 0;
  let fullPairs = 0, fullSym = 0, fullInf = 0;
  const formats = {};
  for (const { head, tail, full } of byKey.values()) {
    if (!head) continue;
    const h = head.gps_ifd_present;
    const t = tail && tail.tail_gps_ifd_present;
    if (t === true || t === false) {
      tailPairs++;
      if (h === true && t === false) tailSym++;
      if (h === false && t === true) { tailSym++; tailInf++; }
      if (h === null && t === true) superseded++;
    }
    const f = full && full.gps_ifd_present;
    if (f === true || f === false) {
      fullPairs++;
      if ((h === true) !== (f === true)) { fullSym++; if (f === true) fullInf++; }
    }
  }
  for (const r of readJsonl(P.files)) {
    const k = r.mime || (r.frame === 'S' ? 'from_media_attachment' : 'unknown');
    formats[k] = (formats[k] || 0) + 1;
  }
  const out = {
    head_rows: units.filter((u) => u.kind === 'head').length,
    tail_rows: units.filter((u) => u.kind === 'tail').length,
    full_rows: units.filter((u) => u.kind === 'full').length,
    tail_pairs: tailPairs,
    tail_disagrees_symmetric: tailSym,
    tail_disagree_rate_symmetric: tailPairs ? tailSym / tailPairs : null,
    tail_disagrees_informative: tailInf,
    tail_disagree_rate: tailPairs ? tailInf / tailPairs : null,
    head_superseded_by_tail: superseded,
    full_pairs: fullPairs,
    full_disagrees_symmetric: fullSym,
    full_disagrees_informative: fullInf,
    full_disagree_rate: fullPairs ? fullInf / fullPairs : null,
    formats,
    lower_bound_required: (tailPairs ? tailInf / tailPairs > 0.05 : false)
      || (fullPairs ? fullInf / fullPairs > 0.05 : false),
    written_at: new Date().toISOString(),
  };
  writeFileSync(P.validation, JSON.stringify(out, null, 2) + '\n');
  return out;
}

// ---------------------------------------------------------------- main

const stages = process.argv.includes('--stage')
  ? [process.argv[process.argv.indexOf('--stage') + 1]] : ['all'];

async function main() {
  if (stages.includes('all') || stages.includes('init')) await init();
  if (stages.includes('all') || stages.includes('sampleW')) await sampleW();
  if (stages.includes('all') || stages.includes('widen')) await widenSubsample();
  if (stages.includes('all') || stages.includes('backfill')) await backfillArchive();
  if (stages.includes('all') || stages.includes('sampleS')) await sampleS();
  if (stages.includes('all') || stages.includes('fetch')) await fetchStage();
  const v = validate();
  const unitRows = readJsonl(P.units);
  const fileRows = readJsonl(P.files);
  const have = new Set(unitRows.map((u) => u.unit_key));
  const outstanding = [];
  for (const r of fileRows) {
    for (const u of planUnits(r)) {
      if (!have.has(`${r.file_id}|${u.arm}|head`)) outstanding.push(`${r.file_id}|${u.arm}`);
    }
  }
  const done = outstanding.length === 0 && unitRows.length > 0;
  writeFileSync(P.progress, JSON.stringify({
    stage: done ? 'fetch-complete' : 'fetch-partial',
    files: fileRows.length,
    units: unitRows.length,
    outstanding: outstanding.length,
    outstanding_sample: outstanding.slice(0, 10),
    elapsed_ms: Date.now() - STARTED,
    validation: v,
  }, null, 2) + '\n');
  console.log(JSON.stringify({
    files: fileRows.length,
    units: unitRows.length,
    outstanding: outstanding.length,
    stage: done ? 'fetch-complete' : 'fetch-partial',
    head: v.head_rows, tail: v.tail_rows, full: v.full_rows,
    tail_disagree_rate: v.tail_disagree_rate, full_disagree_rate: v.full_disagree_rate,
  }));
}

main().catch((e) => { console.error('DRIVER ERROR', e); process.exit(1); });