#!/usr/bin/env node

/**
 * scan-ats-full.mjs — Reverse ATS discovery scanner. Part of #230.
 *
 * Where scan.mjs scans the companies you track in portals.yml, this script
 * inverts the direction: it walks public directories of companies per ATS
 * (Greenhouse, Lever, Ashby, Workday, iCIMS, BambooHR, Paylocity, join.com,
 * Workable) and surfaces fresh postings that match your portals.yml
 * `title_filter` / `location_filter` — no manual company curation needed.
 *
 * Optional `title_filter_full` in portals.yml overrides `title_filter` for
 * THIS scanner only, so the keywords tuned for scan.mjs's curated company
 * list do not have to double as the filter for every public board. Absent,
 * `title_filter` is used exactly as before.
 *
 * Optional `title_filter_overrides` broadens the title net further still,
 * but scoped to specific companies (matched by slug) rather than the whole
 * sweep — e.g. letting known university/college employers surface general
 * campus-admin postings without loosening the net for everyone else. See
 * scan.mjs's buildTitleFilterOverrides()/buildTitleFilterWithOverrides().
 *
 * Optional `domain_filter` in portals.yml gates whole BOARDS before any title
 * is tested (#3105): a board with no domain-bearing posting is not this user's
 * industry, and is skipped entirely. Absent — the default — every board is
 * swept exactly as before. Not applied to --seeds: a VC portfolio is already a
 * curated corpus, so gating it would be redundant.
 *
 * Company directories come from the public job-board-aggregator dataset
 * (github.com/Feashliaa/job-board-aggregator), except join.com's, which is
 * read from join.com's own company sitemap. Each is cached in data/cache/ for
 * 24h. Workable has no directory: its public XML job feed carries every
 * tenant's postings, so that source reads the feed once per run (about 140 MB
 * compressed), keeps a copy, and on later runs downloads it again only when
 * Workable says it changed (see workableFeedCache below and loadWorkableFeed
 * in providers/workable.mjs).
 *
 * Zero LLM tokens — pure HTTP + JSON, same providers/ modules as scan.mjs.
 * Postings without a usable publish date are skipped: a reverse scan is only
 * useful for fresh postings, and stale results would flood the pipeline.
 *
 * Usage:
 *   node scan-ats-full.mjs                      # scan all ATS directories, last 3 days
 *   node scan-ats-full.mjs --since 7            # postings from the last 7 days
 *   node scan-ats-full.mjs --ats greenhouse,workday  # subset of sources
 *   node scan-ats-full.mjs --history-seeds --ats successfactors # scan history-derived boards
 *   node scan-ats-full.mjs --limit 200          # max companies per ATS (default: all)
 *   node scan-ats-full.mjs --dry-run            # preview without writing files
 *   node scan-ats-full.mjs --liveness           # Playwright-verify matches before writing
 *   node scan-ats-full.mjs --include-blacklisted # audit: let data/blacklist.md matches through, annotated
 *   node scan-ats-full.mjs --verbose            # log per-board fetch failures
 *   node scan-ats-full.mjs --md-out <dir>       # also write a dated markdown digest to <dir>
 *   node scan-ats-full.mjs --resume             # continue an interrupted sweep from its checkpoint
 *   node scan-ats-full.mjs --help               # print this usage block and exit
 */

import { captureWorkplaceEvidence } from './providers/_country.mjs';
import { readFileSync, writeFileSync, existsSync, mkdirSync, statSync, unlinkSync, createReadStream, createWriteStream } from 'fs';
import { finished } from 'node:stream/promises';
import { createGunzip, createGzip } from 'node:zlib';
import { createHash } from 'crypto';
import path from 'path';
import { fileURLToPath } from 'url';
import { loadYaml } from './lib/yaml-document.mjs';
import { renameSyncWithRetry } from './tracker-utils.mjs';

import { makeHttpCtx, fetchJson, fetchText, fetchTextHead, fetchTextStream } from './providers/_http.mjs';
import { withConnectionScope, withAbortScope } from './providers/_connection-scope.mjs';
import { isResolverFailure, dnsPacingStats } from './providers/_dns-cache.mjs';
import greenhouse from './providers/greenhouse.mjs';
import lever from './providers/lever.mjs';
import ashby from './providers/ashby.mjs';
import workday, { WORKDAY_TRUNCATED_REASON } from './providers/workday.mjs';
import icims from './providers/icims.mjs';
import bamboohr from './providers/bamboohr.mjs';
import paylocity, { boardUrl as paylocityBoardUrl } from './providers/paylocity.mjs';
import join, { parseCompanySitemap } from './providers/join.mjs';
import { loadWorkableFeed, WORKABLE_FEED_URL } from './providers/workable.mjs';
import { openProviderIdCache, PROVIDER_ID_CACHE_DIR } from './provider-id-cache.mjs';
import { decodeEntities } from './providers/_html-entities.mjs';
import { buildTitleFilter, buildTitleFilterOverrides, buildTitleFilterWithOverrides, buildLocationFilter, buildContentFilterOrExit, matchedTitleKeywords, loadSeenUrls, scanHistoryPolicy, seenJobKey, appendToPipeline, appendToScanHistory, findBlacklistEntry, loadBlacklist, parseSinceDays, BROWSER_PROCESS_ENV, DATA_ROOT as SCAN_DATA_ROOT, PORTALS_PATH, PIPELINE_PATH, SCAN_HISTORY_PATH } from './scan.mjs';
import { localToday } from './lib/local-today.mjs';
import { printScanSummaryHeader } from './lib/scan-summary-marker.mjs';
import { SEED_SOURCES, toPortalEntry } from './seeds/vc-portfolios.mjs';
import { validateFlags, flagValue, hasFlag } from './lib/cli-flags.mjs';
import { isMainModule } from './lib/is-main-module.mjs';
import { boardKey, loadDeadBoards, recordBoardFailure, recordBoardResult, saveDeadBoards, shouldSkipDeadBoard } from './dead-boards.mjs';
import { getCareerOpsRoot } from './path-resolver.mjs';
import { barePrefixDomainKeywords, buildDomainFilter } from './title-keywords.mjs';
import { KNOWN_ATS_VENDORS } from './ats-vendor.mjs';
import { loadHistoryAtsSeeds } from './history-ats-seeds.mjs';
import { classifyUnknownHostSeeds, makeHeadProbe } from './branded-successfactors.mjs';
import { loadProviders } from './providers/_registry.mjs';
import { browserLaunchOptions } from './lib/browser-launch.mjs';

// ── Config ──────────────────────────────────────────────────────────

// PORTALS_PATH and PIPELINE_PATH are imported from scan.mjs rather than
// re-derived here (#3510). This file appends results through scan.mjs's
// appendToPipeline, so its own bare-relative copy meant it could create an empty
// data/pipeline.md in the cwd and then write the actual matches somewhere else.
// Its portals fallback had the same split: it honored CAREER_OPS_PORTALS but
// otherwise looked in the cwd instead of the data root.
const DATA_ROOT = getCareerOpsRoot();
const PROVIDERS_DIR = path.join(path.dirname(fileURLToPath(import.meta.url)), 'providers');
const CACHE_DIR = path.join(DATA_ROOT, 'data/cache/ats-companies');
const CACHE_TTL_HOURS = 24;
// Tracks `main` deliberately: the dataset's value is freshness (new boards
// appear weekly), so pinning a commit would defeat the purpose. Integrity rests
// on two layers instead: SLUG_RE validates every entry against a safe charset
// before interpolation, and entryOnHost (below) re-parses each finished
// careers_url and drops anything that doesn't resolve to the ATS's own host —
// so a tampered dataset can at worst name boards that don't exist.
const DATASET_BASE = 'https://raw.githubusercontent.com/Feashliaa/job-board-aggregator/main/data';

// Default fan-out. Correct for sources whose boards each live on their OWN
// host — workday is {tenant}.{instance}.myworkdayjobs.com, so 20 in flight is
// 20 different servers and the resolver, not any one vendor, is the limit.
const CONCURRENCY = 20;

// Greenhouse, Lever and Ashby each serve their ENTIRE directory from a single
// hostname (boards-api.greenhouse.io, api.lever.co, api.ashbyhq.com), so the
// default is 20 sustained connections to ONE API across thousands of boards.
// That earns an HTTP throttle, and a throttled sweep loses live boards rather
// than failing loudly: two full sweeps an hour apart saw lever's unreachable
// count go 2,436 -> 4,100 and ashby's 683 -> 1,675, then recover to 2,525 / 684
// after a cooldown with no change to the dataset. The boards were never dead —
// they were refused, and the matches on them were silently missed.
const SINGLE_HOST_CONCURRENCY = 6;
// recruiting.paylocity.com is also one host for every board, but throttles far
// earlier: at SINGLE_HOST_CONCURRENCY a full sweep got HTTP 429 on 9,052 of
// 10,252 boards. The provider retries a 429 with backoff; this keeps the sweep
// from provoking one on nearly every board.
const PAYLOCITY_CONCURRENCY = 2;
// join.com's job list endpoint holds every client to about ten boards a second
// across all tenants and answers faster requests with a rate-limit error,
// which the provider retries with backoff. In sweeps of the same tenants, more
// than two workers was no faster and, from four, lost boards whose retries ran
// out.
const JOIN_CONCURRENCY = 2;
// A refusing resolver fails every lookup in milliseconds, so a sweep that
// keeps going just feeds it (#2229). Stop after this many consecutive
// resolver-level failures — high enough that a handful of unlucky boards
// can't trip it, low enough to stop within seconds of a real outage.
const RESOLVER_FAILURE_LIMIT = 50;

// Crash insurance for multi-hour directory sweeps: progress + matches are
// checkpointed every CHECKPOINT_EVERY companies so --resume can continue a
// dead run (with its ORIGINAL date window) instead of restarting from zero.
// Anchored too (#3510): a sweep resumed from a different directory found no
// checkpoint and silently restarted a multi-hour run from zero — the one failure
// mode --resume exists to prevent.
const CHECKPOINT_PATH = path.join(DATA_ROOT, 'data/cache/ats-full-checkpoint.json');
const CHECKPOINT_EVERY = 500;
const DEAD_BOARDS_PATH = path.join(DATA_ROOT, 'data/dead-boards.tsv');

export function loadCheckpoint(file = CHECKPOINT_PATH) {
  if (!existsSync(file)) return null;
  try {
    const cp = JSON.parse(readFileSync(file, 'utf-8'));
    if (cp?.version !== 1) return null;
    // A version tag alone isn't enough: `current` is consumed unchecked below
    // (`entriesAll.slice(resumeAt)`), so a truncated or hand-edited record with
    // a missing/non-numeric resumeAt would silently rescan from zero and then
    // persist `startAt + resumeAt === NaN` into every later checkpoint — a state
    // no subsequent resume can recover from. Reject the malformed record instead.
    if (cp.current !== null && cp.current !== undefined && !validCheckpointCurrent(cp.current)) return null;
    return cp;
  } catch {
    return null;
  }
}

function validCheckpointCurrent(cur) {
  return typeof cur === 'object'
    && typeof cur.name === 'string'
    && Number.isInteger(cur.resumeAt) && cur.resumeAt >= 0
    && Number.isInteger(cur.datasetLen) && cur.datasetLen >= 0
    && (cur.deferred === undefined
      || (Array.isArray(cur.deferred) && cur.deferred.every((i) => Number.isInteger(i) && i >= 0)));
}

// A checkpoint written under different scan settings must not be resumed —
// silently mixing sources, caps, or undated policy would corrupt the run's
// semantics. --shuffle is never resumable: the sampled order isn't reproducible.
export function checkpointCompatible(cp, opts) {
  return Boolean(cp)
    && !opts.shuffle
    && JSON.stringify(cp.ats) === JSON.stringify(opts.ats)
    && (cp.limit ?? null) === (opts.limit === Infinity ? null : opts.limit)
    && cp.includeUndated === opts.includeUndated;
}

// Flags are not the only settings a resume can mix. The portals.yml filters
// decide which boards and postings the swept part kept, so a filter edited
// between a run and its --resume leaves the checkpointed matches judged by one
// configuration and the rest by another. Each filter is fingerprinted under its
// own key, so the refusal can name what changed. title_filter_full replaces
// title_filter when present, so the pair is recorded under whichever key the
// sweep actually read. Keys are sorted before hashing: reordering a YAML
// mapping changes nothing the sweep does.
export function filterFingerprints(config) {
  const titleKey = config?.title_filter_full ? 'title_filter_full' : 'title_filter';
  const fp = (v) => datasetFingerprint(sortKeys(v ?? null));
  return {
    [titleKey]: fp(resolveTitleFilterConfig(config)),
    title_filter_overrides: fp(config?.title_filter_overrides),
    location_filter: fp(config?.location_filter),
    content_filter: fp(config?.content_filter),
    domain_filter: fp(config?.domain_filter),
  };
}

function sortKeys(v) {
  if (Array.isArray(v)) return v.map(sortKeys);
  if (v && typeof v === 'object') {
    return Object.fromEntries(Object.keys(v).sort().map((k) => [k, sortKeys(v[k])]));
  }
  return v;
}

// A checkpoint written before filters were recorded carries no fingerprints, so
// there is nothing to compare and it resumes as it did before.
export function changedFilterKeys(cp, fingerprints) {
  if (!cp?.filters) return [];
  return Object.keys(fingerprints).filter((k) => cp.filters[k] !== fingerprints[k]);
}

// Never throws: this runs from parallelEach's `finally`, so an escaping error
// (ENOSPC, EACCES, read-only volume) would reject the whole sweep and discard
// every in-memory match from a multi-hour run — the checkpoint killing the work
// it exists to protect. A failed write costs resumability, not the results.
function writeCheckpoint(cp) {
  try {
    mkdirSync(CACHE_DIR, { recursive: true });
    const tmp = `${CHECKPOINT_PATH}.tmp`;
    writeFileSync(tmp, JSON.stringify(cp), 'utf-8');
    renameSyncWithRetry(tmp, CHECKPOINT_PATH); // atomic: a crash mid-write can't corrupt the checkpoint; retries Windows contention
    return true;
  } catch (err) {
    console.error(`\n⚠ checkpoint write failed (${err.message}) — sweep continues, --resume unavailable`);
    return false;
  }
}

// Dead-board memory is useful resumability metadata, but a cache write must
// never discard the matches from a long-running sweep. Keep it best-effort,
// just like the main checkpoint write above.
function saveDeadBoardsBestEffort(rows) {
  try {
    saveDeadBoards(DEAD_BOARDS_PATH, rows);
    return true;
  } catch (err) {
    console.error(`\n⚠ dead-board cache write failed (${err.message}) — sweep continues`);
    return false;
  }
}

/**
 * `ctx` with `provider`'s slice of the id cache as ctx.idCache, or `ctx` as is
 * when there is no store (a caller that does not pass one, tests).
 */
export function withIdCache(ctx, idCaches, provider) {
  return idCaches ? { ...ctx, idCache: idCaches.forProvider(provider?.id) } : ctx;
}

// Same rule for the provider id cache: a failed write costs later runs a
// lookup, never this sweep's matches. It never rejects, so a checkpoint can
// start it without awaiting.
async function saveProviderIdCacheBestEffort(idCaches) {
  try {
    await idCaches.save();
  } catch (err) {
    console.error(`\n⚠ provider id cache write failed (${err.message}) — sweep continues`);
  }
}

// Cheap content fingerprint of a source's company list. --resume relies on the
// dataset order being byte-for-byte reproducible; a same-length regeneration
// with different members would pass a bare length check and silently resume at
// the wrong offset (companies skipped or re-scanned). Hashing the list detects
// that drift so we can fail loudly instead.
export function datasetFingerprint(list) {
  return createHash('sha1').update(JSON.stringify(list)).digest('hex').slice(0, 16);
}

// Dataset entries are external input destined for URL interpolation — reject
// anything outside a conservative slug charset.
const SLUG_RE = /^[A-Za-z0-9._-]+$/;

// ~47% of workday_companies.json is a data-quality defect in the upstream
// job-board-aggregator dataset: the tenant field holds an instance-name
// lookalike (wd1, wd5, wd12, ...) instead of a real company, with the site
// field copied from the corresponding real entry — every one of these hosts
// is unresolvable. Confirmed against the full dataset (2026-09): exactly 15
// distinct values match this pattern, every matching row sits inside one
// contiguous corrupted block, and none corresponds to a real company (#4454).
const WORKDAY_JUNK_TENANT_RE = /^wd\d+$/i;

// SSRF guard / defense in depth: confirm a constructed careers_url actually
// resolves to the expected ATS host before it reaches provider.fetch. Returns
// the synthetic entry, or null if the URL won't parse or the host isn't canonical.
export function entryOnHost(name, careersUrl, isCanonicalHost) {
  let hostname;
  try {
    ({ hostname } = new URL(careersUrl));
  } catch {
    return null;
  }
  return isCanonicalHost(hostname) ? { name, careers_url: careersUrl } : null;
}

// The public iCIMS dataset is not consistent about what an entry is. Most are a
// bare tenant ("acmefreight", served at careers-acmefreight.icims.com), but
// thousands are already the full portal subdomain: "careers-acmefreight",
// "uscareers-acme", "acmecareers-west". Prefixing every entry with "careers-"
// built hosts like careers-careers-acmefreight.icims.com that do not exist, so
// those boards answered 404 and were recorded as dead. Neither reading is safe
// on its own (a bare tenant can contain a hyphen, and some bare tenants are
// served without the prefix), so return the likelier host first and the other
// shape as a fallback that icims.fetch() tries only on a first-page 404.
export function icimsHostCandidates(slug) {
  const s = String(slug ?? '').toLowerCase().replace(/^-+/, '');
  if (!s) return [];
  const asIs = `${s}.icims.com`;
  const prefixed = `careers-${s}.icims.com`;
  if (s.startsWith('careers-')) return [asIs];
  return s.includes('careers') ? [asIs, prefixed] : [prefixed, asIs];
}

// A Paylocity dataset row is a board GUID plus the company name. The dataset
// stores names HTML-escaped ("A &amp; B"), so they are decoded here. An empty
// name is kept empty: the provider then names the company from the board's
// own title.
export function paylocityEntry(row) {
  const careersUrl = paylocityBoardUrl(row?.guid);
  if (!careersUrl) return null;
  const name = typeof row.name === 'string' ? decodeEntities(row.name).trim() : '';
  return entryOnHost(name, careersUrl, h => h === 'recruiting.paylocity.com');
}

// The Workable source's boards arrive with their postings from the feed, so
// its "provider" only hands them back. It is not registered in providers/:
// nothing in portals.yml can name it, and scan.mjs never downloads the feed.
const WORKABLE_FEED_PROVIDER = Object.freeze({
  id: 'workable-feed',
  fetch: async (entry) => (Array.isArray(entry?.jobs) ? entry.jobs : []),
});

// Each source: the provider module that does the fetching, plus how to turn a
// dataset entry into a synthetic PortalEntry the provider can detect/fetch.
// A source with `loadDataset` builds its list itself instead of downloading a
// company directory (see loadSourceList).
export const SOURCES = {
  greenhouse: {
    provider: greenhouse,
    // Whole directory behind one host — see SINGLE_HOST_CONCURRENCY.
    concurrency: SINGLE_HOST_CONCURRENCY,
    // One host means one keep-alive pool, so per-board connection scopes would
    // only add a TLS handshake per board (see boardScope below).
    singleHost: true,
    dataset: `${DATASET_BASE}/greenhouse_companies.json`,
    toEntry: (slug) => SLUG_RE.test(String(slug))
      ? entryOnHost(String(slug), `https://job-boards.greenhouse.io/${slug}`, h => h === 'job-boards.greenhouse.io')
      : null,
  },
  lever: {
    provider: lever,
    // Whole directory behind one host — see SINGLE_HOST_CONCURRENCY.
    concurrency: SINGLE_HOST_CONCURRENCY,
    // One host means one keep-alive pool, so per-board connection scopes would
    // only add a TLS handshake per board (see boardScope below).
    singleHost: true,
    dataset: `${DATASET_BASE}/lever_companies.json`,
    toEntry: (slug) => SLUG_RE.test(String(slug))
      ? entryOnHost(String(slug), `https://jobs.lever.co/${slug}`, h => h === 'jobs.lever.co')
      : null,
  },
  ashby: {
    provider: ashby,
    // Whole directory behind one host — see SINGLE_HOST_CONCURRENCY.
    concurrency: SINGLE_HOST_CONCURRENCY,
    // One host means one keep-alive pool, so per-board connection scopes would
    // only add a TLS handshake per board (see boardScope below).
    singleHost: true,
    dataset: `${DATASET_BASE}/ashby_companies.json`,
    toEntry: (slug) => SLUG_RE.test(String(slug))
      ? entryOnHost(String(slug), `https://jobs.ashbyhq.com/${slug}`, h => h === 'jobs.ashbyhq.com')
      : null,
  },
  workday: {
    provider: workday,
    dataset: `${DATASET_BASE}/workday_companies.json`,
    // Dataset entries are "tenant|instance|site" triples.
    toEntry: (line) => {
      const [tenant, instance, site] = String(line).split('|');
      if (![tenant, instance, site].every(p => p && SLUG_RE.test(p))) return null;
      if (WORKDAY_JUNK_TENANT_RE.test(tenant)) return null;
      return entryOnHost(
        tenant,
        `https://${tenant}.${instance}.myworkdayjobs.com/${site}`,
        h => h === `${tenant}.${instance}.myworkdayjobs.com` && h.endsWith('.myworkdayjobs.com'),
      );
    },
  },
  icims: {
    provider: icims,
    dataset: `${DATASET_BASE}/icims_companies.json`,
    toEntry: (slug) => {
      if (!SLUG_RE.test(String(slug))) return null;
      const hosts = icimsHostCandidates(slug);
      if (hosts.length === 0) return null;
      const [primary, ...fallbacks] = hosts.map((h) => `https://${h}/jobs/search?ss=1&in_iframe=1`);
      const entry = entryOnHost(String(slug), primary, (h) => h === hosts[0]);
      if (entry && fallbacks.length) entry.fallback_urls = fallbacks;
      return entry;
    },
  },
  bamboohr: {
    provider: bamboohr,
    // Per-tenant own host (<slug>.bamboohr.com), like workday/icims — default
    // CONCURRENCY is correct here, not SINGLE_HOST_CONCURRENCY.
    dataset: `${DATASET_BASE}/bamboohr_companies.json`,
    toEntry: (slug) => SLUG_RE.test(String(slug))
      ? entryOnHost(String(slug), `https://${slug}.bamboohr.com/careers`, h => h === `${slug}.bamboohr.com`)
      : null,
  },
  paylocity: {
    provider: paylocity,
    // Whole directory behind one host that throttles early — see PAYLOCITY_CONCURRENCY.
    concurrency: PAYLOCITY_CONCURRENCY,
    // One host means one keep-alive pool, so per-board connection scopes would
    // only add a TLS handshake per board (see boardScope below).
    singleHost: true,
    // Entries are { guid, name, jobs } objects, not slugs. `jobs` is the
    // dataset's count at snapshot time and does not decide whether a board is
    // scanned: it is out of date by the time the sweep runs.
    dataset: `${DATASET_BASE}/paylocity_companies_clean.json`,
    toEntry: (row) => paylocityEntry(row),
  },
  join: {
    provider: join,
    // Whole directory behind one host that rate-limits — see JOIN_CONCURRENCY.
    concurrency: JOIN_CONCURRENCY,
    singleHost: true,
    // job-board-aggregator has no join file; join.com publishes its own
    // company directory, listed under `Sitemap:` in its robots.txt.
    dataset: 'https://join.com/companies/sitemap.xml',
    parseDataset: parseCompanySitemap,
    toEntry: (slug) => SLUG_RE.test(String(slug))
      ? entryOnHost(String(slug), `https://join.com/companies/${slug}`, h => h === 'join.com')
      : null,
  },
  workable: {
    // Workable publishes every tenant's postings in one hourly XML feed, so
    // there is no company directory to walk and nothing to fetch per board:
    // loadDataset reads the feed once (loadWorkableFeed in
    // providers/workable.mjs, through workableFeedCache) and hands back
    // one board per employer with its postings attached, already narrowed to
    // the --since window and the title filter. The rest of the sweep (domain
    // gate, location and content filters, blacklist, dedup) runs per employer
    // as for any other source. Title-filter overrides match the employer name.
    provider: WORKABLE_FEED_PROVIDER,
    singleHost: true,
    // The feed changes every hour, so an interrupted pass cannot resume at an
    // offset into it; it starts again, and seenUrls keeps the matches already
    // saved from being added twice.
    restartOnResume: true,
    dataset: WORKABLE_FEED_URL,
    loadDataset: async ({ cutoff, includeUndated, titleFilter, cacheDir }) => {
      const cache = workableFeedCache(cacheDir ? { cacheDir } : {});
      try {
        const { companies } = await loadWorkableFeed({ cutoffMs: cutoff, includeUndated, titleFilter, fetchStream: cache.fetchStream });
        cache.commit();
        return { list: companies, status: cache.from === 'stale-cache' ? 'stale' : 'ok' };
      } catch (err) {
        cache.discard();
        throw err;
      }
    },
    toEntry: (row) => (row && typeof row.company === 'string' && row.company
      ? { name: row.company, ...(row.website ? { website: row.website } : {}), jobs: Array.isArray(row.jobs) ? row.jobs : [] }
      : null),
  },
};

// ── CLI args ────────────────────────────────────────────────────────

const KNOWN_FLAGS = [
  '--since', '--limit', '--ats', '--seeds', '--history-seeds', '--probe-unknown-hosts', '--dry-run', '--liveness',
  '--verbose', '--md-out', '--json', '--include-undated', '--include-blacklisted',
  '--shuffle', '--resume', '--help', '-h',
];

// Flags that consume the next argv token as a value (space-separated form —
// the `--flag=value` form is self-contained and never needs this).
const VALUE_FLAGS = ['--since', '--limit', '--ats', '--seeds', '--md-out'];

const USAGE = `Usage:
  node scan-ats-full.mjs                      # scan all ATS directories, last 3 days
  node scan-ats-full.mjs --since 7            # postings from the last 7 days
  node scan-ats-full.mjs --ats greenhouse,workday  # subset of sources
  node scan-ats-full.mjs --history-seeds       # also scan boards from local application history
  node scan-ats-full.mjs --history-seeds --ats successfactors # history-derived boards for this ATS
  node scan-ats-full.mjs --history-seeds --probe-unknown-hosts # also detect branded SuccessFactors hosts (one GET per unknown host)
  node scan-ats-full.mjs --limit 200          # max companies per ATS (default: all)
  node scan-ats-full.mjs --dry-run            # preview without writing files
  node scan-ats-full.mjs --liveness           # Playwright-verify matches before writing
  node scan-ats-full.mjs --include-blacklisted # audit: let data/blacklist.md matches through, annotated
  node scan-ats-full.mjs --verbose            # log per-board fetch failures
  node scan-ats-full.mjs --md-out <dir>       # also write a dated markdown digest to <dir>
  node scan-ats-full.mjs --resume             # continue an interrupted sweep from its checkpoint
  node scan-ats-full.mjs --help               # print this usage block and exit`;

export function parseArgs(argv) {
  const args = argv.slice(2);

  // Shared with reply-watch.mjs/dedup-tracker.mjs/scan.mjs via
  // lib/cli-flags.mjs (#2775). This also fixes a latent ordering bug this
  // script had before the pattern was consolidated: the unrecognized-flag
  // check now runs BEFORE --help, so `--help --bogus` still errors instead
  // of exiting 0 having never looked at `--bogus` (the same ordering
  // CodeRabbit flagged on #2745/#2746).
  validateFlags(args, KNOWN_FLAGS, USAGE, { valueFlags: VALUE_FLAGS });

  const valueOf = (flag) => {
    const idx = args.indexOf(flag);
    if (idx !== -1 && args[idx + 1] && !args[idx + 1].startsWith('--')) return args[idx + 1];
    const kv = args.find(a => a.startsWith(flag + '='));
    return kv ? kv.split('=').slice(1).join('=') : null;
  };
  // Validated by the SAME parser scan.mjs uses, so one flag name cannot mean
  // two different things (#2498). `Number(...) || 3` silently swallowed every
  // malformed operand: `--since abc` and `--since 0` became 3 while the user
  // believed they had scanned the window they typed; `--since -5` put the
  // cutoff in the FUTURE so nothing was ever eligible, which reads exactly like
  // "no new postings"; `--since 1e400` became Infinity → an -Infinity cutoff,
  // i.e. no window at all. Only the DEFAULT stays local: absent --since means
  // 3 days here, where scan.mjs means no bound.
  const sinceArg = parseSinceDays(args);
  if (sinceArg.error) {
    console.error(`Error: ${sinceArg.error}`);
    process.exit(1);
  }
  const sinceDays = sinceArg.days ?? 3;
  // The same rule --since follows (#2498): a malformed operand is an error,
  // never a silent default. `Number(...) || Infinity` turned `--limit abc`,
  // `--limit 0`, a typo like `--limit 1O`, and a bare `--limit` into an
  // UNBOUNDED sweep, the slowest and most network-hungry run this script has,
  // while the user believed they had capped it. The pattern alone is not
  // enough: a long enough run of digits passes it and converts to Infinity,
  // the same sweep, so the value must also be a safe integer.
  let limit = Infinity;
  if (hasFlag(args, '--limit')) {
    // Two caps are refused, as two --since windows are: which one applied
    // depended on how each was spelled, not on the order typed.
    const occurrences = args.filter((a) => a === '--limit' || a.startsWith('--limit='));
    if (occurrences.length > 1) {
      console.error(`Error: --limit given ${occurrences.length} times; pass it once`);
      process.exit(1);
    }
    const raw = String(flagValue(args, '--limit') ?? '').trim();
    if (!/^[1-9]\d*$/.test(raw) || !Number.isSafeInteger(Number(raw))) {
      console.error(`Error: --limit expects a positive whole number of companies, got ${raw ? `"${raw}"` : 'no value'}`);
      process.exit(1);
    }
    limit = Number(raw);
  }
  const atsArg = valueOf('--ats');
  // --seeds: optional comma-separated VC portfolio sources (e.g. yc,a16z).
  // When set, the seed companies are fetched and probed via the ATS providers
  // instead of (or in addition to) the regular ATS directory walk.
  const seedsArg = valueOf('--seeds');
  const seeds = seedsArg
    ? seedsArg.split(',').map(s => s.trim().toLowerCase()).filter(Boolean)
    : [];
  const unknownSeeds = seeds.filter(s => !SEED_SOURCES[s]);
  if (unknownSeeds.length) {
    console.error(`Error: unknown seed source(s): ${unknownSeeds.join(', ')}. Valid: ${Object.keys(SEED_SOURCES).join(', ')}`);
    process.exit(1);
  }
  // When --seeds is the only discovery flag (no --ats), default --ats to none
  // so we don't also walk the full ATS directories unintentionally.
  const ats = atsArg
    ? atsArg.split(',').map(s => s.trim().toLowerCase()).filter(Boolean)
    : (seeds.length > 0 ? [] : Object.keys(SOURCES));
  const validAts = new Set([...Object.keys(SOURCES), ...KNOWN_ATS_VENDORS]);
  const unknown = ats.filter(a => !validAts.has(a));
  if (unknown.length) {
    console.error(`Error: unknown ATS source(s): ${unknown.join(', ')}. Valid: ${[...validAts].join(', ')}`);
    process.exit(1);
  }
  const historySeeds = args.includes('--history-seeds');
  const historyOnly = ats.filter(a => !SOURCES[a]);
  if (!historySeeds && historyOnly.length) {
    throw new Error(
      `--ats ${historyOnly.join(',')} ${historyOnly.length === 1 ? 'has' : 'have'} no public directory source; ` +
      're-run with --history-seeds to scan boards derived from local application history.',
    );
  }
  return {
    sinceDays,
    limit,
    ats,
    atsExplicit: Boolean(atsArg),
    historySeeds,
    probeUnknownHosts: args.includes('--probe-unknown-hosts') && historySeeds,
    seeds,
    dryRun: args.includes('--dry-run'),
    liveness: args.includes('--liveness'),
    verbose: args.includes('--verbose'),
    mdOut: valueOf('--md-out'),
    json: args.includes('--json'),
    includeUndated: args.includes('--include-undated'),
    includeBlacklisted: args.includes('--include-blacklisted'),
    shuffle: args.includes('--shuffle'),
    resume: args.includes('--resume'),
  };
}

// ── Company list cache (24h TTL) ────────────────────────────────────

// Returns { list, status } where status is:
//   'ok'    — fresh fetch, or a cache entry still within TTL
//   'stale' — network fetch failed; falling back to an expired cache
//   'empty' — no data at all (no cache, fetch failed/non-array)
// The status lets callers (and --json) distinguish a degraded scan from an empty one.
//
// A source with `parseDataset(text) → array` publishes its directory in some
// other format (join: a sitemap). The text is parsed before it is cached, so
// the cache file always holds the array, like every other source's.
export async function loadCompanyList(name, url, parseDataset, { cacheDir = CACHE_DIR, fetchJsonFn = fetchJson, fetchTextFn = fetchText } = {}) {
  mkdirSync(cacheDir, { recursive: true });
  const cacheFile = path.join(cacheDir, `${name}.json`);
  if (existsSync(cacheFile)) {
    const ageHours = (Date.now() - statSync(cacheFile).mtimeMs) / 3_600_000;
    if (ageHours < CACHE_TTL_HOURS) {
      try { return { list: JSON.parse(readFileSync(cacheFile, 'utf-8')), status: 'ok' }; } catch { /* refetch below */ }
    }
  }
  try {
    const data = typeof parseDataset === 'function'
      ? parseDataset(await fetchTextFn(url, { timeoutMs: 30_000 }))
      : await fetchJsonFn(url, { timeoutMs: 30_000 });
    if (Array.isArray(data)) {
      writeFileSync(cacheFile, JSON.stringify(data), 'utf-8');
      return { list: data, status: 'ok' };
    }
  } catch (err) {
    console.error(`⚠️  ${name}: could not download company list — ${err.message}`);
  }
  // Stale cache beats nothing.
  if (existsSync(cacheFile)) {
    try { return { list: JSON.parse(readFileSync(cacheFile, 'utf-8')), status: 'stale' }; } catch { /* fall through */ }
  }
  return { list: [], status: 'empty' };
}

// A source with `loadDataset` (workable) builds its own list from the sweep's
// window and title filter; every other source downloads its company directory
// through loadCompanyList. Both answer { list, status }, and a failed
// loadDataset reads as 'empty', reported once, like a directory that could not
// be downloaded and has no cache.
export async function loadSourceList(name, source, datasetOpts, { loadCompanyListFn = loadCompanyList } = {}) {
  if (typeof source?.loadDataset !== 'function') return loadCompanyListFn(name, source.dataset, source.parseDataset);
  try {
    const loaded = await source.loadDataset(datasetOpts);
    // A loader answers its list, or { list, status } when it can say more
    // (a list read from an expired cache is 'stale').
    const list = Array.isArray(loaded) ? loaded : loaded?.list;
    const status = !Array.isArray(loaded) && ['ok', 'stale'].includes(loaded?.status) ? loaded.status : 'ok';
    return { list: Array.isArray(list) ? list : [], status };
  } catch (err) {
    console.error(`⚠️  ${name}: could not read ${source.dataset} — ${err.message}`);
    return { list: [], status: 'empty' };
  }
}

// Workable rebuilds its job feed about once an hour and refuses a second
// download inside the hour (HTTP 429, Retry-After up to 3600). It answers a
// conditional request, though: the feed carries Last-Modified, and a GET with
// that date in If-Modified-Since gets 304 with no body and no refusal. So the
// last download is kept, gzipped, beside its Last-Modified, and every run asks
// whether the feed changed: on 304 it reads the copy, on 200 it keeps the new
// download. When the request is refused before any of the feed arrived, the
// copy is read and the source reports 'stale', as an expired directory cache
// does.
const WORKABLE_FEED_CACHE_FILE = 'workable.xml.gz';
const WORKABLE_FEED_META_FILE = 'workable.meta.json';

/**
 * Wait until a full gzip stream can take more. Also ends when the gzip stream
 * or the file it feeds fails or closes: a broken file never drains the gzip
 * stream. Every listener is removed when the wait ends.
 */
export function drainOrFailure(gzip, out) {
  if (gzip.destroyed || out.destroyed) return Promise.resolve();
  return new Promise((resolve) => {
    const events = [[gzip, 'drain'], [gzip, 'error'], [gzip, 'close'], [out, 'error'], [out, 'close']];
    const done = () => {
      for (const [emitter, name] of events) emitter.off(name, done);
      resolve();
    };
    for (const [emitter, name] of events) emitter.on(name, done);
  });
}

/**
 * A fetchTextStream stand-in for loadWorkableFeed that serves and keeps the
 * feed copy. A fresh download is written to a temporary file and becomes the
 * copy only on commit(), after the caller accepted the whole document, so a
 * copy is always a complete feed; discard() drops a download the caller
 * refused.
 *
 * @param {{ cacheDir?: string, fetchStream?: typeof fetchTextStream, now?: () => number }} [opts]
 */
export function workableFeedCache({ cacheDir = CACHE_DIR, fetchStream = fetchTextStream, now = Date.now } = {}) {
  const file = path.join(cacheDir, WORKABLE_FEED_CACHE_FILE);
  const metaFile = path.join(cacheDir, WORKABLE_FEED_META_FILE);
  const tmp = `${file}.tmp-${process.pid}`;
  /** @type {'network' | 'cache' | 'stale-cache'} */
  let from = 'network';
  /** Last-Modified of the download in `tmp`, written beside the copy on commit(). */
  let downloadedLastModified = null;
  const removeQuietly = (p) => { try { unlinkSync(p); } catch { /* not there */ } };
  const copyLastModified = () => {
    if (!existsSync(file)) return null;
    try {
      const value = JSON.parse(readFileSync(metaFile, 'utf-8'))?.lastModified;
      return typeof value === 'string' && value.trim() ? value.trim() : null;
    } catch {
      return null;
    }
  };

  async function readCopy(onText) {
    const decoder = new TextDecoder('utf-8');
    let bytes = 0;
    for await (const chunk of createReadStream(file).pipe(createGunzip())) {
      bytes += chunk.length;
      if (await onText(decoder.decode(chunk, { stream: true })) === false) return { bytes, complete: false };
    }
    const tail = decoder.decode();
    if (tail) await onText(tail);
    return { bytes, complete: true };
  }

  async function stream(url, opts, onText) {
    const validator = copyLastModified();
    const requestOpts = validator
      ? { ...opts, headers: { ...(opts?.headers ?? {}), 'if-modified-since': validator }, acceptNotModified: true }
      : opts;
    // The copy is written only once text arrives, so a refused request never
    // touches the disk. A failure to write it costs the copy, not the read.
    /** @type {import('node:zlib').Gzip | null} */
    let gzip = null;
    /** @type {import('node:fs').WriteStream | null} */
    let out = null;
    let writeFailed = false;
    const sink = () => {
      if (!gzip) {
        mkdirSync(cacheDir, { recursive: true });
        gzip = createGzip({ level: 1 });
        out = createWriteStream(tmp);
        gzip.on('error', () => { writeFailed = true; });
        out.on('error', () => { writeFailed = true; });
        gzip.pipe(out);
      }
      return gzip;
    };
    const closeSink = async (keep) => {
      if (!gzip) return;
      if (keep && !writeFailed) {
        gzip.end();
        await finished(out).catch(() => { writeFailed = true; });
      } else {
        gzip.destroy();
        out.destroy();
        await finished(out).catch(() => {});
      }
      if (!keep || writeFailed) removeQuietly(tmp);
    };
    let delivered = false;
    try {
      const result = await fetchStream(url, requestOpts, async (text) => {
        delivered = true;
        if (!writeFailed) {
          const g = sink();
          if (!g.write(text) && !writeFailed) await drainOrFailure(g, out);
        }
        return onText(text);
      });
      if (result?.notModified) {
        await closeSink(false);
        from = 'cache';
        return readCopy(onText);
      }
      await closeSink(true);
      from = 'network';
      downloadedLastModified = typeof result?.lastModified === 'string' ? result.lastModified : null;
      return result;
    } catch (err) {
      await closeSink(false);
      // Only a refusal that delivered nothing can fall back: once part of the
      // feed reached the caller, the copy would be read on top of it.
      if (!delivered && existsSync(file)) {
        from = 'stale-cache';
        return readCopy(onText);
      }
      if (err?.status === 429) {
        // A missing header is unknown, not 0 seconds.
        const raw = typeof err.retryAfter === 'string' ? err.retryAfter.trim() : '';
        const retryS = raw === '' ? NaN : Number(raw);
        const when = Number.isFinite(retryS) && retryS >= 0 ? `after ${new Date(now() + retryS * 1000).toISOString()}` : 'later';
        throw new Error(`Workable serves its job feed about once an hour and refused this request (HTTP 429); try again ${when}`);
      }
      throw err;
    }
  }

  return {
    fetchStream: stream,
    get from() { return from; },
    /**
     * Keep a fresh download as the copy, with the date the next run asks
     * about. A failure here costs the copy, not the feed already read.
     */
    commit() {
      if (from !== 'network' || !existsSync(tmp)) return;
      try {
        renameSyncWithRetry(tmp, file);
        if (downloadedLastModified) writeFileSync(metaFile, JSON.stringify({ lastModified: downloadedLastModified }), 'utf-8');
        else removeQuietly(metaFile);
      } catch (err) {
        console.error(`⚠️  workable: could not keep the feed copy (${err.message}); the next run downloads the feed again`);
        removeQuietly(tmp);
        removeQuietly(metaFile);
      }
    },
    /** Drop a download the caller refused. */
    discard() {
      removeQuietly(tmp);
    },
  };
}

// Date gate for one posting in a reverse (fresh-first) scan:
//   'stale'   — dated, but older than the cutoff → always dropped
//   'undated' — no usable publish date → dropped by default, kept with --include-undated
//   'keep'    — dated and within the window
export function classifyPostingDate(job, cutoff) {
  if (job.postedAt && job.postedAt < cutoff) return 'stale';
  if (!job.postedAt) return 'undated';
  return 'keep';
}

// #6020. content_filter.rescue_title_misses is not applied here.
// listingMayMatch() below skips a whole board when no title on it matches,
// before any posting body is downloaded; that precheck is what makes a sweep of
// a full public directory affordable, and rescuing title misses would mean
// downloading every board. Say so instead of ignoring the key silently.
export const RESCUE_TITLE_MISSES_IGNORED_WARNING = '⚠️  content_filter.rescue_title_misses is ignored by scan-ats-full.mjs: it skips boards by title before downloading descriptions. Title misses are dropped here as without the key; scan.mjs and scan-dayforce.mjs apply it.';
export function warnRescueTitleMissesIgnored(contentFilter, warn = console.error) {
  if (contentFilter?.rescueTitleMisses === true) warn(RESCUE_TITLE_MISSES_IGNORED_WARNING);
}

// Can this board hold a match? Asked of a provider's cheap listing (titles and
// dates, no posting bodies) before its full fetch. Only the two checks that
// need no body run here, the same ones processJobs() starts with, so a board
// that might match still goes through fetch() and processJobs() unchanged.
// Undated postings count as possible: processJobs() decides them.
export function listingMayMatch(listing, { cutoff, titleFilter, companySlug }) {
  return listing.some(job => job.url && job.title
    && classifyPostingDate(job, cutoff) !== 'stale'
    && titleFilter(job.title, companySlug));
}

// Undated postings on a board the listing ruled out. processJobs() counts
// these into "Undated dropped" before its title filter, so a skipped board
// must report them too, or the pre-check would hide the degraded-scan signal.
export function undatedInListing(listing, cutoff) {
  return listing.filter(job => job.url && job.title
    && classifyPostingDate(job, cutoff) === 'undated').length;
}

// Apply the same user-owned do-not-apply gate as scan.mjs to reverse-scan
// results. Absent/empty blacklist is a no-op. Default skips are counted and
// never silent; --include-blacklisted keeps matches but marks them for audit.
export function filterBlacklistedOffers(offers, blacklist, { includeBlacklisted = false } = {}) {
  if (!blacklist || blacklist.size === 0) {
    return { offers, filteredBlacklist: 0, annotatedBlacklisted: 0 };
  }

  const kept = [];
  let filteredBlacklist = 0;
  let annotatedBlacklisted = 0;

  for (const offer of offers) {
    const entry = findBlacklistEntry(blacklist, offer.company || '', offer.url);
    if (!entry) {
      kept.push(offer);
      continue;
    }

    if (!includeBlacklisted) {
      filteredBlacklist++;
      continue;
    }

    annotatedBlacklisted++;
    const label = `blacklisted${entry.reason ? `: ${entry.reason}` : ''}`;
    kept.push({
      ...offer,
      blacklisted: true,
      note: typeof offer.note === 'string' && offer.note.trim()
        ? `${label} — ${offer.note}`
        : label,
    });
  }

  return { offers: kept, filteredBlacklist, annotatedBlacklisted };
}

// `title_filter` is written for scan.mjs, whose corpus is the curated
// tracked_companies list. That corpus is what makes a broad keyword safe:
// "Backend" at a company you already vetted is a real lead, and dropping it
// costs real roles — so tightening the shared key to protect this sweep is a
// regression for the scanner that actually produces applications.
//
// This sweep removes the corpus and leaves the same keywords carrying the
// whole burden against every public board. Broad keywords then match on a
// scale they were never chosen for, and ordinary English words land as
// unrelated product names: one 38,854-board sweep wrote 426 postings, of
// which 142 passed on "Backend" alone and the rest on Fuel Associate,
// Traffic Anchor, ICP Mass Spec, Principal Mechanical Canister Eng. Across
// 194 distinct companies, one matched a crypto name — a gas-station chain.
//
// Optional `title_filter_full` lets one portals.yml carry a strict profile
// for this sweep and a broad one for scan.mjs. Absent — the default — this
// returns `title_filter` and behaviour is byte-identical to before.
export function resolveTitleFilterConfig(config) {
  return config?.title_filter_full ?? config?.title_filter;
}

// The company-level gate (#3105). `title_filter` runs on the wrong axis for a
// reverse sweep: it asks the TITLE to answer both "is this the right role" and
// "is this employer in my industry", and a title cannot answer the second one —
// "Senior Backend Engineer" is character-for-character identical at a Solana
// infrastructure company and at a supermarket chain. scan.mjs never had that
// problem because tracked_companies settles the employer before a title is
// read; the sweep deletes that premise and hands the whole burden to keywords
// that were never chosen for it.
//
// So ask the cheaper question of the board as a whole first, using data already
// paid for — the board is fully in memory before filtering starts, so this
// costs one extra pass over an array and no HTTP request at all.
//
// THRESHOLD IS ONE, and that is measured rather than cautious. Over 546
// postings on 249 boards, raising it does not trade recall for precision, it
// inverts the gate: at two, twelve boards survive and every genuine crypto
// employer is gone, because ten of the twelve qualify on `defi` appearing
// repeatedly inside a Portuguese "Deficiência" or an Ohio "Defiance". A
// repeated false-positive substring is a property of large boards; the real
// finds are single openings at small companies — conduit, ethereum-foundation,
// mesh and ing each had exactly one domain-bearing posting, which is precisely
// the find the sweep exists for.
//
// KNOWN LIMIT: the gate assumes one board is one employer, and an aggregator
// breaks that premise. `jobgether` carried 25 domain-bearing postings among 93
// and so passes at every threshold under every matcher, admitting 68 unrelated
// ones on the strength of its crypto listings. No threshold fixes it — the
// board is simply not an employer. There the gate degrades to today's
// behaviour, which is a bounded failure and one of the reasons the whole
// feature is opt-in.
export function boardInDomain(jobs, domainFilter) {
  return jobs.some(job => job?.title && domainFilter(job.title));
}

/**
 * Decide what the sweep does with one fetched board, before any title is read.
 *
 * Split out from the call site because the truncation rule is the whole point
 * and an inline `if` hid it: an earlier revision gated a `workdayTruncated`
 * response on the part that was fetched, "like every other gate here". That
 * reasoning is wrong for exactly this gate. Every other filter here judges a
 * posting it has in hand, so a short response costs it the rows it never saw;
 * this one judges the BOARD from its rows, so a short response can invert the
 * verdict — the only domain-bearing posting may sit in the tail the sequential
 * retry is about to fetch, and the board is then dropped into a counter that
 * reads as "correctly excluded" rather than "not yet known".
 *
 * So an unmatched truncated board is deferred, not gated: it stays queued for
 * the retry and the fuller result decides it. A truncated board that ALREADY
 * matches needs no deferral — a domain-bearing posting in hand settles the
 * verdict whatever the retry returns — and processing its partial page keeps
 * today's behaviour of banking those matches even if the retry later fails.
 * What the retry then does with each kind is retryGateDecision's job.
 *
 * Reachability, since the obvious objection is that a truncated page might be
 * a ranked prefix and so already carry any domain-bearing posting: it is not.
 * `workdayTruncated` is set on `fetch-error` (retries exhausted MID-pagination
 * while other tenants hammered the same uplink), on `splitIncomplete` and on
 * `budgetExhausted` — see providers/workday.mjs. The cut is wherever the error
 * or the budget landed, which bears no relation to posting order, so no
 * ordering argument makes the deferral unnecessary.
 *
 * @param {object[]} jobs - Postings as returned by the provider.
 * @param {((title: string) => boolean)|null} domainFilter - Compiled gate, or null when opt-out.
 * @returns {'process'|'gate'|'defer'} What the caller should do with the board.
 */
export function boardGateDecision(jobs, domainFilter) {
  if (!domainFilter) return 'process';
  if (boardInDomain(jobs, domainFilter)) return 'process';
  // An iCIMS board stopped at the provider's page cap is never retried, so,
  // like a structural Workday cut, it is admitted ungated rather than judged
  // on the pages that happened to fit under the cap.
  if (jobs.icimsTruncated) return 'process';
  if (!jobs.workdayTruncated) return 'gate';
  // Only a transient cut is retried. A structural one comes back cut at the
  // same bound, so there is no fuller fetch to defer to: the board is admitted
  // ungated now, as the retry admits a board truncated twice.
  return jobs.workdayTruncated === WORKDAY_TRUNCATED_REASON.TRANSIENT ? 'defer' : 'process';
}

/**
 * Decide what the sequential retry does with a board the sweep queued.
 *
 * The retry is where a deferred board gets its verdict, so it needs the same
 * callable shape as the sweep, plus the one fact only the sweep knows: whether
 * the board was deferred or already admitted. Two cases hinge on that:
 *
 *   1. Already admitted (its truncated page matched). The retry cannot take
 *      that back: the retry can come back truncated too, cut somewhere else,
 *      so "the retry returns a superset" does not hold, and a cut that misses
 *      the domain posting would count a board whose matches are already banked
 *      as gated. It is processed, whatever the retry shows.
 *   2. Deferred, and the retry is truncated AGAIN with nothing domain-bearing.
 *      There is no third fetch to defer to, and the gated counter means
 *      "correctly excluded", which an incomplete page cannot establish. The
 *      board is admitted ungated, the same fallback the gate takes whenever it
 *      cannot judge a board (see the aggregator note on boardInDomain); the
 *      caller still counts it as an error for staying truncated.
 *
 * Only a deferred board whose retry is complete and carries nothing
 * domain-bearing is gated.
 *
 * @param {object[]} jobs - Postings from the retry fetch.
 * @param {((title: string) => boolean)|null} domainFilter - Compiled gate, or null when opt-out.
 * @param {boolean} wasDeferred - Whether the sweep deferred this board rather than processing it.
 * @returns {'process'|'gate'} What the caller should do with the board.
 */
export function retryGateDecision(jobs, domainFilter, wasDeferred) {
  if (!wasDeferred) return 'process';
  return boardGateDecision(jobs, domainFilter) === 'gate' ? 'gate' : 'process';
}

// Title/location/content filter chain for one posting, used by runSeedScan().
// The main ATS-directory loop below inlines the same three checks (it tracks
// a droppedContent counter per stage for the run summary), but this shared,
// pure, exported helper keeps the content_filter.by_title_keyword wiring
// (#1846) unit-testable without mocking providers or duplicating the rule
// order in two places for the caller that doesn't need per-stage counts.
export function passesFilters(job, { titleFilter, locationFilter, contentFilter, titleFilterConfig, companySlug }) {
  if (!titleFilter(job.title, companySlug)) return false;
  // job.url is passed so the location filter can fall back to the URL's own
  // location segment when the provider reports a rolled-up "N Locations" string;
  // job.title so a title-stated remote role survives a city-only location.
  if (!locationFilter(job.location, job.url, job.title, job)) return false;
  if (contentFilter && !contentFilter(job.description, matchedTitleKeywords(job.title, titleFilterConfig))) return false;
  return true;
}

// Prefer a provider's own scoped dedup key over URL normalization when the
// provider can derive one — e.g. workday.mjs's requisition ID, which
// collapses the same posting served under several sites of one tenant
// (different paths, sometimes different hosts) that seenJobKey
// can't recognize as duplicates (#3439). `provider` is optional so this stays
// safe to call for seed-pass offers, which carry no SOURCES entry to look one
// up from. Falls back to the pre-#3439 behavior whenever no key is derived.
export function dedupTokenFor(job, provider) {
  return provider?.dedupKey?.(job) || seenJobKey(job.url);
}

// Resolve an offer's provider from its recorded `source` string — shared by
// the checkpoint-resume reseed below and by loadSeenUrls' extraTokensFor
// hook. `source` is "{sourcesKey}-full" for the main sweep and
// "{seedId}-seed" for seed offers; SOURCES has no seed entries, so a seed
// offer's lookup misses and callers fall back to URL-only dedup, unchanged
// from before #3439.
export function providerForSource(source) {
  return SOURCES[String(source || '').replace(/(?:-history)?-full$/, '')]?.provider;
}

// Cap-aware company sampling. Default: the dataset's natural (alphabetical)
// prefix. With --shuffle: a random sample of `limit` companies, so a capped
// scan isn't always biased to the same alphabetical-first slice. Pure; returns
// a new array and never mutates `list`.
export function sampleCompanies(list, limit, shuffle = false) {
  if (!shuffle || limit >= list.length) return list.slice(0, limit);
  const copy = list.slice();
  for (let i = copy.length - 1; i > 0; i--) {
    const j = Math.floor(Math.random() * (i + 1));
    [copy[i], copy[j]] = [copy[j], copy[i]];
  }
  return copy.slice(0, limit);
}

/** One stderr JSON line per kept offer in `--json` mode so Explore can paint
 *  cards while the walk continues. Stdout stays the single summary object (#1199). */
export function formatLiveOfferLine(job, source) {
  let postedAt = null;
  if (job?.postedAt) {
    const d = new Date(job.postedAt);
    if (!Number.isNaN(d.getTime())) postedAt = d.toISOString().slice(0, 10);
  }
  return JSON.stringify({
    kind: 'offer',
    company: job.company,
    title: job.title,
    url: job.url,
    location: job.location || null,
    ...(job.geographyEvidence ? { geographyEvidence: job.geographyEvidence } : {}),
    postedAt,
    source: source || job.source || '',
  });
}

export function parseLiveOfferLine(line) {
  const raw = String(line || '');
  const start = raw.indexOf('{');
  if (start < 0) return null;
  let ev;
  try {
    ev = JSON.parse(raw.slice(start));
  } catch {
    return null;
  }
  if (!ev || ev.kind !== 'offer') return null;
  const url = typeof ev.url === 'string' ? ev.url.trim() : '';
  if (!url || !ev.company || !ev.title) return null;
  return ev;
}

export function emitLiveOffer(job, source, { json } = {}) {
  if (!json || !job?.url || !job.company || !job.title) return;
  console.error(formatLiveOfferLine(job, source));
}

export function keepAndMaybeEmit(job, source, sink, blacklist, opts) {
  const kept = { ...job, ...(opts.workplacePolicy ? { geographyEvidence: captureWorkplaceEvidence(job) } : {}), source, dateStatus: job.postedAt ? 'dated' : 'unknown' };
  sink.push(kept);
  const live = filterBlacklistedOffers([kept], blacklist, { includeBlacklisted: opts.includeBlacklisted });
  if (live.offers.length) emitLiveOffer(live.offers[0], source, { json: opts.json });
}

// ── VC portfolio seed scan ──────────────────────────────────────────

// ATS providers that can auto-detect from a careers_url, in probe order.
// Workday is excluded: its URL format requires a tenant|instance|site triple
// that can't be derived from a portfolio slug alone.
const SEED_PROVIDERS = [greenhouse, lever, ashby];

/**
 * Scan a VC portfolio seed source and return matching job offers.
 * Companies are converted to PortalEntry shape, then each ATS provider's
 * detect() is tried in order (greenhouse → lever → ashby). The first hit
 * wins and its fetch() is called — identical to how portals.yml tracked
 * companies flow through scan.mjs.
 *
 * @param {string}   seedId      Key from SEED_SOURCES (e.g. 'yc').
 * @param {object}   opts        Parsed CLI options.
 * @param {object}   ctx         HTTP context from makeHttpCtx().
 * @param {Set}      seenUrls    Shared dedup set (mutated in place).
 * @param {string}   label       Human-readable source label for logs.
 * @param {object}   [idCaches]  Store from openProviderIdCache(); each provider gets its slice as ctx.idCache.
 * @returns {Promise<object[]>}  New job offers (same shape as ATS scan offers).
 */
export async function runSeedScan(seedId, opts, ctx, seenUrls, label, idCaches) {
  const source = SEED_SOURCES[seedId];
  if (!source) throw new Error(`runSeedScan: unknown seed "${seedId}"`);

  let companies;
  try {
    companies = await source.fetch();
  } catch (err) {
    console.error(`⚠️  ${seedId}: could not fetch portfolio — ${err.message}`);
    return [];
  }

  // Apply the --limit cap here too (by slug, consistent with sampleCompanies).
  const capped = opts.limit < companies.length
    ? (opts.shuffle
      ? (() => { const c = companies.slice(); for (let i = c.length - 1; i > 0; i--) { const j = Math.floor(Math.random() * (i + 1)); [c[i], c[j]] = [c[j], c[i]]; } return c.slice(0, opts.limit); })()
      : companies.slice(0, opts.limit))
    : companies;

  const cutoff = Date.now() - opts.sinceDays * 86_400_000;
  const offers = [];
  let errors = 0;

  await parallelEach(capped, CONCURRENCY, async (company) => {
    const entry = toPortalEntry(company);
    if (!entry.careers_url) return;

    // Try each ATS provider's detect() — first hit wins.
    let provider = null;
    for (const p of SEED_PROVIDERS) {
      try {
        if (p.detect?.(entry)) { provider = p; break; }
      } catch { /* no-op */ }
    }
    if (!provider) return; // No ATS detected — skip silently (or log in --verbose).

    let jobs;
    try {
      jobs = await provider.fetch(entry, withIdCache(ctx, idCaches, provider));
    } catch (err) {
      errors++;
      if (opts.verbose) console.error(`  ✗ ${seedId}/${entry.name}: ${err.message}`);
      return;
    }

    const sourceName = `${seedId}-seed`;
    for (const job of jobs) {
      if (!job.url || !job.title) continue;
      const dateClass = classifyPostingDate(job, cutoff);
      if (dateClass === 'stale') continue;
      if (dateClass === 'undated' && !opts.includeUndated) continue;
      if (!passesFilters(job, {
        titleFilter: opts.titleFilter,
        locationFilter: opts.locationFilter,
        contentFilter: opts.contentFilter,
        companySlug: entry.name,
        titleFilterConfig: opts.titleFilterConfig,
      })) continue;
      // provider is always one of SEED_PROVIDERS (greenhouse/lever/ashby) here —
      // none currently define dedupKey, so this is the same seenJobKey
      // behavior as before; kept via the shared helper so the two dedup sites
      // in this file can't quietly drift apart (#3439).
      const dedupToken = dedupTokenFor(job, provider);
      if (seenUrls.has(dedupToken)) continue;
      seenUrls.add(dedupToken);
      keepAndMaybeEmit(job, sourceName, offers, opts.blacklist, opts);
    }
  });

  return { offers, errors, total: capped.length };
}

/**
 * Scan the ATS boards derived from the user's tracker and scan history.
 * Unsupported vendor labels stay in the local seed set but make no network
 * request. Known labels begin feeding the sweep automatically when a provider
 * with the same id is added later, without a tracker migration.
 * `idCaches` (optional, from openProviderIdCache()) gives each provider its
 * slice as ctx.idCache.
 */
export async function runHistorySeedScan(seeds, providers, opts, ctx, processJobs, idCaches) {
  const selected = opts.atsExplicit
    ? seeds.filter((seed) => opts.ats.includes(seed.vendor))
    : seeds;
  const grouped = new Map();
  for (const seed of selected) {
    if (!grouped.has(seed.vendor)) grouped.set(seed.vendor, []);
    grouped.get(seed.vendor).push(seed);
  }
  const capped = [...grouped.values()].flatMap((group) => sampleCompanies(group, opts.limit, opts.shuffle));
  const scannable = [];
  for (const seed of capped) {
    const provider = providers.get(seed.vendor);
    if (!provider) continue;
    const entry = { name: seed.company, careers_url: seed.careersUrl, provider: seed.vendor };
    try {
      // probeConfirmed: the host was verified as SuccessFactors from its page,
      // equivalent to an explicit `provider:` in portals.yml (bypasses URL detect).
      if (seed.probeConfirmed || provider.detect?.(entry)) scannable.push({ seed, provider, entry });
    } catch { /* unroutable history is a skipped seed, not a failed network request */ }
  }
  let errors = 0;
  const timeoutMs = opts.companyTimeoutMs ?? COMPANY_TIMEOUT_MS;
  const withHostSlot = createKeyedLimiter(SINGLE_HOST_CONCURRENCY);
  // An arbitrary provider may ignore cancellation or never settle. Once one
  // request times out, quarantine that host for the rest of this run: queued
  // boards are counted as errors instead of starting more requests behind the
  // six still potentially in flight. This lets the sweep finish without ever
  // exceeding the host cap.
  const quarantinedHosts = new Set();

  await parallelEach(scannable, CONCURRENCY, async ({ seed, provider, entry }) => {
    const hostKey = (() => {
      try { return new URL(entry.careers_url).hostname.toLowerCase(); } catch { return seed.vendor; }
    })();
    await withHostSlot(hostKey, () => {
      // Give the history board its own abort signal so a timed-out board's
      // in-flight requests stop (#5570). The operation.active flag below only
      // gates work BETWEEN fetches; it can't stop the fetch already in flight.
      const controller = new AbortController();
      return withConnectionScope(async () => {
      if (quarantinedHosts.has(hostKey)) {
        errors++;
        if (opts.verbose) console.error(`  ✗ ${seed.vendor}-history/${entry.name}: host skipped after an earlier timeout`);
        return;
      }
      const operation = { active: true };
      const operationPromise = (async () => {
        const jobs = await provider.fetch(entry, withIdCache(ctx, idCaches, provider));
        if (!operation.active) return;
        await processJobs(jobs, `${seed.vendor}-history`, provider, entry.name, () => operation.active);
      })();
      let timedOut = false;
      try {
        await withTimeout(operationPromise, timeoutMs, `${seed.vendor}-history/${entry.name}`, () => {
          timedOut = true;
          operation.active = false;
          controller.abort();
        });
      } catch (err) {
        operation.active = false;
        if (timedOut) quarantinedHosts.add(hostKey);
        errors++;
        if (opts.verbose) console.error(`  ✗ ${seed.vendor}-history/${entry.name}: ${err.message}`);
      } finally {
        // Observe a late rejection without waiting indefinitely. A timed-out
        // operation keeps running at most as one of the host's original six;
        // quarantining prevents queued work from replacing it.
        operationPromise.catch(() => {});
      }
      }, { signal: controller.signal });
    });
  });

  return {
    total: scannable.length,
    derived: selected.length,
    unsupported: capped.length - scannable.length,
    errors,
  };
}

// ── Parallel fetch with concurrency limit ───────────────────────────

// One company's whole fetch may not exceed this — even with per-request
// timeouts in _http.mjs, an unforeseen hang (DNS, a provider bug) must cost
// one company, not freeze a worker slot for the rest of a 12k-company sweep.
const COMPANY_TIMEOUT_MS = 5 * 60_000;

// Run one board's unit of work. For a source that gives every tenant its own
// hostname (workday, icims, bamboohr), the board gets a private connection
// pool that is destroyed when it finishes; without it, the process keeps one
// idle socket per tenant for the rest of the run, and a full sweep exhausts
// local connection capacity partway through (see providers/_connection-scope.mjs).
// Single-host sources share one pool already and run unscoped.
// The watchdog races the board's work against a timer, and on timeout it
// aborts the board's AbortController. That signal rides in the connection/abort
// scope, so every request the board issues folds it into its own timeout: a
// timed-out board stops issuing requests and retries at once (#5570), instead
// of paginating on in the background until it finishes on its own — which kept
// the process alive minutes past its own summary and burned a 4-attempt retry
// on every abandoned request.
export function boardScope(source, label, timeoutMs, fn) {
  const controller = new AbortController();
  // The signal is handed to `fn` as well as carried in the scope, so a caller
  // can tell a timed-out board from a failed one and skip any post-timeout
  // fallback work (e.g. processing a deferred board's partial page outside the
  // scope, where it would run uncancelled, #5570).
  const run = () => withTimeout(fn(controller.signal), timeoutMs, label, () => controller.abort());
  return source.singleHost
    ? withAbortScope(controller.signal, run)
    : withConnectionScope(run, { signal: controller.signal });
}

export function createKeyedLimiter(limit) {
  const states = new Map();
  return async function withKeySlot(key, fn) {
    let state = states.get(key);
    if (!state) {
      state = { active: 0, queue: [] };
      states.set(key, state);
    }
    if (state.active >= limit) {
      await new Promise((resolve) => state.queue.push(resolve));
    } else {
      state.active++;
    }
    try {
      return await fn();
    } finally {
      const next = state.queue.shift();
      if (next) next();
      else {
        state.active--;
        if (state.active === 0) states.delete(key);
      }
    }
  };
}

export function withTimeout(promise, ms, label, onTimeout = null) {
  let timer;
  const timeout = new Promise((_, reject) => {
    timer = setTimeout(() => {
      onTimeout?.();
      reject(new Error(`${label}: timed out after ${Math.round(ms / 1000)}s`));
    }, ms);
  });
  return Promise.race([promise, timeout]).finally(() => clearTimeout(timer));
}

export async function parallelEach(items, limit, fn, onItemDone = null, shouldStop = null) {
  let next = 0;
  let done = 0;
  const inFlight = new Set();
  async function worker() {
    while (next < items.length) {
      // Checked before claiming an index, so a stopped run leaves `next` where
      // the unclaimed work starts and onItemDone's resumeAt stays truthful.
      if (shouldStop && shouldStop()) return;
      const idx = next++;
      inFlight.add(idx);
      try {
        await fn(items[idx], idx);
      } finally {
        inFlight.delete(idx);
        done++;
        // Lowest not-yet-finished index: everything below it is complete, so
        // a resumed run can restart exactly here without skipping work.
        const resumeAt = inFlight.size ? Math.min(...inFlight) : next;
        if (onItemDone) onItemDone({ done, resumeAt });
      }
    }
  }
  await Promise.all(Array.from({ length: Math.min(limit, items.length) }, () => worker()));
}

// ── Liveness verification (reuses liveness-browser.mjs) ────────────

async function filterLive(offers) {
  let chromium, checkUrlLiveness, newLivenessPage;
  try {
    ({ chromium } = await import('playwright'));
    ({ checkUrlLiveness, newLivenessPage } = await import('./liveness-browser.mjs'));
  } catch (err) {
    throw new Error(
      `--liveness requires Playwright with Chromium (run "npx playwright install chromium"): ${err.message}`,
      { cause: err },
    );
  }
  console.error(`\nVerifying liveness of ${offers.length} match(es) with Playwright (sequential)...`);
  const browser = await chromium.launch(browserLaunchOptions(
    { headless: true },
    { projectRoot: SCAN_DATA_ROOT, processEnv: BROWSER_PROCESS_ENV },
  ));
  const live = [];
  try {
    const page = await newLivenessPage(browser);
    // Sequential — project rule: never Playwright in parallel
    for (const offer of offers) {
      const { result, reason } = await checkUrlLiveness(page, offer.url);
      const icon = result === 'active' ? '✅' : result === 'expired' ? '❌' : '⚠️';
      console.error(`  ${icon} ${result.padEnd(9)} ${offer.company} | ${offer.title}${result === 'expired' ? ` (${reason})` : ''}`);
      if (result !== 'expired') live.push(offer); // keep 'uncertain' — transient errors retry next scan
    }
  } finally {
    await browser.close();
  }
  console.error(`  → ${live.length}/${offers.length} passed liveness`);
  return live;
}

// ── Main ────────────────────────────────────────────────────────────

// A provider can hand back an unparseable postedAt; new Date(bad).toISOString()
// throws on an invalid Date, which in the SIGTERM partial path would abort the
// whole dump and lose every collected offer. Degrade one bad value to null.
function isoDay(v) {
  if (!v) return null;
  const d = new Date(v);
  return Number.isNaN(d.getTime()) ? null : d.toISOString().slice(0, 10);
}

async function main() {
  const opts = parseArgs(process.argv);
  let checkpoint = null;
  if (opts.resume) {
    const cp = loadCheckpoint();
    if (!cp) {
      console.error(`Error: --resume passed but no checkpoint found at ${CHECKPOINT_PATH}.`);
      process.exit(1);
    }
    if (!checkpointCompatible(cp, opts)) {
      console.error('Error: checkpoint was written with different settings (--ats/--limit/--include-undated, or --shuffle is set) — rerun with the original flags, or delete the checkpoint to start fresh.');
      process.exit(1);
    }
    checkpoint = cp;
  } else if (loadCheckpoint() && !opts.dryRun) {
    console.error(`⚠️  Unfinished sweep checkpoint found at ${CHECKPOINT_PATH} — pass --resume to continue it; this fresh run will overwrite it.`);
  }
  // Resume reuses the ORIGINAL cutoff so the date window stays consistent
  // across the interruption instead of silently sliding forward.
  const cutoff = checkpoint ? checkpoint.cutoffMs : Date.now() - opts.sinceDays * 86_400_000;
  // In --json mode, stdout is reserved for the single machine-readable result,
  // so every human-facing line goes to stderr instead.
  const log = opts.json ? (...a) => console.error(...a) : (...a) => console.log(...a);
  // Same rule as `log` above: under --json the counter goes to stderr rather
  // than being dropped. Suppressing it left a sweep with no progress signal on
  // either stream, and `--dry-run --json` has no checkpoint to fall back on
  // (dry runs write no state), so a long run was indistinguishable from a hung
  // one.
  // `--json` consumers parse stderr as newline-delimited lines. The human
  // path overwrites one TTY row with `\r`; converting that to `\n` on stderr
  // keeps progress ticks and live-offer JSON from gluing onto the same line.
  const progress = (s) => {
    if (opts.json) process.stderr.write(String(s).replace(/\r$/, '\n'));
    else process.stdout.write(s);
  };

  if (!existsSync(PORTALS_PATH)) {
    console.error('Error: portals.yml not found. Run onboarding first — the reverse scan reuses its title_filter/location_filter.');
    process.exit(1);
  }
  const config = loadYaml(readFileSync(PORTALS_PATH, 'utf-8'));
  const fullTitleFilterConfig = resolveTitleFilterConfig(config);
  // title_filter_overrides is independent of title_filter_full: it broadens
  // the net for specific companies on top of whichever title filter config
  // (title_filter or title_filter_full) this run is already using.
  const titleFilterOverrides = buildTitleFilterOverrides(config?.title_filter_overrides);
  const titleFilter = buildTitleFilterWithOverrides(fullTitleFilterConfig, titleFilterOverrides);
  const locationFilter = buildLocationFilter(config?.location_filter);
  opts.workplacePolicy = config?.location_filter?.workplace_policy;
  // Same content_filter (incl. by_title_keyword scoping) scan.mjs applies —
  // see #1846. Built once here from the same portals.yml config.
  const contentFilter = buildContentFilterOrExit(config?.content_filter);
  warnRescueTitleMissesIgnored(contentFilter);
  // Opt-in, and the opt-in IS the key: `domain_filter` does not exist in any
  // portals.yml written before this feature, so every existing install keeps
  // its exact behaviour and nobody loses a board without having asked for the
  // gate. buildDomainFilter returns null for an absent or empty list, which is
  // why the summary can tell "off" from "on and nothing was skipped".
  // A bare `word:`/`stem:` would build a gate that matches nothing and drops
  // every complete board, so refuse to start rather than sweep blind. Checked
  // here as well as in validate-portals.mjs, which a direct run never calls.
  const bareDomain = barePrefixDomainKeywords(config?.domain_filter);
  if (bareDomain.length) {
    console.error(`Error: portals.yml domain_filter[${bareDomain.join(', ')}]: a word:/stem: prefix needs a term after it.`);
    process.exit(1);
  }
  const domainFilter = buildDomainFilter(config?.domain_filter);
  const filters = filterFingerprints(config);
  const changedFilters = changedFilterKeys(checkpoint, filters);
  if (changedFilters.length) {
    console.error(`Error: portals.yml ${changedFilters.join(', ')} changed since the checkpoint was written — the boards already swept were filtered under the old settings. Restore them to finish this sweep, or delete ${CHECKPOINT_PATH} to start fresh.`);
    process.exit(1);
  }
  if (!fullTitleFilterConfig?.positive?.length) {
    const key = config?.title_filter_full ? 'title_filter_full' : 'title_filter';
    console.error(`⚠️  portals.yml has no ${key}.positive — every fresh posting on every board will match. Consider adding keywords.`);
  }
  // Attach filters to opts so runSeedScan can use them without extra parameters.
  opts.titleFilter = titleFilter;
  opts.locationFilter = locationFilter;
  opts.contentFilter = contentFilter;
  // Raw title_filter config, needed by matchedTitleKeywords() to scope
  // content_filter.by_title_keyword the same way scan.mjs does.
  opts.titleFilterConfig = fullTitleFilterConfig;

  // User-layer history is an opt-in additive board directory. It fills vendors
  // that have no public community dataset without changing the tracker schema,
  // but an ordinary reverse scan never reads the user's application history.
  let historySeeds = opts.historySeeds
    ? loadHistoryAtsSeeds({ dataRoot: DATA_ROOT, scanHistoryPath: SCAN_HISTORY_PATH })
    : [];
  // Opt-in (#5739): one plain GET per distinct unknown host, verdict cached;
  // a failed probe leaves the hostname label as it was.
  if (opts.probeUnknownHosts && historySeeds.length) {
    historySeeds = await classifyUnknownHostSeeds(historySeeds, {
      probe: makeHeadProbe(fetchTextHead),
    });
  }

  const atsSummary = opts.ats.length ? `ats: ${opts.ats.join(', ')}` : '';
  const seedsSummary = opts.seeds.length ? `seeds: ${opts.seeds.join(', ')}` : '';
  const historySummary = opts.historySeeds ? 'history seeds' : '';
  const sourcesSummary = [atsSummary, seedsSummary, historySummary].filter(Boolean).join(' | ');
  log(`Reverse ATS scan — ${sourcesSummary} | since ${opts.sinceDays}d${opts.limit < Infinity ? ` | limit ${opts.limit}/ats` : ''}${opts.shuffle ? ' | shuffled' : ''}${opts.includeUndated ? ' | +undated' : ''}${opts.liveness ? ' | liveness' : ''}${domainFilter ? ' | domain gate' : ''}${opts.dryRun ? ' | DRY RUN' : ''}`);

  // extraTokensFor: a historical scan-history.tsv row records the URL it was
  // FIRST seen on, so without this a Workday requisition seen last run under
  // site A's URL wouldn't be recognized when this run only sees it under
  // site B — the plain seenJobKey comparison never matches across
  // sites, and only fresh-run offers were reseeded with the provider key
  // (#3439). `portal` here is scan-history.tsv's recorded `offer.source`, so
  // providerForSource resolves it the same way the checkpoint reseed above
  // does.
  // scanHistoryPolicy: the recheck window scan.mjs applies to the same history.
  const { seen: seenUrls } = loadSeenUrls(scanHistoryPolicy(config), {
    extraTokensFor: (url, portal) => providerForSource(portal)?.dedupKey?.({ url }),
  });
  const blacklist = loadBlacklist();
  opts.blacklist = blacklist;
  // sinceMs and includeUndated let providers (currently only workday.mjs)
  // stop paginating a tenant early instead of always walking to max_pages:
  // sinceMs once postings are confidently past the --since window, and
  // includeUndated (when false) for a tenant that exposes no postedOn at
  // all, since its postings would all be dropped as undated below anyway.
  //
  // syntheticEntries states what this scanner's entries ARE: built from the
  // external ATS dataset, not read from portals.yml tracked_companies.
  // workday.mjs picks its cap-hit warning from it — there is no portal entry
  // here for the user to edit, so "raise max_pages on this entry" would be
  // inactionable. It used to infer that from sinceMs being set, which stopped
  // being true once #2418 taught scan.mjs --since to set it too (#2495).
  const ctx = {
    ...makeHttpCtx(),
    sinceMs: cutoff,
    includeUndated: opts.includeUndated,
    syntheticEntries: true,
    locationHints: config?.location_filter,
  };
  // Each directory source hands its provider a slice of this store as ctx.idCache.
  const idCaches = openProviderIdCache(path.join(DATA_ROOT, PROVIDER_ID_CACHE_DIR));
  // The LOCAL calendar day, not the UTC one. This value lands in
  // scan-history.tsv's first_seen, which shouldDedupScanHistoryRow measures the
  // recheck window against using the local day (#3070). Stamping it in UTC put
  // the two sides of that comparison on different clocks.
  const date = localToday();

  // Same defensive default as completedSources/counters below: a version-1
  // checkpoint that lost its offers array would otherwise set this to undefined
  // and throw on the next line, instead of degrading to an empty resume.
  //
  // Minus the matches this data root already holds. A sweep the resolver
  // breaker stopped saves its matches AND keeps its checkpoint (so --resume can
  // finish the source), so by the time --resume reads them they are already in
  // pipeline.md and scan-history: seeding them again appended every one a
  // second time at this run's persist step. A match from a run that died
  // before persisting is in neither file, so it is not in seenUrls and stays.
  // Same token as the reseed below and the sweep's own dedup check.
  const newOffers = (checkpoint?.offers || [])
    .filter(o => !seenUrls.has(dedupTokenFor(o, providerForSource(o.source))));
  // Checkpointed matches were already deduped once — without re-seeding, a
  // resumed run re-scanning the in-flight overlap would duplicate them.
  // Reseed with the SAME token a fresh dedup check would produce (provider
  // key when the source's provider has one, URL otherwise, matching
  // processJobs below) — reseeding by URL alone would miss a Workday
  // requisition's key, letting the other of its two sites' offers back in
  // after a resume even though the pre-checkpoint sweep had already
  // collapsed them (#3439). o.source is "{sourcesKey}-full" for the main
  // sweep and "{seedId}-seed" for seed offers; SOURCES has no seed entries,
  // so a seed offer's lookup misses and falls back to URL — its unchanged,
  // pre-#3439 behavior.
  for (const o of newOffers) {
    seenUrls.add(dedupTokenFor(o, providerForSource(o.source)));
  }
  const completedSources = new Set(checkpoint?.completedSources || []);
  const cc = checkpoint?.counters || {};
  let totalCompaniesScanned = cc.totalCompaniesScanned || 0;
  let totalCompaniesAvailable = 0;
  let totalErrors = cc.totalErrors || 0;
  let totalRetiredBoardsSkipped = cc.totalRetiredBoardsSkipped || 0;
  let droppedNoDate = cc.droppedNoDate || 0;
  let droppedContent = cc.droppedContent || 0;
  let capHit = false;
  // Aggregated from providers/workday.mjs's jobs.workdayNoDateSkip tag — see
  // there for why this is a counter instead of a per-company console.error
  // (thousands of tenants hit this on a full directory scan; one line each
  // would just be the same message repeated thousands of times).
  let noDateSkipCompanies = cc.noDateSkipCompanies || 0;
  let noDateSkipJobs = cc.noDateSkipJobs || 0;
  // Boards that hit a provider's hard page cap (icims). Unlike a Workday
  // truncation this isn't a network hiccup, so a sequential retry would just
  // re-hit the cap — it's reported, not retried, so capped coverage is visible
  // instead of passing for a fully-walked board.
  let cappedBoards = cc.cappedBoards || 0;
  // Skipped boards are COUNTED, never silent. The gate's real failure mode is a
  // term missing from the domain list, and a missing term costs a whole board —
  // so the run has to say how much it declined to look at, and --verbose names
  // each board so a suspicious skip can be checked by hand.
  let domainGatedBoards = cc.domainGatedBoards || 0;
  let domainGatedPostings = cc.domainGatedPostings || 0;
  const datasetStatus = {};
  // Companies each source's list held, kept in the checkpoint beside its
  // status so a resumed run can report a finished source without reading its
  // list again (see restartOnResume).
  const datasetSizes = {};

  // Graceful stop for the web layer: it SIGTERMs us when its scan budget elapses
  // (see web/src/lib/core/scan-timeout.mjs). A full sweep only prints its result
  // at the very end, so a plain kill loses everything found so far. In --json mode
  // we instead flush the matches collected up to this point as a well-formed but
  // PARTIAL result (`stoppedEarly: true`) so the caller can still surface those
  // roles. CLI/human runs keep the default terminate behavior (no handler).
  // Live deltas for the partial (stoppedEarly) report, so a mid-source SIGTERM
  // reports the work ACTUALLY done. totalCompaniesScanned is bumped by a whole
  // source's planned entries up front, and its errors fold into totalErrors only
  // once it finishes — the handler applies the same correction the checkpoint uses.
  let curEntries = 0;
  let curDone = 0;
  let curErrors = 0;
  let curDeadSkipped = 0;
  if (opts.json) {
    let flushed = false;
    process.on('SIGTERM', () => {
      if (flushed) return;
      flushed = true;
      try {
        const partial = filterBlacklistedOffers(newOffers, blacklist, {
          includeBlacklisted: opts.includeBlacklisted,
        }).offers.map((o) => ({
          company: o.company,
          title: o.title,
          url: o.url,
          location: o.location || null,
          ...(o.geographyEvidence ? { geographyEvidence: o.geographyEvidence } : {}),
          postedAt: isoDay(o.postedAt),
          dateStatus: o.dateStatus || (o.postedAt ? 'dated' : 'unknown'),
          source: o.source,
        }));
        process.stdout.write(JSON.stringify({
          date,
          sources: opts.ats,
          stoppedEarly: true,
          companiesAvailable: totalCompaniesAvailable,
          companiesScanned: Math.max(0, totalCompaniesScanned - (curEntries - curDone) - curDeadSkipped),
          capHit,
          datasetStatus,
          postingsKept: partial.length,
          postingsDroppedNoDate: droppedNoDate,
          unreachableBoards: totalErrors + curErrors,
          domainFilterActive: Boolean(domainFilter),
          domainGatedBoards,
          domainGatedPostings,
          offers: partial,
        }) + '\n', () => process.exit(0));
      } catch {
        process.exit(0);
      }
    });
  }


  const snapshotCounters = () => ({
    totalCompaniesScanned, totalErrors, totalRetiredBoardsSkipped,
    droppedNoDate, droppedContent,
    noDateSkipCompanies, noDateSkipJobs, cappedBoards,
    domainGatedBoards, domainGatedPostings,
  });
  const checkpointBase = () => ({
    version: 1,
    cutoffMs: cutoff,
    ats: opts.ats,
    limit: opts.limit === Infinity ? null : opts.limit,
    includeUndated: opts.includeUndated,
    filters,
    completedSources: [...completedSources],
    datasets: Object.fromEntries(Object.keys(datasetStatus).map((n) => [n, { status: datasetStatus[n], available: datasetSizes[n] ?? 0 }])),
    offers: newOffers,
    savedAt: new Date().toISOString(),
  });

  // Per-job filter chain, shared by the parallel sweep, the truncation retry
  // pass (workday), and date enrichment (icims). Closure over the filters and
  // counters so both passes update the same run totals.
  const processJobs = async (jobs, sourceName, provider, companySlug, shouldContinue = () => true) => {
    for (const job of jobs) {
      if (!shouldContinue()) return;
      if (!job.url || !job.title) continue;
      // Confirmed-stale postings are always dropped. Undated postings are
      // dropped by default (a reverse scan targets *fresh* roles) but
      // COUNTED so callers see the gap; --include-undated keeps them, marked.
      let dateClass = classifyPostingDate(job, cutoff);
      if (dateClass === 'stale') continue;
      // Providers whose list pages carry no date (icims) get one shot at dating
      // the posting from its detail page — but only after the cheap title and
      // location filters pass, so a 10k-tenant sweep never pays a detail-page
      // request for noise. The same (location, url) pair as the real filter
      // below: a stricter gate here would deny enrichment to jobs the final
      // check would have kept.
      //
      // Deliberately OUTSIDE the includeUndated branch. Gating enrichment on
      // !includeUndated meant --include-undated left every icims posting
      // undated, and since classifyPostingDate can never call an undated
      // posting stale, --since was silently ignored for the entire source.
      // Enrich first, then let the undated policy decide.
      if (dateClass === 'undated' && provider.enrichDate
          && titleFilter(job.title, companySlug) && locationFilter(job.location, job.url, job.title, job)) {
        try { await provider.enrichDate(job, ctx); } catch { /* stays undated */ }
        if (!shouldContinue()) return;
        dateClass = classifyPostingDate(job, cutoff);
      }
      if (dateClass === 'stale') continue;
      if (dateClass === 'undated' && !opts.includeUndated) { droppedNoDate++; continue; }
      if (!titleFilter(job.title, companySlug)) continue;
      // job.url is passed so the location filter can fall back to the URL's own
      // location segment when the provider reports a rolled-up "N Locations" string;
      // job.title so a title-stated remote role survives a city-only location.
      if (!locationFilter(job.location, job.url, job.title, job)) continue;
      if (!contentFilter(job.description, matchedTitleKeywords(job.title, fullTitleFilterConfig))) { droppedContent++; continue; }
      const dedupToken = dedupTokenFor(job, provider);
      if (seenUrls.has(dedupToken)) continue;
      seenUrls.add(dedupToken); // intra-scan dedup
      keepAndMaybeEmit(job, `${sourceName}-full`, newOffers, blacklist, opts);
    }
  };

  // Run-level, because resolverOutage below is per-source and long out of
  // scope by the time the checkpoint's fate is decided at the end of main().
  let stoppedByOutage = false;

  for (const name of opts.ats.filter((ats) => SOURCES[ats])) {
    const source = SOURCES[name];
    // A finished source whose list is rebuilt rather than cached (the Workable
    // feed, about 140 MB a read) is not read again on --resume: its status and
    // size come from the checkpoint. A checkpoint that predates those fields
    // reports it as read ('ok', since it finished) with no companies counted.
    if (completedSources.has(name) && source.restartOnResume) {
      const saved = checkpoint?.datasets?.[name];
      datasetStatus[name] = ['ok', 'stale', 'empty'].includes(saved?.status) ? saved.status : 'ok';
      const available = Number.isInteger(saved?.available) && saved.available >= 0 ? saved.available : 0;
      datasetSizes[name] = available;
      totalCompaniesAvailable += available;
      if (opts.limit < available) capHit = true;
      log(`\n⚙  ${name} — already complete in resumed checkpoint, skipping (its list is not read again)`);
      continue;
    }
    // For every other source, stats are accumulated BEFORE the completed-source
    // skip: on --resume a source finished before the checkpoint was written
    // still contributed its companies to the sweep, so it must still count
    // toward the availability / cap figures the final report and --json output
    // show. Reading its list again is cheap, since the list comes from the 24h
    // directory cache.
    const { list, status } = await loadSourceList(name, source, { cutoff, includeUndated: opts.includeUndated, titleFilter });
    datasetStatus[name] = status;
    datasetSizes[name] = list.length;
    totalCompaniesAvailable += list.length;
    if (opts.limit < list.length) capHit = true;
    if (completedSources.has(name)) {
      log(`\n⚙  ${name} — already complete in resumed checkpoint, skipping`);
      continue;
    }
    // A source that restarts on resume never compares fingerprints, and its
    // list carries every kept posting, so it is not hashed.
    const datasetHash = source.restartOnResume ? null : datasetFingerprint(list);
    const entriesAll = sampleCompanies(list, opts.limit, opts.shuffle).map(source.toEntry).filter(Boolean);

    let startAt = 0;
    if (checkpoint && checkpoint.current?.name === name && source.restartOnResume) {
      log(`  ${name} — interrupted in the checkpoint; its list is rebuilt, so the pass starts again`);
    } else if (checkpoint && checkpoint.current?.name === name) {
      // datasetHash is absent in checkpoints written before this guard existed;
      // fall back to the length-only check for those rather than hard-failing.
      const hashMismatch = checkpoint.current.datasetHash != null
        && checkpoint.current.datasetHash !== datasetHash;
      if (checkpoint.current.datasetLen !== list.length || hashMismatch) {
        console.error(`Error: ${name} company dataset changed since the checkpoint — resume order is no longer valid. Delete ${CHECKPOINT_PATH} and rerun.`);
        process.exit(1);
      }
      startAt = checkpoint.current.resumeAt;
    }
    const entries = entriesAll.slice(startAt);
    totalCompaniesScanned += entries.length;
    curEntries = entries.length; curDone = 0; curErrors = 0; curDeadSkipped = 0;
    log(`\n⚙  ${name} — ${entriesAll.length} companies${status !== 'ok' ? ` (dataset: ${status})` : ''}${startAt ? ` — resuming at ${startAt}` : ''}`);

    // This source's slice of the provider id cache (provider-id-cache.mjs).
    const sourceCtx = { ...ctx, idCache: idCaches.forProvider(source.provider.id) };
    let errors = 0;
    let deadBoardsSkipped = 0;
    const deadBoards = loadDeadBoards(DEAD_BOARDS_PATH);
    let consecutiveResolverFailures = 0;
    let resolverOutage = false;
    // The board whose failure tripped the breaker. `name` is only the ATS
    // vendor, and the resume offset is only a position — neither tells the user
    // which board to try by hand once the resolver is back.
    let resolverOutageCompany = null;
    // Latest progress reported by parallelEach. It computes both on every item
    // but keeps neither, and a run stopped mid-source needs them after the
    // call returns to checkpoint where it actually stopped (#2283).
    let lastDone = 0;
    let lastResumeAt = 0;
    const truncated = [];
    // The subset of `truncated` the gate deferred rather than admitted; the
    // retry needs to know which, see retryGateDecision. Each keeps its index in
    // entriesAll, so a checkpoint can carry it past the resume offset, and its
    // partial page, so a failed retry still has something to process.
    const deferred = new Map();
    // A board the interrupted run deferred sits below the resume offset, so
    // the sweep never revisits it: it goes straight to the retry. Its partial
    // page died with that run, so a failed retry has nothing to fall back to.
    for (const index of checkpoint?.current?.name === name ? checkpoint.current.deferred ?? [] : []) {
      if (index >= startAt || !entriesAll[index]) continue;
      truncated.push(entriesAll[index]);
      deferred.set(entriesAll[index], { index, jobs: null });
    }
    const deferredIndices = () => [...deferred.values()].map((d) => d.index);
    await parallelEach(entries, source.concurrency ?? CONCURRENCY, async (entry, idx) => {
      const deadBoard = boardKey(entry);
      if (shouldSkipDeadBoard(deadBoards, name, deadBoard)) {
        deadBoardsSkipped++;
        return;
      }
      try {
        // The whole per-company unit — fetch AND processJobs (which may issue
        // per-job detail-page requests via provider.enrichDate) — runs inside
        // one watchdog, so enrichment latency can't blow past COMPANY_TIMEOUT_MS.
        await boardScope(source, `${name}/${entry.name}`, COMPANY_TIMEOUT_MS, async () => {
          // Most boards in a public directory hold nothing the title filter
          // wants; where the provider can list a board cheaply, find that out
          // before downloading every posting body.
          if (typeof source.provider.fetchListing === 'function') {
            const listing = await source.provider.fetchListing(entry, sourceCtx);
            if (!listingMayMatch(listing, { cutoff, titleFilter, companySlug: entry.name })) {
              if (!opts.includeUndated) droppedNoDate += undatedInListing(listing, cutoff);
              recordBoardResult(deadBoards, name, deadBoard, 200);
              consecutiveResolverFailures = 0;
              return;
            }
          }
          const jobs = await source.provider.fetch(entry, sourceCtx);
          recordBoardResult(deadBoards, name, deadBoard, 200);
          consecutiveResolverFailures = 0;
          // Queued BEFORE the gate, deliberately: a board the gate cannot yet
          // judge has to stay in the retry list, and the gate below is what
          // decides whether it was ever eligible. Only 'transient' is worth a
          // sequential retry — 'structural' means the board hit a fixed bound
          // (facet-split slice/depth/page budget) that a repeat run reaches
          // again, paying the same expensive split for the same result.
          if (jobs.workdayTruncated === WORKDAY_TRUNCATED_REASON.TRANSIENT) truncated.push(entry);
          // Ahead of processJobs, deliberately: a gated board is not merely
          // quieter but CHEAPER, since provider.enrichDate() issues a per-job
          // detail request for undated providers (icims) and a board dropped
          // here never pays for one. The one board that does not get decided
          // here is the truncated one — see boardGateDecision.
          const decision = boardGateDecision(jobs, domainFilter);
          if (decision === 'gate') {
            domainGatedBoards++;
            domainGatedPostings += jobs.length;
            if (opts.verbose) console.error(`  ⊘ ${name}/${entry.name}: no domain-bearing posting — board skipped`);
            return;
          }
          // Deferred: the retry below re-fetches the whole board and gates
          // THAT. Its partial page is not processed here — those rows would
          // bypass a gate that has not been decided yet, which is the loose
          // direction this feature exists to close. It is kept, though, for
          // the case where the retry fetch fails outright.
          if (decision === 'defer') {
            deferred.set(entry, { index: startAt + idx, jobs });
            return;
          }
          if (jobs.icimsTruncated) {
            cappedBoards++;
            if (opts.verbose) console.error(`  ⚠ ${name}/${entry.name}: hit the page cap — later postings not scanned`);
          }
          if (jobs.workdayNoDateSkip) { noDateSkipCompanies++; noDateSkipJobs += jobs.length; }
          await processJobs(jobs, name, source.provider, entry.name);
        });
      } catch (err) {
        // Mostly defunct boards in the public dataset — expected noise, so the
        // default stays quiet; --verbose surfaces per-board failures.
        errors++;
        recordBoardFailure(deadBoards, name, deadBoard, err);
        // A dead board and a dead resolver look identical one at a time; only
        // the *consecutive* run tells them apart, so any non-resolver outcome
        // resets the count (#2229).
        if (isResolverFailure(err)) {
          // Record only the first board past the limit: in-flight boards keep
          // failing after the flag is set, and the last of them is not the one
          // that tripped it.
          if (++consecutiveResolverFailures >= RESOLVER_FAILURE_LIMIT && !resolverOutage) {
            resolverOutage = true;
            resolverOutageCompany = entry.name;
          }
        } else {
          consecutiveResolverFailures = 0;
        }
        if (opts.verbose) console.error(`  ✗ ${name}/${entry.name}: ${err.message}`);
      }
    }, ({ done, resumeAt }) => {
      lastDone = done;
      curDone = done; curErrors = errors; curDeadSkipped = deadBoardsSkipped;
      lastResumeAt = resumeAt;
      if (done % 200 === 0 || done === entries.length) {
        progress(`  ${done}/${entries.length} scanned, ${newOffers.length} total matches\r`);
      }
      if (done % CHECKPOINT_EVERY === 0 && !opts.dryRun) {
        saveDeadBoardsBestEffort(deadBoards);
        // So a sweep that dies later keeps the ids it resolved so far.
        void saveProviderIdCacheBestEffort(idCaches);
        writeCheckpoint({
          ...checkpointBase(),
          current: { name, resumeAt: startAt + resumeAt, datasetLen: list.length, datasetHash, deferred: deferredIndices() },
          counters: {
            ...snapshotCounters(),
            totalRetiredBoardsSkipped: totalRetiredBoardsSkipped + deadBoardsSkipped,
            // totalCompaniesScanned was bumped by the FULL entries.length up
            // front; a checkpoint must store only work actually attempted, or
            // a resumed run (which re-adds its own slice) double-counts the
            // completed portion in the final summary. Retired boards were not
            // attempted and must not be counted as scanned either.
            totalCompaniesScanned: totalCompaniesScanned - (entries.length - done) - deadBoardsSkipped,
            totalErrors: totalErrors + errors,
          },
        });
      }
    }, () => resolverOutage);
    // parallelEach reports a completed callback even when a retired board was
    // skipped, so remove those entries from the live scan total after the
    // source pass. The retired count remains visible separately.
    totalCompaniesScanned -= deadBoardsSkipped;
    // Second chance for boards the parallel sweep truncated: retry alone on a
    // quiet line. Re-processing the full board is safe — seenUrls already
    // holds every match from the partial first pass.
    // Skipped entirely under a resolver outage: retrying boards one by one is
    // more of exactly the traffic the breaker just stopped. A board deferred by
    // the gate is then neither processed nor counted here; the checkpoint
    // carries its index, and the resumed run retries it.
    if (truncated.length && !resolverOutage) {
      log(`\n  ↻ retrying ${truncated.length} truncated board(s) sequentially...`);
      for (const entry of truncated) {
        let retrySignal;
        try {
          await boardScope(source, `${name}/${entry.name} (retry)`, COMPANY_TIMEOUT_MS, async (signal) => {
            retrySignal = signal;
            const jobs = await source.provider.fetch(entry, sourceCtx);
            recordBoardResult(deadBoards, name, boardKey(entry), 200);
            // Where a board deferred by the parallel sweep is actually decided.
            // A board the sweep already admitted is never gated here, and one
            // still truncated is admitted rather than counted as gated.
            if (retryGateDecision(jobs, domainFilter, deferred.has(entry)) === 'gate') {
              domainGatedBoards++;
              domainGatedPostings += jobs.length;
              if (opts.verbose) console.error(`  ⊘ ${name}/${entry.name}: no domain-bearing posting — board skipped`);
            } else {
              await processJobs(jobs, name, source.provider, entry.name);
            }
            if (jobs.workdayTruncated) {
              errors++; // still not fully covered — move on
              // A board pushed here as 'transient' can legitimately come back
              // 'structural': the retry's root crawl succeeded, the clamp got
              // detected for the first time, and the split then hit its own
              // bound — that's a real first split, not a repeat.
              if (opts.verbose) {
                const why = jobs.workdayTruncated === WORKDAY_TRUNCATED_REASON.STRUCTURAL
                  ? 'facet split hit its bound' : 'still truncated';
                console.error(`  ✗ ${name}/${entry.name}: ${why} after sequential retry`);
              }
            }
          });
        } catch (err) {
          errors++;
          recordBoardFailure(deadBoards, name, boardKey(entry), err);
          if (opts.verbose) console.error(`  ✗ ${name}/${entry.name} (retry): ${err.message}`);
          // No fuller fetch is coming, so a deferred board's partial page is
          // admitted ungated, the fallback the retry takes for a board that
          // comes back truncated again. Without it, a title match already in
          // hand would be lost to the retry's failure.
          // Skip the fallback when the retry timed out: processJobs() can issue
          // enrichDate detail fetches, and here it would run outside boardScope,
          // uncancelled — the post-timeout work this fix exists to stop (#5570).
          const partial = deferred.get(entry)?.jobs;
          if (partial && !retrySignal?.aborted) await processJobs(partial, name, source.provider, entry.name);
        }
      }
    }
    totalErrors += errors;
    totalRetiredBoardsSkipped += deadBoardsSkipped;
    if (!opts.dryRun) {
      saveDeadBoardsBestEffort(deadBoards);
      await saveProviderIdCacheBestEffort(idCaches);
    }
    if (resolverOutage) {
      // Deliberately before completedSources/checkpoint: this source did NOT
      // finish, and marking it done would make --resume skip the rest of it.
      stoppedByOutage = true;
      // The counter was bumped by the FULL entries.length up front, but the
      // breaker left entries.length - lastDone boards unattempted. Correct the
      // live counter, not just the checkpoint payload: this run's own summary
      // and --json would otherwise report companies nobody ever contacted as
      // scanned. Correcting it here also gives the checkpoint the right figure
      // for free — a resumed run re-adds its own slice, so a checkpoint holding
      // the full slice makes the completed portion count twice.
      totalCompaniesScanned -= entries.length - lastDone;
      // Pin the resume point here rather than leaving the last periodic write
      // to stand for it. That one fires every CHECKPOINT_EVERY companies, so
      // it can be up to 500 boards behind where the breaker actually stopped —
      // and --resume would replay all of them against the resolver that just
      // refused.
      let checkpointWritten = false;
      if (!opts.dryRun) {
        checkpointWritten = writeCheckpoint({
          ...checkpointBase(),
          current: { name, resumeAt: startAt + lastResumeAt, datasetLen: list.length, datasetHash, deferred: deferredIndices() },
          counters: snapshotCounters(),
        });
      }
      log(`\n  ⛔ stopped ${name}/${resolverOutageCompany}: ${RESOLVER_FAILURE_LIMIT} consecutive DNS failures.`);
      log(`     Your resolver is refusing queries — it may be rate-limiting this host.`);
      log(`     Lower CONCURRENCY, raise the resolver's per-client limit, or set`);
      log(`     CAREER_OPS_NO_DNS_CACHE=1 only if you know the cache is at fault.`);
      // Only claim resumability when a checkpoint actually exists: --dry-run
      // writes none, and a failed write (ENOSPC, read-only volume) leaves at
      // best the last periodic checkpoint — nothing at the offset named here.
      if (checkpointWritten) log(`     Rerun with --resume once the resolver recovers — checkpointed at company ${startAt + lastResumeAt} of ${name}.`);
      break;
    }
    completedSources.add(name);
    // Source finished and folded into the totals — clear the live deltas so a
    // SIGTERM between sources reports the totals, not this source's slice twice.
    curEntries = 0; curDone = 0; curErrors = 0; curDeadSkipped = 0;
    if (!opts.dryRun) {
      writeCheckpoint({ ...checkpointBase(), current: null, counters: snapshotCounters() });
    }
    log(`\n  done (${errors} unreachable boards, ${deadBoardsSkipped} retired boards skipped)`);
  }

  // ── User tracker + scan-history ATS seeds (#3697) ───────────────
  if (historySeeds.length && !stoppedByOutage) {
    const providers = await loadProviders(PROVIDERS_DIR);
    log(`\n🌱 Application history — ${historySeeds.length} unique ATS board(s) derived locally...`);
    const result = await runHistorySeedScan(historySeeds, providers, opts, ctx, processJobs, idCaches);
    totalCompaniesScanned += result.total;
    totalErrors += result.errors;
    log(`  done — ${result.total} board(s) probed, ${result.unsupported} without a scanner provider (${result.errors} errors)`);
  }

  // ── VC portfolio seed sources (--seeds flag) ───────────────────────
  for (const seedId of opts.seeds) {
    const seedSource = SEED_SOURCES[seedId];
    log(`\n🌱 ${seedSource.label} (${seedId}-seed) — fetching portfolio...`);
    const result = await runSeedScan(seedId, opts, ctx, seenUrls, seedSource.label, idCaches);
    if (result && result.offers) {
      totalCompaniesScanned += result.total || 0;
      totalErrors += result.errors || 0;
      newOffers.push(...result.offers);
      log(`  done — ${result.total} companies probed, ${result.offers.length} matches (${result.errors} errors)`);
    }
  }
  // The history and VC seed passes run after every directory source has
  // saved, so what they resolved is saved here.
  if (!opts.dryRun) await saveProviderIdCacheBestEffort(idCaches);

  const blacklistResult = filterBlacklistedOffers(newOffers, blacklist, { includeBlacklisted: opts.includeBlacklisted });
  let offers = blacklistResult.offers;
  if (offers.length && opts.liveness) offers = await filterLive(offers);
  offers.sort((a, b) => (b.postedAt || 0) - (a.postedAt || 0));

  printScanSummaryHeader('Reverse ATS Scan', date, log);
  log(`Companies scanned:  ${totalCompaniesScanned}${capHit ? ` of ${totalCompaniesAvailable} (capped)` : ''}`);
  log(`Unreachable boards: ${totalErrors}`);
  if (cappedBoards) log(`Page-capped boards: ${cappedBoards} (partial coverage — later postings not scanned)`);
  // Reported whenever the gate RAN, including when it skipped nothing: a zero
  // there is the useful reading that the domain list admitted every board, and
  // silence would be indistinguishable from the gate being off.
  if (domainFilter) {
    log(`Domain-gated boards: ${domainGatedBoards} skipped, ${domainGatedPostings} posting${domainGatedPostings === 1 ? '' : 's'} never filtered${domainGatedBoards && !opts.verbose ? ' (--verbose names them)' : ''}`);
  }
  // A paced sweep is slower on purpose. Say so, or the operator reads the
  // wall-clock time as a hang (#2229).
  const pacing = dnsPacingStats();
  if (pacing.delayed > 0) {
    log(`DNS pacing:         ${pacing.delayed} lookup${pacing.delayed === 1 ? '' : 's'} delayed, ${Math.round(pacing.waitedMs / 1000)}s total wait (CAREER_OPS_DNS_LOOKUPS_PER_MIN to tune, 0 disables)`);
  }
  // noDateSkipJobs is a subset of droppedNoDate, not a separate pool: every
  // no-postedOn workday posting counted here also hits the per-job undated
  // filter in the scan loop above and gets dropped there too. Report it as
  // a breakdown, not a second count — the two aren't additive.
  if (droppedNoDate) {
    const breakdown = noDateSkipCompanies
      ? ` (incl. ${noDateSkipJobs} from ${noDateSkipCompanies} workday companies with no postedOn)`
      : '';
    log(`Undated dropped: ${droppedNoDate}${breakdown}${opts.includeUndated ? '' : '. Use --include-undated to keep'}`);
  }
  if (blacklist.size > 0) {
    if (opts.includeBlacklisted) {
      log(`Blacklisted:      ${blacklistResult.annotatedBlacklisted} let through annotated (--include-blacklisted)`);
    } else {
      log(`Blacklisted:      ${blacklistResult.filteredBlacklist} skipped (blacklist)`);
    }
  }
  if (droppedContent) log(`Content-filtered:   ${droppedContent}`);
  log(`New matches:        ${offers.length}`);

  if (offers.length) {
    log('\nNew offers:');
    for (const o of offers) {
      const posted = o.postedAt ? new Date(o.postedAt).toISOString().slice(0, 10) : 'n/a';
      const blacklistSuffix = o.blacklisted ? ' [BLACKLISTED — on your do-not-apply list]' : '';
      log(`  + [${o.source}] ${posted} | ${o.company} | ${o.title} | ${o.location || 'N/A'}${blacklistSuffix}\n    ${o.url}`);
    }
  }

  // Persist (unless dry-run, or nothing to save).
  let saved = false;
  if (offers.length && !opts.dryRun) {
    await appendToPipeline(offers);
    await appendToScanHistory(offers, date);
    saved = true;
    log(`\nResults saved to ${PIPELINE_PATH} and data/scan-history.tsv`);

    if (opts.mdOut) {
      try {
        mkdirSync(opts.mdOut, { recursive: true });
        const digest = [
          `# Reverse ATS Scan — ${date}`,
          `> ${offers.length} jobs | since ${opts.sinceDays}d | ${opts.liveness ? 'liveness ✓' : 'no liveness check'}`,
          '',
          ...offers.map(o => {
            const posted = o.postedAt ? new Date(o.postedAt).toISOString().slice(0, 10) : 'n/a';
            return `- [${o.title} @ ${o.company}](${o.url}) — ${o.location || 'N/A'} | ${o.source} | ${posted}`;
          }),
          '',
        ].join('\n');
        writeFileSync(path.join(opts.mdOut, `${date}.md`), digest, 'utf-8');
        log(`Markdown digest saved to ${path.join(opts.mdOut, `${date}.md`)}`);
      } catch (err) {
        console.error(`⚠️  Could not write markdown digest: ${err.message}`);
      }
    }
  }

  // Sweep completed — the checkpoint's job is done. A run the breaker stopped
  // did NOT complete: deleting here would destroy the resume point the outage
  // branch just wrote, which is the one case --resume exists for (#2283).
  if (!opts.dryRun && !stoppedByOutage && existsSync(CHECKPOINT_PATH)) unlinkSync(CHECKPOINT_PATH);

  // The authoritative machine-readable result: lets a caller (e.g. the web)
  // tell a *degraded* scan (capped / stale dataset / undated dropped) apart
  // from a genuinely *empty* one. In --json mode stdout carries ONLY this.
  if (opts.json) {
    process.stdout.write(JSON.stringify({
      date,
      sources: opts.ats,
      resumed: Boolean(checkpoint),
      sinceDays: opts.sinceDays,
      companiesAvailable: totalCompaniesAvailable,
      companiesScanned: totalCompaniesScanned,
      capHit,
      // The most degraded outcome this payload can report: the sweep did not
      // finish its sources, and a checkpoint is waiting for --resume. Without
      // it a caller can't tell a stopped run from a complete one except by
      // stat'ing the checkpoint file itself.
      stoppedByOutage,
      datasetStatus,
      postingsKept: offers.length,
      postingsDroppedNoDate: droppedNoDate,
      postingsFilteredBlacklist: blacklistResult.filteredBlacklist,
      postingsAnnotatedBlacklisted: blacklistResult.annotatedBlacklisted,
      postingsDroppedContent: droppedContent,
      unreachableBoards: totalErrors,
      retiredBoardsSkipped: totalRetiredBoardsSkipped,
      cappedBoards,
      // `domainFilterActive` is not derivable from the two counts: zero skipped
      // boards is equally what an absent domain_filter and a permissive one
      // produce, and a caller comparing two sweeps needs to know which.
      domainFilterActive: Boolean(domainFilter),
      domainGatedBoards,
      domainGatedPostings,
      dnsPacing: { delayed: pacing.delayed, waitedMs: Math.round(pacing.waitedMs) },
      saved,
      offers: offers.map(o => ({
        company: o.company,
        title: o.title,
        url: o.url,
        location: o.location || null,
        ...(o.geographyEvidence ? { geographyEvidence: o.geographyEvidence } : {}),
        postedAt: o.postedAt ? new Date(o.postedAt).toISOString().slice(0, 10) : null,
        dateStatus: o.dateStatus || (o.postedAt ? 'dated' : 'unknown'),
        blacklisted: Boolean(o.blacklisted),
        note: o.note || null,
        source: o.source,
      })),
    }) + '\n');
    return;
  }

  if (!offers.length) {
    log('\nNothing new.');
    return;
  }
  if (opts.dryRun) {
    log('\n(dry run — run without --dry-run to save results)');
    return;
  }
  log(`\n→ Run /career-ops pipeline to evaluate new offers.`);
}

// Only run main() when invoked directly, not when imported by tests.
if (isMainModule(import.meta.url)) {
  // Exit explicitly once main() returns. The watchdog now cancels a timed-out
  // board's work, but a keep-alive socket a single-host pool still holds (or a
  // request aborting just as the summary prints) can otherwise keep the event
  // loop open for minutes past the summary (#5570). Flush stdout first so a
  // piped summary is never truncated.
  const flushAndExit = (code) => {
    // Exit from a write-completion callback, queued after whatever is already
    // buffered, so a piped summary flushes first. `writableLength > 0` is not a
    // reliable "wait for drain" signal — `drain` only fires for a write that
    // returned false (past the high-water mark), so bytes queued below it would
    // never drain and the process would hang (the opposite of the intent).
    process.exitCode = code;
    process.stdout.write('', () => process.exit(code));
  };
  main()
    .then(() => flushAndExit(0))
    .catch(err => {
      console.error('Fatal:', err.message);
      flushAndExit(1);
    });
}