#!/usr/bin/env node
// rebuild-trigger: 2026-04-23

import { readFileSync, existsSync } from 'node:fs';
import { execFileSync } from 'node:child_process';
import { dirname, join } from 'node:path';
import { fileURLToPath } from 'node:url';
import { createRequire } from 'node:module';
import { flushPendingLlmEvents } from './lib/llm-telemetry.cjs';

import { buildEnvelope, unwrapEnvelope } from './_seed-envelope-source.mjs';
import { MAX_DECODED_PAYLOAD_BYTES, encodeGzipPayload, parseStoredSeedJson } from './_seed-payload-codec.mjs';
import { resolveRecordCount } from './_seed-contract.mjs';
// scripts/shared mirror, not ../shared: seeders deploy with rootDirectory=scripts.
import { COMPARE_AND_DELETE_SCRIPT } from './shared/compare-and-delete-script.cjs';
// Re-exported: 149 seeders import loadEnvFile from here, and the digest cron
// imports the leaf module directly so its image does not need the seed kernel.
import { isTestRuntime, loadEnvFile } from './_env-file.mjs';

export { isTestRuntime, loadEnvFile };

// process.exit does not drain in-flight promises — drain any fire-and-forget
// llm_call telemetry first (bounded by its 1.5s fetch timeout; a no-op when
// nothing is pending). Used by every runSeed exit reachable after fetchFn,
// where seeder LLM calls may have emitted events (#4954 review).
async function exitAfterTelemetryFlush(code) {
  activeLockHeartbeat?.stop();
  try { await flushPendingLlmEvents(); } catch { /* never block exit */ }
  process.exit(code);
}

const CHROME_UA = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36';
const MAX_PAYLOAD_BYTES = 5 * 1024 * 1024; // 5MB per key
const MAX_REDIS_REQUEST_BYTES = 10_000_000;
export const SEED_REDIS_COMMAND_TIMEOUT_MS = 15_000;
export const SEED_REDIS_RETRY_ATTEMPTS = 3;
export const SEED_REDIS_RETRY_BASE_MS = 1_000;
export const SEED_EXTRA_KEY_COMMAND_TIMEOUT_MS = 10_000;
export const SEED_VERIFY_COMMAND_TIMEOUT_MS = 5_000;
export const SEED_VERIFY_ATTEMPTS = 2;
export const SEED_VERIFY_RETRY_DELAY_MS = 500;

const __seed_dirname = dirname(fileURLToPath(import.meta.url));

export { CHROME_UA, MAX_PAYLOAD_BYTES };

/**
 * Resolve the CoinGecko base URL + auth header for the configured key tier.
 *
 * CoinGecko's free **Demo** plan and paid **Pro** plan share the `CG-` key
 * prefix but use *different* hosts and auth headers — a Demo key sent to the
 * Pro host fails with `HTTP 400` (error 10011: "change your root URL from
 * pro-api.coingecko.com to api.coingecko.com"), and a Pro key on the public
 * host is unauthenticated. The key string alone can't tell the tiers apart,
 * so the tier is selected explicitly by which env var is set:
 *
 *   - `COINGECKO_API_KEY`      → Pro    (pro-api.coingecko.com, x-cg-pro-api-key)
 *   - `COINGECKO_DEMO_API_KEY` → Demo   (api.coingecko.com,     x-cg-demo-api-key)
 *   - neither                  → keyless public endpoint (shared IP, 429-prone)
 *
 * Pro takes precedence so existing Pro deployments are unaffected.
 *
 * @param {Record<string, string>} [extraHeaders] merged into the returned headers
 * @returns {{ baseUrl: string, headers: Record<string, string>, tier: 'pro' | 'demo' | 'keyless' }}
 */
export function coingeckoEndpoint(extraHeaders = {}) {
  const proKey = process.env.COINGECKO_API_KEY;
  const demoKey = process.env.COINGECKO_DEMO_API_KEY;
  const headers = { Accept: 'application/json', 'User-Agent': CHROME_UA, ...extraHeaders };
  if (proKey) {
    headers['x-cg-pro-api-key'] = proKey;
    return { baseUrl: 'https://pro-api.coingecko.com/api/v3', headers, tier: 'pro' };
  }
  if (demoKey) {
    headers['x-cg-demo-api-key'] = demoKey;
    return { baseUrl: 'https://api.coingecko.com/api/v3', headers, tier: 'demo' };
  }
  return { baseUrl: 'https://api.coingecko.com/api/v3', headers, tier: 'keyless' };
}

/**
 * Fetch a CoinGecko URL, retrying 429s only while the whole phase still fits
 * `budgetMs`.
 *
 * The budget is a ceiling on the phase, in-flight request included: before
 * each backoff sleep the next attempt's full request timeout is charged
 * alongside the sleep, and the loop gives up when they would not fit. A seeder
 * run as a bundle section can therefore size `budgetMs` so its CoinPaprika
 * fallback still fits the section's timeoutMs (tests/seed-fetch-budget.test.mjs
 * gates that arithmetic). Retrying by attempt count instead let identical
 * copies of this loop sleep 150s into a 120s section (2026-04-14, 2026-09-20).
 *
 * @param {string} url
 * @param {{ headers?: Record<string, string>, requestTimeoutMs: number, budgetMs: number, fetchFn?: typeof fetch, sleepFn?: (ms: number) => Promise<void>, now?: () => number }} options
 * @returns {Promise<Response>} the first OK response; any non-429 error status throws
 */
export async function fetchCoinGeckoWithRetryBudget(url, {
  headers = { Accept: 'application/json', 'User-Agent': CHROME_UA },
  requestTimeoutMs,
  budgetMs,
  fetchFn = fetch,
  sleepFn = sleep,
  now = Date.now,
}) {
  // An undefined budget would compare NaN and retry forever, which is the
  // failure this helper exists to rule out.
  for (const [name, value] of Object.entries({ requestTimeoutMs, budgetMs })) {
    if (!Number.isFinite(value) || value <= 0) throw new TypeError(`fetchCoinGeckoWithRetryBudget: ${name} must be a positive number, got ${value}`);
  }
  const startedAt = now();
  for (let attempt = 1; ; attempt++) {
    const resp = await fetchFn(url, { headers, signal: AbortSignal.timeout(requestTimeoutMs) });
    if (resp.status === 429) {
      const elapsed = now() - startedAt;
      const wait = Math.min(5_000 * 2 ** (attempt - 1), 60_000);
      if (elapsed + wait + requestTimeoutMs > budgetMs) {
        throw new Error(`CoinGecko rate limit exceeded after ${attempt} attempt(s) in ${Math.round(elapsed / 1000)}s (${budgetMs / 1000}s retry budget)`);
      }
      console.warn(`  CoinGecko 429 — waiting ${wait / 1000}s (attempt ${attempt}, ${Math.round(elapsed / 1000)}s of ${budgetMs / 1000}s budget)`);
      await sleepFn(wait);
      continue;
    }
    if (!resp.ok) throw new Error(`CoinGecko HTTP ${resp.status}`);
    return resp;
  }
}

/**
 * Unwrap fetch / network errors so log lines surface the actual cause
 * (DNS / TCP reset / TLS abort) instead of undici's bare "fetch failed".
 * Pulls `err.cause.code` (preferred — `ENOTFOUND`, `ECONNRESET`, etc.),
 * `err.cause.errno`, or `err.cause.message` in that order; falls back to
 * the outer error message when no cause is attached. Used by seeders
 * with multi-tier fallback chains (FATF, GDELT) where the failure mode
 * dictates the next-tier decision and operators need to distinguish
 * routing / DNS / handshake failures from per-host throttling.
 *
 * @param {unknown} err
 * @returns {string}
 */
export function describeErr(err) {
  if (!err) return 'unknown';
  const cause = err.cause;
  const causeCode = cause?.code || cause?.errno || cause?.message || (typeof cause === 'string' ? cause : null);
  return causeCode ? `${err.message} (cause: ${causeCode})` : (err.message || String(err));
}

/**
 * Fetch only configured CoinPaprika ticker IDs instead of the full /tickers
 * catalog. The catalog endpoint currently returns 13k+ records; seeders only
 * need a small mapped subset, so per-ID reads avoid Edge/cron memory and
 * latency spikes while preserving the same ticker shape.
 *
 * @param {string[]} paprikaIds CoinPaprika ids, e.g. btc-bitcoin
 * @param {{ fetchFn?: typeof fetch, headers?: Record<string, string>, timeoutMs?: number }} [options]
 * @returns {Promise<object[]>}
 */
export async function fetchCoinPaprikaTickersById(paprikaIds, options = {}) {
  const ids = [...new Set(paprikaIds.filter(Boolean))];
  if (ids.length === 0) return [];

  const fetchFn = options.fetchFn || fetch;
  const headers = { Accept: 'application/json', 'User-Agent': CHROME_UA, ...(options.headers || {}) };
  const timeoutMs = options.timeoutMs || 15_000;
  const concurrency = Math.max(1, Math.min(Number(options.concurrency || 4), ids.length));

  const results = await allSettledWithConcurrency(ids, concurrency, async (id) => {
    const resp = await fetchFn(`https://api.coinpaprika.com/v1/tickers/${encodeURIComponent(id)}?quotes=USD`, {
      headers,
      signal: AbortSignal.timeout(timeoutMs),
    });
    if (!resp.ok) throw new Error(`CoinPaprika ${id} HTTP ${resp.status}`);
    return resp.json();
  });

  const tickers = [];
  const failures = [];
  for (let i = 0; i < results.length; i += 1) {
    const result = results[i];
    if (result.status === 'fulfilled') {
      tickers.push(result.value);
    } else {
      failures.push(result.reason);
      console.warn(`[CoinPaprika] Skipping ${ids[i]}: ${describeErr(result.reason)}`);
    }
  }

  if (tickers.length === 0 && failures.length > 0) {
    throw new Error(`All ${failures.length} CoinPaprika ticker request(s) failed`);
  }

  return tickers;
}

export async function allSettledWithConcurrency(items, concurrency, mapper) {
  const results = new Array(items.length);
  let nextIndex = 0;
  // Clamp to a finite integer ≥1 so an invalid concurrency (0, NaN, negative, float)
  // can't spin up zero workers and silently return a sparse, all-unprocessed array —
  // it degrades to sequential instead. (items.length===0 → 0 workers → empty result.)
  const safe = Number.isFinite(concurrency) && concurrency >= 1 ? Math.floor(concurrency) : 1;
  const workerCount = Math.min(safe, items.length);

  await Promise.all(Array.from({ length: workerCount }, async () => {
    while (nextIndex < items.length) {
      const index = nextIndex;
      nextIndex += 1;
      try {
        results[index] = { status: 'fulfilled', value: await mapper(items[index], index) };
      } catch (reason) {
        results[index] = { status: 'rejected', reason };
      }
    }
  }));

  return results;
}

/**
 * Return the bundle-run start timestamp injected by `_bundle-runner.mjs`
 * as the `BUNDLE_RUN_STARTED_AT_MS` env var, or `null` when the seeder
 * is running STANDALONE (manual invocation outside the bundle).
 *
 * All sibling seeders in a single bundle run share ONE value (captured
 * at `runBundle` start, not at spawn time). Use this when a consumer
 * seeder reads a peer's output inside the same bundle and must detect
 * stale data from a previous bundle tick:
 *
 *   const bundleStartMs = getBundleRunStartedAtMs();
 *   if (bundleStartMs != null && fetchedAt < bundleStartMs) {
 *     // in-bundle context + peer did NOT run in THIS bundle → fallback
 *   }
 *
 * The null-on-unset contract matters. Earlier designs fell back to
 * `Date.now()` when the env was absent, which regressed standalone
 * runs: a sibling seeder invoked manually just before the consumer
 * wrote `fetchedAt = (process start - 5s)`, and the consumer's own
 * `bundleStartMs = Date.now()` rejected that perfectly-fresh peer
 * envelope as "stale". Returning null keeps the gate scoped to its
 * real purpose: protecting against across-bundle-tick staleness,
 * which has no analog outside a bundle context.
 *
 * @returns {number | null} epoch milliseconds when spawned by the
 *   bundle runner; null when running standalone.
 */
export function getBundleRunStartedAtMs() {
  const raw = Number(process.env.BUNDLE_RUN_STARTED_AT_MS);
  return Number.isFinite(raw) && raw > 0 ? raw : null;
}

// Point-of-use FX recovery constants — used by conversion seeds when Yahoo fails
// or a gap cannot be retried in budget. Canonical seed-fx-rates must NOT fill
// shared:fx-rates:v1 with these values (publish nulls + fallbackCurrencies instead).
// EGP: 0.0192 is the most recently observed live rate (2026-03-21 seed run).
export const SHARED_FX_FALLBACKS = {
  USD: 1, GBP: 1.2700, EUR: 1.0850, JPY: 0.0067, CHF: 1.1300,
  CNY: 0.1380, INR: 0.0120, AUD: 0.6500, CAD: 0.7400, NZD: 0.5900,
  BRL: 0.1900, MXN: 0.0490, ZAR: 0.0540, TRY: 0.0290, KRW: 0.0007,
  SGD: 0.7400, HKD: 0.1280, TWD: 0.0310, THB: 0.0280, IDR: 0.000063,
  NOK: 0.0920, SEK: 0.0930, DKK: 0.1450, PLN: 0.2450, CZK: 0.0430,
  HUF: 0.0028, RON: 0.2200, PHP: 0.0173, VND: 0.000040, MYR: 0.2250,
  PKR: 0.0036, ILS: 0.2750, ARS: 0.00084, COP: 0.000240, CLP: 0.00108,
  UAH: 0.0240, NGN: 0.00062, KES: 0.0077,
  // Fuel-price countries (#10051); XOF is pegged at 655.957 per EUR.
  ZMW: 0.0495, GHS: 0.0860, XOF: 0.00165, UYU: 0.0249,
  AED: 0.2723, SAR: 0.2666, QAR: 0.2747, KWD: 3.2520,
  BHD: 2.6525, OMR: 2.5974, JOD: 1.4104, EGP: 0.0192, LBP: 0.0000112,
};

export function loadSharedConfig(filename) {
  for (const base of [join(__seed_dirname, '..', 'shared'), join(__seed_dirname, 'shared')]) {
    const p = join(base, filename);
    if (existsSync(p)) return JSON.parse(readFileSync(p, 'utf8'));
  }
  throw new Error(`Cannot find shared/${filename} — checked ../shared/ and ./shared/`);
}

/**
 * Load the checkout's `.env.local` into process.env.
 *
 * `only` restricts the import to the named keys, for callers that want their
 * two credentials rather than every variable in the file. It exists so those
 * callers do not need a private loader — a private loader silently opts out of
 * the test-runtime guard and the checkout scoping, which is exactly how #5767
 * survived in two files.
 */
/**
 * Resolve the Convex HTTP-actions origin from an env bag.
 *
 * `CONVEX_URL` is the client/websocket origin (`*.convex.cloud`); HTTP actions
 * — every `/relay/*` route — are served from the sibling `*.convex.site`. The
 * mapping is a one-line string swap, but it is a one-line string swap that
 * three callers were each carrying their own copy of, and a caller that gets
 * it wrong silently POSTs to a host that does not route the request.
 * Returns '' when neither variable is set, so callers can report what is
 * missing rather than building a URL against `undefined`.
 *
 * @param {Record<string, string | undefined>} env
 */
export function resolveConvexSiteUrl(env) {
  const raw = env.CONVEX_SITE_URL || (env.CONVEX_URL ?? '').replace('.convex.cloud', '.convex.site');
  return raw ? raw.replace(/\/+$/, '') : '';
}

// The commit a seeder was deployed from: Railway, then Vercel, then Actions.
export function getDeployRevision(env = process.env) {
  return env.RAILWAY_GIT_COMMIT_SHA || env.VERCEL_GIT_COMMIT_SHA || env.GITHUB_SHA || '';
}

export function maskToken(token) {
  if (!token || token.length < 8) return '***';
  return token.slice(0, 4) + '***' + token.slice(-4);
}

export function getRedisCredentials() {
  const url = process.env.UPSTASH_REDIS_REST_URL;
  const token = process.env.UPSTASH_REDIS_REST_TOKEN;
  if (!url || !token) {
    console.error('Missing UPSTASH_REDIS_REST_URL or UPSTASH_REDIS_REST_TOKEN');
    process.exit(1);
  }
  return { url, token };
}

export async function parseRedisCommandResponse(resp, label = 'Redis command') {
  if (!resp.ok) {
    const text = typeof resp.text === 'function' ? await resp.text().catch(() => '') : '';
    const detail = text ? ` — ${text.slice(0, 200)}` : '';
    const err = new Error(`${label} failed: HTTP ${resp.status}${detail}`);
    // Tag errors so callers wrapping in withRetry know whether to back off.
    // Permanent 4xx (auth, payload-too-large, etc.) won't recover on retry —
    // mark non-retryable so withRetry exits the loop in ~10ms instead of
    // wasting backoff on a guaranteed-fail. Only 429 (rate-limited) should
    // keep retrying among the 4xx set, with the upstream Retry-After hint
    // honoured. Transient 5xx and timeouts fall through with no flag —
    // withRetry's default backoff applies.
    if (PERMANENT_4XX_STATUSES.has(resp.status)) {
      err.nonRetryable = true;
    } else if (resp.status === 429) {
      const retryAfterMs = parseRetryAfterUncappedMs(getResponseHeader(resp.headers, 'Retry-After'));
      if (retryAfterMs != null) {
        err.retryAfterMs = retryAfterMs;
        // A bounded Redis operation cannot wait out a longer cooldown.
        if (retryAfterMs > MAX_RETRY_AFTER_MS) err.nonRetryable = true;
      }
    }
    err.httpStatus = resp.status;
    throw err;
  }
  let body;
  try {
    body = await resp.json();
  } catch (cause) {
    throw Object.assign(new Error(`${label} returned invalid JSON (HTTP ${resp.status})`), { cause });
  }
  if (!body || typeof body !== 'object' || Array.isArray(body)) {
    throw new Error(`${label} returned an unexpected Upstash response`);
  }
  if (body.error != null) {
    throw new Error(`${label} rejected by Upstash: ${String(body.error)}`);
  }
  if (!Object.hasOwn(body, 'result')) {
    throw new Error(`${label} returned an Upstash response without a result`);
  }
  return body;
}

export async function redisCommand(url, token, command, options = {}) {
  const commandName = String(command?.[0] || 'command').toUpperCase();
  const label = options.label || `Redis ${commandName}`;
  const resp = await fetch(url, {
    method: 'POST',
    headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json', 'User-Agent': CHROME_UA },
    body: JSON.stringify(command),
    signal: AbortSignal.timeout(options.timeoutMs ?? SEED_REDIS_COMMAND_TIMEOUT_MS),
  });
  return parseRedisCommandResponse(resp, label);
}

async function redisGet(url, token, key, options = {}) {
  // Retry transient failures (timeout / network tear / 5xx / 429) with the
  // redisCommand tagging contract. A single unretried blip here silently read
  // as "key missing", which killed seed-gdelt-intel's cache-merge fallback for
  // 21h while the canonical key was healthy (issue #5437). The external
  // contract is unchanged: HTTP failures still degrade to null (now loudly),
  // thrown failures still propagate — both only after retries.
  //
  // `options.strict` opts a single caller out of the HTTP degrade. Degrading is
  // right for a cache-merge reader that can proceed without the value, and
  // wrong for one whose next step reads "no value" as a first run — the arms
  // sweep republished a 56-row slice over its ~200-row canonical key that way.
  // Default false keeps every existing caller byte-identical.
  let data;
  try {
    data = await withRetry(async () => {
      const resp = await fetch(`${url}/get/${encodeURIComponent(key)}`, {
        headers: { Authorization: `Bearer ${token}` },
        signal: AbortSignal.timeout(SEED_VERIFY_COMMAND_TIMEOUT_MS),
      });
      if (!resp.ok) {
        const err = new Error(`Redis GET ${key} failed: HTTP ${resp.status}`);
        if (PERMANENT_4XX_STATUSES.has(resp.status)) {
          err.nonRetryable = true;
        } else if (resp.status === 429) {
          const retryAfterMs = parseRetryAfterUncappedMs(getResponseHeader(resp.headers, 'Retry-After'));
          if (retryAfterMs != null) {
            err.retryAfterMs = retryAfterMs;
            // A bounded Redis operation cannot wait out a longer cooldown.
            if (retryAfterMs > MAX_RETRY_AFTER_MS) err.nonRetryable = true;
          }
        }
        err.httpStatus = resp.status;
        throw err;
      }
      return resp.json();
    }, SEED_REDIS_RETRY_ATTEMPTS - 1, SEED_REDIS_RETRY_BASE_MS);
  } catch (err) {
    if (err.httpStatus == null || options.strict) throw err;
    console.warn(`  Redis GET ${key}: degraded to null (${err.message})`);
    return null;
  }
  if (!data.result) return null;
  // Envelope-aware: returns inner `data` for seeded keys written in contract
  // mode, passes through legacy (bare-shape) values unchanged. Fixes WoW/cross-
  // seed reads that were silently getting `{_seed, data}` after PR 2a enveloped
  // the writer side of 91 canonical keys.
  return unwrapEnvelope(parseStoredSeedJson(data.result)).data;
}

async function redisSet(url, token, key, value, ttlSeconds) {
  const payload = JSON.stringify(value);
  const cmd = ttlSeconds
    ? ['SET', key, payload, 'EX', ttlSeconds]
    : ['SET', key, payload];
  return redisCommand(url, token, cmd);
}

async function redisDel(url, token, key) {
  return redisCommand(url, token, ['DEL', key]);
}

// Upstash REST calls surface transient network issues through fetch/undici
// errors rather than stable app-level error codes, so we normalize the common
// timeout/reset/DNS variants here before deciding to skip a seed run.
export function isTransientRedisError(err) {
  const message = String(err?.message || '');
  const causeMessage = String(err?.cause?.message || '');
  const code = String(err?.code || err?.cause?.code || '');
  const combined = `${message} ${causeMessage} ${code}`;
  return /UND_ERR_|Connect Timeout Error|fetch failed|ECONNRESET|ENOTFOUND|ETIMEDOUT|EAI_AGAIN/i.test(combined);
}

export async function acquireLock(domain, runId, ttlMs) {
  const { url, token } = getRedisCredentials();
  const lockKey = `seed-lock:${domain}`;
  const result = await redisCommand(url, token, ['SET', lockKey, runId, 'NX', 'PX', ttlMs]);
  return result?.result === 'OK';
}

export async function acquireLockSafely(domain, runId, ttlMs, opts = {}) {
  const label = opts.label || domain;
  try {
    const locked = await withRetry(
      () => acquireLock(domain, runId, ttlMs),
      opts.maxRetries ?? SEED_REDIS_RETRY_ATTEMPTS - 1,
      opts.delayMs ?? SEED_REDIS_RETRY_BASE_MS,
    );
    return { locked, skipped: false, reason: null };
  } catch (err) {
    if (isTransientRedisError(err)) {
      console.warn(`  SKIPPED: Redis unavailable during lock acquisition for ${label}`);
      return { locked: false, skipped: true, reason: 'redis_unavailable' };
    }
    throw err;
  }
}

export async function releaseLock(domain, runId) {
  const { url, token } = getRedisCredentials();
  const lockKey = `seed-lock:${domain}`;
  try {
    await redisCommand(url, token, ['EVAL', COMPARE_AND_DELETE_SCRIPT, 1, lockKey, runId]);
  } catch (err) {
    // Best-effort release; the lock still expires via TTL. Log the failure:
    // an empty catch hid a pinned-script mismatch until the TTL (#8490).
    const message = err instanceof Error ? err.message : String(err);
    console.warn(`  releaseLock failed for ${lockKey}: ${message}`);
  }
}

// ─── Seed lock fencing (SM-122) ───
// runSeed's fetch may run past lockTtlMs, so ownership is re-checked inside
// Redis at every publish point. Both scripts return {1, ''} when KEYS[1] still
// holds ARGV[1] (and renew it for ARGV[2] ms), or {0, <current holder or ''>}
// when the lock is gone, so the caller can tell a takeover from an expiry.
// The redis-rest proxy pins both texts byte for byte
// (docker/redis-rest-proxy.mjs, tests/redis-rest-proxy-command-parity.test.mjs).
export const SEED_LOCK_RENEW_SCRIPT = [
  "local current = redis.call('GET', KEYS[1])",
  'if current ~= ARGV[1] then',
  "  return {0, current or ''}",
  'end',
  "redis.call('PEXPIRE', KEYS[1], ARGV[2])",
  "return {1, ''}",
].join('\n');

// Fenced SET: writes KEYS[2] = ARGV[3] (with EX ARGV[4] unless it is '') only
// while KEYS[1] still holds ARGV[1].
export const SEED_LOCK_FENCED_SET_SCRIPT = [
  "local current = redis.call('GET', KEYS[1])",
  'if current ~= ARGV[1] then',
  "  return {0, current or ''}",
  'end',
  "redis.call('PEXPIRE', KEYS[1], ARGV[2])",
  "if ARGV[4] == '' then",
  "  redis.call('SET', KEYS[2], ARGV[3])",
  'else',
  "  redis.call('SET', KEYS[2], ARGV[3], 'EX', ARGV[4])",
  'end',
  "return {1, ''}",
].join('\n');

export class SeedLockLostError extends Error {
  constructor(lockKey, holder) {
    super(holder
      ? `seed lock ${lockKey} is now held by run ${holder}`
      : `seed lock ${lockKey} expired and is held by no run`);
    this.name = 'SeedLockLostError';
    this.lockKey = lockKey;
    this.holder = holder || null;
    // A lost lock never comes back by retrying the same write.
    this.nonRetryable = true;
  }
}

export function isSeedLockLostError(err) {
  return err instanceof SeedLockLostError;
}

/**
 * Throw SeedLockLostError unless an EVAL reply from one of the fence scripts
 * says the lock is still ours. Any other reply shape is a protocol error.
 */
function assertFenceReply(reply, lockKey) {
  const result = reply?.result;
  if (!Array.isArray(result) || (result[0] !== 1 && result[0] !== 0)) {
    throw new Error(`Seed lock fence for ${lockKey} returned an unexpected reply: ${JSON.stringify(result)}`);
  }
  if (result[0] === 0) throw new SeedLockLostError(lockKey, typeof result[1] === 'string' ? result[1] : '');
}

/**
 * Renew `seed-lock:<domain>` for ttlMs if runId still holds it. Throws
 * SeedLockLostError when it does not; Redis failures propagate after retries.
 */
export async function renewLockIfHeld(domain, runId, ttlMs) {
  const { url, token } = getRedisCredentials();
  const lockKey = `seed-lock:${domain}`;
  const reply = await withRetry(
    () => redisCommand(url, token, ['EVAL', SEED_LOCK_RENEW_SCRIPT, 1, lockKey, runId, ttlMs], {
      label: `Seed lock renew ${lockKey}`,
    }),
    SEED_REDIS_RETRY_ATTEMPTS - 1,
    SEED_REDIS_RETRY_BASE_MS,
  );
  assertFenceReply(reply, lockKey);
}

function fencedSetCommand(lockFence, key, payload, ttlSeconds) {
  return [
    'EVAL', SEED_LOCK_FENCED_SET_SCRIPT, 2, lockFence.lockKey, key,
    lockFence.runId, lockFence.ttlMs, payload, ttlSeconds ? ttlSeconds : '',
  ];
}

// The heartbeat of the lock runSeed currently holds. exitAfterTelemetryFlush
// stops it so a test harness that traps process.exit leaves no live timer.
let activeLockHeartbeat = null;

/**
 * Keep renewing a held seed lock every ttlMs/3 so a legitimately long fetch
 * does not lose it. Stops for good once Redis reports the lock gone; a
 * transient Redis error is logged and the next tick tries again.
 */
function startLockHeartbeat(domain, runId, ttlMs) {
  const state = { lost: null, timer: null, inFlight: false };
  const intervalMs = Math.max(250, Math.floor(ttlMs / 3));
  const stop = () => {
    if (state.timer) clearInterval(state.timer);
    state.timer = null;
    if (activeLockHeartbeat === heartbeat) activeLockHeartbeat = null;
  };
  const tick = async () => {
    if (state.inFlight || state.lost) return;
    state.inFlight = true;
    try {
      const { url, token } = getRedisCredentials();
      const lockKey = `seed-lock:${domain}`;
      const reply = await redisCommand(url, token, ['EVAL', SEED_LOCK_RENEW_SCRIPT, 1, lockKey, runId, ttlMs], {
        label: `Seed lock renew ${lockKey}`,
      });
      assertFenceReply(reply, lockKey);
    } catch (err) {
      if (isSeedLockLostError(err)) {
        state.lost = err;
        stop();
        console.warn(`  [${domain}] ${err.message} — this run will not publish`);
      } else {
        console.warn(`  [${domain}] seed lock renewal failed (will retry): ${err?.message || err}`);
      }
    } finally {
      state.inFlight = false;
    }
  };
  state.timer = setInterval(tick, intervalMs);
  state.timer.unref?.();
  const heartbeat = { state, stop };
  activeLockHeartbeat = heartbeat;
  return heartbeat;
}

export async function atomicPublish(canonicalKey, data, validateFn, ttlSeconds, options = {}) {
  const { url, token } = getRedisCredentials();

  if (validateFn) {
    const valid = validateFn(data);
    if (!valid) {
      return { payloadBytes: 0, skipped: true };
    }
  }

  // Some seeds publish a cohort of source-health records that must succeed
  // before the canonical archive becomes visible. Keep that work outside the
  // Redis retry loop: the callback's own writes own their retry policy, and a
  // failure must abort before the staging/canonical SET sequence begins.
  if (options.beforePublish) {
    await options.beforePublish(data);
  }

  // When the seeder opts into the contract (options.envelopeMeta provided), wrap
  // the payload in the seed envelope before publishing so the data key and its
  // freshness metadata share one lifecycle. Legacy seeders pass no envelopeMeta
  // and publish bare data, preserving pre-contract behavior. seed-meta:* keys
  // are always kept bare (shouldEnvelopeKey invariant).
  const payloadValue = options.envelopeMeta && shouldEnvelopeKey(canonicalKey)
    ? buildEnvelope({ ...options.envelopeMeta, data })
    : data;
  const json = JSON.stringify(payloadValue);
  // Readers stop inflating at MAX_DECODED_PAYLOAD_BYTES; never publish what they cannot read.
  if (options.compressPayload && Buffer.byteLength(json, 'utf8') > MAX_DECODED_PAYLOAD_BYTES) {
    throw new Error(`Payload too large: ${(Buffer.byteLength(json, 'utf8') / 1024 / 1024).toFixed(1)}MB decoded > ${MAX_DECODED_PAYLOAD_BYTES / 1024 / 1024}MB decode limit`);
  }
  const payload = options.compressPayload ? encodeGzipPayload(json) : json;
  const payloadBytes = Buffer.byteLength(payload, 'utf8');
  const maxPayloadBytes = options.maxPayloadBytes ?? MAX_PAYLOAD_BYTES;
  if (!Number.isSafeInteger(maxPayloadBytes) || maxPayloadBytes < 1 || maxPayloadBytes > MAX_SEEDED_VALUE_BYTES) {
    throw new Error(`Invalid maxPayloadBytes: expected an integer from 1 to ${MAX_SEEDED_VALUE_BYTES}`);
  }
  if (payloadBytes > maxPayloadBytes) {
    throw new Error(`Payload too large: ${(payloadBytes / 1024 / 1024).toFixed(1)}MB > ${maxPayloadBytes / 1024 / 1024}MB limit`);
  }

  // Retry the entire 3-call publish unit on transient Upstash failures.
  // Pre-fix: a single timeout on the canonical SET would crash the whole
  // seeder run and Railway just waited for the next cron tick (1h for
  // seed-forecasts). On 2026-05-10 this exact failure mode produced a
  // ~3h gap on forecasts + marketImplications. Wrapping the body means
  // a transient 5xx/timeout retries automatically with exponential
  // backoff. Permanent 4xx (auth, payload-too-large) get the
  // `nonRetryable: true` flag from redisCommand and abort immediately.
  //
  // Each attempt re-stages with a fresh runId so a previous attempt's
  // staging key (if it landed server-side but the response was lost)
  // doesn't shadow the retry. The 5-min staging TTL cleans up any
  // orphaned stagings naturally.
  return await withRetry(
    async () => {
      if (options.publishAtomically) {
        // A custom atomic publisher owns its Redis writes, so the fence can
        // only be checked (and the lease renewed) just before it runs.
        if (options.lockFence) await options.lockFence.renew();
        await options.publishAtomically({
          canonicalKey,
          payload,
          payloadValue,
          ttlSeconds,
        });
        return { payloadBytes, recordCount: Array.isArray(data) ? data.length : null };
      }

      const runId = `${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
      const stagingKey = `${canonicalKey}:staging:${runId}`;
      // With a lock fence, the canonical SET runs inside Redis only while this
      // run still holds the seed lock (SM-122).
      const canonicalCommand = options.lockFence
        ? fencedSetCommand(options.lockFence, canonicalKey, payload, ttlSeconds)
        : ttlSeconds
          ? ['SET', canonicalKey, payload, 'EX', ttlSeconds]
          : ['SET', canonicalKey, payload];
      if (maxPayloadBytes > MAX_PAYLOAD_BYTES) {
        const requestBytes = Math.max(
          Buffer.byteLength(JSON.stringify(canonicalCommand), 'utf8'),
          Buffer.byteLength(JSON.stringify(['SET', stagingKey, payload, 'EX', 300]), 'utf8'),
        );
        if (requestBytes > MAX_REDIS_REQUEST_BYTES) {
          const error = new Error(`Redis publish request too large: ${requestBytes} bytes > ${MAX_REDIS_REQUEST_BYTES}-byte limit`);
          error.nonRetryable = true;
          throw error;
        }
      }

      // Write to staging key
      await redisCommand(url, token, ['SET', stagingKey, payload, 'EX', 300]); // 5 min staging TTL

      // Overwrite canonical key
      const canonicalReply = await redisCommand(url, token, canonicalCommand);
      if (options.lockFence) {
        try {
          assertFenceReply(canonicalReply, options.lockFence.lockKey);
        } catch (err) {
          await redisDel(url, token, stagingKey).catch(() => {});
          throw err;
        }
      }

      // Cleanup staging
      await redisDel(url, token, stagingKey).catch(() => {});

      return { payloadBytes, recordCount: Array.isArray(data) ? data.length : null };
    },
    SEED_REDIS_RETRY_ATTEMPTS - 1,
    SEED_REDIS_RETRY_BASE_MS,
          // cumulative wait between attempts. Plus per-attempt fetch time
          // (15s timeout each) means total worst-case before propagating ≈ 48s.
  );
}

// Fields the shared seed-meta writer owns. afterPublish freshnessMetaPatch
// (and any preserve-on-skip merge) may add other diagnostics, but never
// re-anchor these — they come from the current runSeed outcome only.
export const FRESHNESS_META_RESERVED_FIELDS = Object.freeze([
  'fetchedAt',
  'recordCount',
  'sourceVersion',
  'newestItemAt',
  'oldestItemAt',
  'maxContentAgeMin',
]);

/**
 * Strip reserved freshness fields from an existing seed-meta object so the
 * remainder can be re-applied as a freshnessMetaPatch. Used when a validation
 * skip rewrites seed-meta while preserving last-good data — without this,
 * diagnostics such as prediction poolCounts are wiped and fail-closed health
 * false-alarms (#5875 review).
 */
export function freshnessMetaDiagnosticsPatch(existingMeta) {
  if (!existingMeta || typeof existingMeta !== 'object' || Array.isArray(existingMeta)) {
    return null;
  }
  const reserved = new Set(FRESHNESS_META_RESERVED_FIELDS);
  const patch = {};
  for (const [key, value] of Object.entries(existingMeta)) {
    if (!reserved.has(key)) patch[key] = value;
  }
  return Object.keys(patch).length > 0 ? patch : null;
}

/**
 * Best-effort read of the current seed-meta value for a domain/resource.
 * Returns the bare meta object, or null when missing / unreadable. Never throws.
 */
export async function readExistingSeedMeta(domain, resource) {
  try {
    const { url, token } = getRedisCredentials();
    const metaKey = `seed-meta:${domain}:${resource}`;
    const value = await redisGet(url, token, metaKey);
    if (!value || typeof value !== 'object' || Array.isArray(value)) return null;
    return value;
  } catch {
    return null;
  }
}

export async function writeFreshnessMetadata(
  domain,
  resource,
  count,
  source,
  ttlSeconds,
  fetchedAtOverride,
  contentAge,
  metaPatch,
  options = {},
) {
  const { url, token } = getRedisCredentials();
  const metaKey = `seed-meta:${domain}:${resource}`;
  const meta = {
    // Default to now; callers that want to mirror an existing canonical
    // envelope (validate-fail branch in runSeed) pass the canonical's
    // original fetchedAt so health doesn't lie about freshness — see
    // readCanonicalEnvelopeMeta() and the skipped-validate path below.
    fetchedAt: typeof fetchedAtOverride === 'number' ? fetchedAtOverride : Date.now(),
    recordCount: count,
    sourceVersion: source || '',
  };
  // Content-age trio (2026-05-04 health-readiness plan). Pass when the seeder
  // opted in. Presence of maxContentAgeMin is the opt-in signal that the
  // health classifier reads. newestItemAt/oldestItemAt may be explicit null
  // when contentMeta returned null — classifier reads as STALE_CONTENT.
  if (contentAge && typeof contentAge === 'object' && Number.isInteger(contentAge.maxContentAgeMin)) {
    meta.newestItemAt = contentAge.newestItemAt ?? null;
    meta.oldestItemAt = contentAge.oldestItemAt ?? null;
    meta.maxContentAgeMin = contentAge.maxContentAgeMin;
  }
  if (metaPatch && typeof metaPatch === 'object') {
    // Hooks may add bounded diagnostics, but the shared writer owns freshness
    // and record-count truth. Never let a hook silently re-anchor those fields.
    const reservedFields = new Set(FRESHNESS_META_RESERVED_FIELDS);
    for (const [key, value] of Object.entries(metaPatch)) {
      if (!reservedFields.has(key)) meta[key] = value;
    }
  }
  // Use the data TTL if it exceeds 7 days so monthly/annual seeds don't lose
  // their meta key before the health check maxStaleMin threshold is reached.
  const metaTtl = resolveSeedMetaTtl(undefined, ttlSeconds);
  // Retry transient Redis failures: this SET runs bare on runSeed's
  // validate-skip path, where an unretried Upstash abort escaped to the
  // seeder's top-level catch as `FATAL: The operation was aborted due to
  // timeout` → exit 1 (seed-gdelt-intel, issue #5437). redisCommand tags
  // permanent 4xx nonRetryable and 429 with Retry-After; withRetry honors both.
  //
  // options.lockFence (runSeed only) makes the SET conditional on the run
  // still holding its seed lock, so a run that lost the lock cannot stamp
  // fetchedAt=now over the newer owner's seed-meta (SM-122).
  const lockFence = options?.lockFence;
  await withRetry(
    async () => {
      if (!lockFence) return redisSet(url, token, metaKey, meta, metaTtl);
      const reply = await redisCommand(
        url,
        token,
        fencedSetCommand(lockFence, metaKey, JSON.stringify(meta), metaTtl),
        { label: `seed-meta ${metaKey}` },
      );
      assertFenceReply(reply, lockFence.lockKey);
      return reply;
    },
    SEED_REDIS_RETRY_ATTEMPTS - 1,
    SEED_REDIS_RETRY_BASE_MS,
  );
  return meta;
}

/**
 * writeFreshnessMetadata for runSeed's OWN bookkeeping call sites: degrades to
 * null (loudly) instead of throwing when Redis stays down past the retry
 * budget. By the time runSeed writes seed-meta, the run's outcome is already
 * decided — the canonical publish succeeded, or the skip path preserved
 * last-good — so letting a metadata SET escape as a throw converts a Redis
 * blip into `FATAL: … exit 1` + a "Deploy Crashed!" alert over pure
 * bookkeeping (seed-gdelt-intel 2026-07-23, issue #5478; the #5438 retry
 * alone was insufficient under the sustained brownout contention window).
 * The degraded write leaves the OLD seed-meta in place, which ages naturally
 * — /api/health STALE_SEED is the durable alarm for a persistent failure.
 *
 * External callers keep using writeFreshnessMetadata directly: its
 * throw-after-retries contract is intentional and pinned by tests.
 */
export async function writeFreshnessMetadataSafely(domain, resource, ...rest) {
  try {
    return await writeFreshnessMetadata(domain, resource, ...rest);
  } catch (err) {
    // Losing the seed lock is an ownership decision, not a bookkeeping blip.
    if (isSeedLockLostError(err)) throw err;
    console.warn(
      `  WARNING: seed-meta write for ${domain}:${resource} failed after retries (${err?.message || err}) — `
      + `continuing; seed-meta will age and /api/health STALE_SEED is the alarm if this persists`,
    );
    return null;
  }
}

/**
 * Read the canonical key's contract-mode envelope meta. Used by runSeed's
 * validate-fail branch to mirror canonical state into seed-meta instead
 * of overwriting it with recordCount=0 (which makes /api/health report
 * EMPTY_DATA when the canonical key still holds last-good data — see
 * PR #3581 for the production incident).
 *
 * Returns the {fetchedAt, recordCount, sourceVersion} block when canonicalKey
 * is contract-mode (envelope dual-write) AND has a valid recordCount > 0.
 * Returns null for legacy (bare-shape) keys, missing keys, parse errors,
 * or zero envelopes — caller falls back to its existing default behavior.
 *
 * Defensive: any read/parse error → null. No throws bubble up.
 */
export async function readCanonicalEnvelopeMeta(canonicalKey) {
  try {
    const { url, token } = getRedisCredentials();
    // Retry transient failures before degrading: a blip here is worse than a
    // crash — the caller falls back to writing recordCount=0 with
    // fetchedAt=NOW, resetting the freshness clock over real staleness
    // (issue #5437). Outer catch preserves the never-throws contract.
    const data = await withRetry(async () => {
      const resp = await fetch(`${url}/get/${encodeURIComponent(canonicalKey)}`, {
        headers: { Authorization: `Bearer ${token}` },
        signal: AbortSignal.timeout(5_000),
      });
      if (!resp.ok) {
        const err = new Error(`Canonical meta GET ${canonicalKey} failed: HTTP ${resp.status}`);
        if (PERMANENT_4XX_STATUSES.has(resp.status)) {
          err.nonRetryable = true;
        } else if (resp.status === 429) {
          const retryAfterMs = parseRetryAfterUncappedMs(getResponseHeader(resp.headers, 'Retry-After'));
          if (retryAfterMs != null) {
            err.retryAfterMs = retryAfterMs;
            // A bounded Redis operation cannot wait out a longer cooldown.
            if (retryAfterMs > MAX_RETRY_AFTER_MS) err.nonRetryable = true;
          }
        }
        throw err;
      }
      return resp.json();
    }, 2, 1000);
    if (!data || !data.result) return null;
    let parsed;
    try { parsed = parseStoredSeedJson(data.result); } catch { return null; }
    if (!parsed || typeof parsed !== 'object') return null;
    const seed = parsed._seed;
    if (!seed || typeof seed !== 'object') return null;
    if (typeof seed.fetchedAt !== 'number' || typeof seed.recordCount !== 'number') return null;
    if (seed.recordCount <= 0) return null;
    // Content-age fields propagate through the validate-fail mirror so the
    // health classifier doesn't lose the STALE_CONTENT signal exactly when
    // last-good-with-stale-content data is being served (Codex round 1 P0b).
    // All three fields are optional in the envelope; carry them through as a
    // trio when present, otherwise undefined (caller checks).
    const contentAge = (typeof seed.maxContentAgeMin === 'number')
      ? {
          newestItemAt: typeof seed.newestItemAt === 'number' ? seed.newestItemAt : null,
          oldestItemAt: typeof seed.oldestItemAt === 'number' ? seed.oldestItemAt : null,
          maxContentAgeMin: seed.maxContentAgeMin,
        }
      : undefined;
    return {
      fetchedAt: seed.fetchedAt,
      recordCount: seed.recordCount,
      sourceVersion: typeof seed.sourceVersion === 'string' ? seed.sourceVersion : '',
      contentAge,
    };
  } catch {
    return null;
  }
}

// HTTP statuses where retrying CAN'T succeed because the failure is in the
// caller's request shape, not server load:
//   400 malformed query   401 bad auth      403 forbidden
//   404 missing path      410 permanently gone
//   413 payload too large 422 semantic error    451 legal block
// 408 Request Timeout and 429 Too Many Requests are deliberately excluded —
// both are explicit "back off and retry" signals, often paired with a
// Retry-After header. Tagging them nonRetryable would convert transient
// rate-limits into immediate seed failures (especially under parallel
// fetches like seed-imf-* WEO bundles).
export const PERMANENT_4XX_STATUSES = new Set([400, 401, 403, 404, 410, 413, 422, 451]);

// sysexits.h EX_TEMPFAIL: fetch failed, last-good TTL was extended, and the
// bundle runner should retry/report non-OK without treating the seeder as a
// generic crash.
export const GRACEFUL_FETCH_FAILURE_EXIT_CODE = 75;

/**
 * Tag a fetch error as an upstream outage the seeder may ride out on last-good
 * data. Only tagged errors are eligible for runSeed's
 * allowPreservedFetchFailure exit-0 path; configuration errors and code bugs
 * stay untagged and keep exiting 75 so Railway still surfaces them.
 *
 * @param {Error} err
 * @returns {Error}
 */
export function markPreservableFetchFailure(err) {
  err.preservableFetchFailure = true;
  return err;
}

// Age in minutes of the retained seed-meta, or null when it cannot be read or
// trusted. Callers treat null as "not fresh" so an unreadable meta fails closed.
async function readSeedMetaAgeMin(metaKey) {
  try {
    const meta = await readCanonicalValue(metaKey);
    const fetchedAt = meta?.fetchedAt;
    if (typeof fetchedAt !== 'number' || !Number.isFinite(fetchedAt)) return null;
    const ageMin = (Date.now() - fetchedAt) / 60_000;
    return ageMin >= 0 ? ageMin : null;
  } catch {
    return null;
  }
}

// #6396: the seeder fetched its data but its coverage gate refused to publish
// (and preserved the last-good TTL instead). Distinct from EX_TEMPFAIL so the
// bundle runner can report PUBLISH_BLOCKED rather than OK for a section whose
// entire purpose — writing the seed keys — did not happen.
export const PUBLISH_BLOCKED_EXIT_CODE = 76;

// Cap upstream Retry-After hints so a stuck/abusive header can't park the
// bundle past its section timeoutMs. Mirrors _yahoo-fetch.mjs convention.
const MAX_RETRY_AFTER_MS = 60_000;

/**
 * Parse `Retry-After` header value (seconds OR HTTP-date). Returns null
 * when missing/unparseable so callers can fall back to default backoff.
 * Duplicates exist in _yahoo-fetch / _gdelt-fetch / _open-meteo-archive —
 * those predate this helper; consolidating them is a separate refactor.
 */
export function parseRetryAfterMs(value) {
  const parsed = parseRetryAfterUncappedMs(value);
  return parsed == null ? null : Math.min(parsed, MAX_RETRY_AFTER_MS);
}

/**
 * #6110: the same parse WITHOUT the `MAX_RETRY_AFTER_MS` cap.
 *
 * The cap exists so a stuck header cannot park a bundle past its timeout — it
 * bounds how long we SLEEP. But it also erases how far out the server actually
 * pushed us, and that magnitude is exactly what tells us a retry is pointless:
 * a provider's daily-quota 429 asked for 1213s, which the cap flattens to 60s. Judging
 * futility on the capped value silently reinstates the bug for any caller whose
 * remaining budget is >= 60s.
 *
 * So: sleep on the capped value, judge on the uncapped one.
 */
function parseRetryAfterUncappedMs(value) {
  if (!value) return null;
  const seconds = Number(value);
  if (Number.isFinite(seconds) && seconds > 0) return seconds * 1000;
  const retryAt = Date.parse(value);
  if (Number.isFinite(retryAt)) return Math.max(retryAt - Date.now(), 1000);
  return null;
}

export async function withRetry(fn, maxRetries = 3, delayMs = 1000) {
  if (!Number.isInteger(maxRetries) || maxRetries < 0 || maxRetries > 10
      || !Number.isFinite(delayMs) || delayMs < 0) {
    throw new RangeError('withRetry requires 0-10 retries and a finite non-negative delay');
  }
  let lastErr;
  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      return await fn();
    } catch (err) {
      lastErr = err;
      // Permanent failures (4xx auth/permission, missing config) burn the
      // bundle timeoutMs if retried. Callers tag with `nonRetryable: true`
      // so the loop fails in ~10ms instead of waiting through every backoff.
      if (err?.nonRetryable) throw err;
      if (attempt < maxRetries) {
        // Honor upstream `Retry-After` hint when caller attached it
        // (typically 429 / 503). Take whichever is longer — the upstream
        // hint OR the exponential backoff — so a generous server hint
        // isn't undercut, and a missing/short hint still gets back-off.
        const capMs = 60_000;
        const ceiling = Math.min(capMs, delayMs * 2 ** attempt * 1.25);
        // Keep the existing minimum delay below the cap, then spread workers
        // across the upper 20% of the capped exponential window.
        const baseWait = Math.round(ceiling * (0.8 + Math.random() * 0.2));
        const retryAfterMs = Number(err?.retryAfterMs);
        // Provider cooldowns can exceed the backoff cap; callers with phase
        // budgets reject hints that cannot fit before starting another attempt.
        const wait = Number.isFinite(retryAfterMs) ? Math.max(baseWait, retryAfterMs) : baseWait;
        const cause = err.cause ? ` (cause: ${err.cause.message || err.cause.code || err.cause})` : '';
        console.warn(`  Retry ${attempt + 1}/${maxRetries} in ${wait}ms: ${err.message || err}${cause}`);
        // Test-only cap on the idle wait between attempts (WM_SEED_RETRY_DELAY_MS):
        // suites that stub persistent upstream failures otherwise sleep through
        // every real backoff. Attempt COUNT and the logged/computed wait stay
        // real — only the sleep shrinks. Dual-gated on NODE_TEST_CONTEXT so the
        // knob is structurally inert outside the node test runner: a stray env
        // var on a Railway seeder must not disable production backoff (and the
        // full-wait log lines would make that near-invisible). Read per call.
        const overrideMs = process.env.NODE_TEST_CONTEXT
          ? Number(process.env.WM_SEED_RETRY_DELAY_MS)
          : Number.NaN;
        const pause = Number.isFinite(overrideMs) ? Math.min(wait, overrideMs) : wait;
        await new Promise(r => setTimeout(r, pause));
      }
    }
  }
  throw lastErr;
}

/**
 * Read a header from either a fetch `Headers` instance or a plain object
 * (test transports commonly pass `{ 'retry-after': '2' }`). Case-insensitive.
 */
export function getResponseHeader(headers, name) {
  if (!headers) return null;
  if (typeof headers.get === 'function') return headers.get(name);
  const lowerName = name.toLowerCase();
  const foundKey = Object.keys(headers).find((key) => key.toLowerCase() === lowerName);
  return foundKey ? headers[foundKey] : null;
}

/** 408 / 429 / 5xx are transient; every other status is a permanent client error. */
export function isRetryableHttpStatus(status) {
  return status === 408 || status === 429 || (typeof status === 'number' && status >= 500 && status <= 599);
}

/**
 * Build an Error from a non-ok provider response for use with `withRetry`.
 * Tags `nonRetryable` for permanent statuses and attaches a capped
 * `retryAfterMs` hint when the server sent one.
 *
 * Three knobs, and the distinction between them is the whole point:
 *   - `maxRetryAfterMs` — a policy CEILING ("never sleep longer than this").
 *     Clamping is legitimate: the server's hint may be conservative, so an
 *     earlier retry can still succeed.
 *   - `capMs` — also a ceiling, plus "when <= 0 there is no time left at all,
 *     so stop". Kept exactly as-is: scripts/_seed-history.mjs passes a fixed
 *     `RELAY_RETRY_AFTER_CAP_MS` here, so it is NOT a remaining-budget signal.
 *   - `remainingBudgetMs` (#6110) — the wall clock the caller actually has
 *     left. This one can produce a VERDICT, not just a clamp: if the server's
 *     own hint meets or exceeds it, no retry inside this run can succeed, so
 *     the error is nonRetryable and the caller falls through immediately with
 *     its budget intact instead of sleeping against a wall that cannot move.
 *     Equality is futile too: sleeping the full remainder leaves usableBudget
 *     at 0, so the next withRetry attempt throws createLlmBudgetError and
 *     aborts the whole provider waterfall rather than failing over.
 *
 * Why the verdict matters — production, seed-insights 2026-08-03 12:10Z/12:20Z:
 * a provider answered 429 with "tokens per day (TPD): Limit 100000, Used 100000 …
 * try again in 20m13.92s". That 1213s hint was clamped to the 10s ceiling and
 * retried twice, spending 20s of a 60s LLM budget (and of a 120s seed lock) on
 * a daily quota that could not reset for another 20 minutes. Those cycles ran
 * 30-36s against 7-17s for healthy ones, and the run still ended with nothing.
 */
export function httpRetryError(resp, { maxRetryAfterMs, capMs, remainingBudgetMs } = {}) {
  const status = resp?.status;
  const err = new Error(`HTTP ${status}`);
  err.status = status;
  err.nonRetryable = !isRetryableHttpStatus(status);
  const rawHeader = getResponseHeader(resp?.headers, 'Retry-After');
  const uncappedRetryAfterMs = parseRetryAfterUncappedMs(rawHeader);
  let retryAfterMs = parseRetryAfterMs(rawHeader);
  if (retryAfterMs != null) {
    // #6110: judge futility on the UNCAPPED hint. Every ceiling in play here —
    // MAX_RETRY_AFTER_MS at parse time, then maxRetryAfterMs and capMs below —
    // answers "how long may we sleep", never "is sleeping worth anything". Only
    // the uncapped hint carries the magnitude that settles that, and comparing
    // the capped value instead would reinstate this very bug for any caller
    // whose budget is >= MAX_RETRY_AFTER_MS (a 1213s hint reads as 60s there).
    // `>=`: equality is futile for waterfall callers. Sleeping a hint that
    // equals the remaining budget spends the whole remainder; the next
    // withRetry attempt hits usableBudgetMs() <= 0 → createLlmBudgetError and
    // aborts every later provider. Fail-fast keeps the budget for fallthrough.
    if (Number.isFinite(remainingBudgetMs) && uncappedRetryAfterMs >= Math.max(0, remainingBudgetMs)) {
      err.nonRetryable = true;
      // Keep the UNCAPPED hint even though we will not sleep on it: only the
      // raw magnitude separates "quota exhausted for 20 minutes" from
      // "throttled for 2 seconds" in the log. The sleep path below still uses
      // the capped parse. withRetry checks nonRetryable before ever reading
      // retryAfterMs, so attaching the uncapped value here cannot cause a sleep.
      err.retryAfterMs = uncappedRetryAfterMs;
      return err;
    }
    if (Number.isFinite(maxRetryAfterMs)) retryAfterMs = Math.min(retryAfterMs, maxRetryAfterMs);
    if (Number.isFinite(capMs)) retryAfterMs = Math.min(retryAfterMs, Math.max(0, capMs));
    // No `remainingBudgetMs` clamp here on purpose: the early return above
    // already guarantees hint < budget, and the ceilings only shrink it
    // further. Adding one would be dead code that reads like a safeguard.
    if (retryAfterMs > 0) err.retryAfterMs = retryAfterMs;
    else err.nonRetryable = true;
  }
  return err;
}

/**
 * Sentinel error for "the LLM call's time budget is spent". `nonRetryable`
 * stops `withRetry` immediately; `llmBudgetExhausted` lets the provider loop
 * distinguish a budget stop (give up, ship degraded) from a provider error
 * (fall through to the next provider).
 */
export function createLlmBudgetError(message = 'llm time budget exhausted') {
  const err = new Error(message);
  err.nonRetryable = true;
  err.llmBudgetExhausted = true;
  return err;
}

export function isLlmBudgetError(err) {
  return Boolean(err?.llmBudgetExhausted);
}

export function logSeedResult(domain, count, durationMs, extra = {}) {
  console.log(JSON.stringify({
    event: 'seed_complete',
    domain,
    recordCount: count,
    durationMs: Math.round(durationMs),
    timestamp: new Date().toISOString(),
    ...extra,
  }));
}

/**
 * Shared envelope-aware reader for cross-seed consumers (e.g. seed-forecasts
 * reading ~40 migrated input keys, seed-chokepoint-flows reading portwatch,
 * seed-thermal-escalation reading wildfire:fires). Returns the inner `data`
 * payload for contract-mode writes; passes legacy bare-shape values through
 * unchanged. Callers MUST NOT parse the envelope themselves.
 */
export async function readCanonicalValue(key, options = {}) {
  const { url, token } = getRedisCredentials();
  return redisGet(url, token, key, options);
}

export async function verifySeedKey(key) {
  // redisGet() now unwraps envelopes internally, so callers that read migrated
  // canonical keys (e.g. seed-climate-anomalies reading climate:zone-normals:v1,
  // seed-thermal-escalation reading wildfire:fires:v1) see bare legacy-shape
  // payloads regardless of whether the writer has migrated to contract mode.
  const { url, token } = getRedisCredentials();
  return redisGet(url, token, key);
}

/**
 * Invariant: `seed-meta:*` keys MUST be bare-shape `{fetchedAt, recordCount, ...}`.
 * Health + bundle runner + every legacy reader parses them as top-level.
 * Enveloping them turns every downstream read into `{_seed, data}` which breaks
 * the whole freshness-registry flow. Enforced at the helper boundary so future
 * callers can't regress this by passing an envelopeMeta that happens to target
 * a seed-meta key (seed-iea-oil-stocks' ANALYSIS_META_EXTRA_KEY did exactly that).
 */
export function shouldEnvelopeKey(key) {
  return typeof key === 'string' && !key.startsWith('seed-meta:');
}

export async function writeExtraKey(key, data, ttl, envelopeMeta, options = {}) {
  const { url, token } = getRedisCredentials();
  const payload = serializeExtraKeyValue(key, data, envelopeMeta);
  // options.lockFence (runSeed only): write only while the run still holds
  // its seed lock (SM-122).
  const lockFence = options?.lockFence;
  // Retry transient Redis timeouts / 5xx / network tears so a single Upstash
  // blip in afterPublish doesn't crash the whole seeder run. Permanent 4xx
  // (auth, payload-too-large) fail fast; 429 honors Retry-After. Mirrors the
  // redisCommand / atomicPublish contract (seed-gdelt-intel PUBLISH_TIMEOUT fix).
  await withRetry(async () => {
    const reply = await redisCommand(
      url,
      token,
      lockFence ? fencedSetCommand(lockFence, key, payload, ttl) : ['SET', key, payload, 'EX', ttl],
      { label: `Extra key ${key}`, timeoutMs: SEED_EXTRA_KEY_COMMAND_TIMEOUT_MS },
    );
    if (lockFence) assertFenceReply(reply, lockFence.lockKey);
  }, SEED_REDIS_RETRY_ATTEMPTS - 1, SEED_REDIS_RETRY_BASE_MS);
  console.log(`  Extra key ${key}: written`);
}

/** Serialize an extra key exactly as it is persisted to Redis. */
export function serializeExtraKeyValue(key, data, envelopeMeta) {
  const value = envelopeMeta && shouldEnvelopeKey(key) ? buildEnvelope({ ...envelopeMeta, data }) : data;
  return JSON.stringify(value);
}

/** Return the UTF-8 size of an extra-key value as Redis receives it. */
export function extraKeyPayloadBytes(key, data, envelopeMeta) {
  return Buffer.byteLength(serializeExtraKeyValue(key, data, envelopeMeta), 'utf8');
}

/**
 * Floor for every seed-meta TTL. A meta key must survive its data key's
 * disappearance so health can report STALE_SEED (present-but-stale) rather than
 * losing the heartbeat at the same moment as the payload — see the "seed-meta
 * outlives its data key" note in api/health.js's absence branch.
 */
export const SEED_META_MIN_TTL_SECONDS = 86400 * 7;

/**
 * The meta TTL for a data key written with `dataTtlSeconds`.
 *
 * The floor alone is not enough once a data key outlives 7 days: health reads
 * freshness from seed-meta and falls through to plain OK when the meta is gone
 * but the data key still has bytes, so a meta that expires FIRST makes the
 * STALE_SEED alarm unreachable for the remainder of the data key's life. The
 * clamp is the same one `writeFreshnessMetadata` has always applied to the
 * canonical key; extra keys need it for the same reason.
 *
 * An explicit `metaTtlSeconds` still wins, so the parameter keeps meaning what
 * it says. The three seeders that already pass one (seed-jodi-gas,
 * seed-natural-events, seed-defense-industrial-suppliers) pass their own data
 * TTL — the value this would have computed — so they are byte-identical either
 * way; the override exists for a future caller that needs a different one.
 */
export function resolveSeedMetaTtl(metaTtlSeconds, dataTtlSeconds) {
  return metaTtlSeconds ?? Math.max(SEED_META_MIN_TTL_SECONDS, dataTtlSeconds || 0);
}

export const SEED_META_KEY_PREFIX = 'seed-meta:';

/**
 * Resolve the seed-meta key for a data key. With no override the meta key is
 * derived (`seed-meta:<dataKey minus :vN>`); an override must be a DISTINCT key
 * inside the `seed-meta:` namespace.
 *
 * #8424: seed-bls-series passed its own data key as the override, so every run
 * overwrote the series it had just written with the 44-byte heartbeat, on the
 * 7-day meta TTL, while health — watching the canonical key — read OK for six
 * months. A wrong override now fails the run before any byte is written
 * instead of silently erasing the payload it describes.
 */
export function resolveSeedMetaKey(dataKey, metaKeyOverride) {
  if (metaKeyOverride === undefined || metaKeyOverride === null || metaKeyOverride === '') {
    return `${SEED_META_KEY_PREFIX}${dataKey.replace(/:v\d+$/, '')}`;
  }
  if (typeof metaKeyOverride !== 'string' || !metaKeyOverride.startsWith(SEED_META_KEY_PREFIX) || metaKeyOverride === dataKey) {
    throw new Error(
      `seed-meta key for ${dataKey} must be a distinct ${SEED_META_KEY_PREFIX}* key, got ${String(metaKeyOverride)} `
      + '(a colliding override overwrites the data it describes, #8424)',
    );
  }
  return metaKeyOverride;
}

function buildSeedMeta(recordCount, coverage, extra, fetchedAt = Date.now()) {
  const meta = { fetchedAt, recordCount: recordCount ?? 0 };
  if (coverage) meta.coverage = coverage;
  // Optional producer diagnostics, copied verbatim onto the meta record.
  // api/health.js decides which fields it trusts (see readSeedMeta), so callers
  // must keep `extra` safe for a public endpoint; the consumer-prices
  // `coverage` block is deliberately NOT merged here (health parses it with a
  // separate retailer schema — see fetchTradeFlows in seed-supply-chain-trade).
  if (extra && typeof extra === 'object') {
    for (const [key, value] of Object.entries(extra)) {
      if (value !== undefined) meta[key] = value;
    }
  }
  return meta;
}

export async function writeSeedMeta(dataKey, recordCount, metaKeyOverride, metaTtlSeconds, coverage, extra, options = {}) {
  const { url, token } = getRedisCredentials();
  const metaKey = resolveSeedMetaKey(dataKey, metaKeyOverride);
  const meta = buildSeedMeta(recordCount, coverage, extra);
  // No data TTL is in scope here — callers that know one resolve it through
  // `resolveSeedMetaTtl` before calling. Bare floor otherwise.
  const metaTtl = resolveSeedMetaTtl(metaTtlSeconds);
  const resp = await fetch(url, {
    method: 'POST',
    headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json', 'User-Agent': CHROME_UA },
    // options.lockFence (runSeed only): write only while the run still holds
    // its seed lock (SM-122).
    body: JSON.stringify(options?.lockFence
      ? fencedSetCommand(options.lockFence, metaKey, JSON.stringify(meta), metaTtl)
      : ['SET', metaKey, JSON.stringify(meta), 'EX', metaTtl]),
    signal: AbortSignal.timeout(5_000),
  });
  if (!resp.ok) {
    console.warn(`  seed-meta ${metaKey}: write failed`);
    return false;
  }
  // Upstash can return HTTP 200 with a command-level error. Treat that as a
  // failed metadata write instead of reporting success while health still
  // points at the previous heartbeat.
  const reply = await parseRedisCommandResponse(resp, `seed-meta ${metaKey}`);
  if (options?.lockFence) assertFenceReply(reply, options.lockFence.lockKey);
  return true;
}

export async function writeExtraKeyWithMeta(key, data, ttl, recordCount, metaKeyOverride, metaTtlSeconds, coverage, extra) {
  // Resolve (and reject) the meta key BEFORE the data write: a colliding pair
  // must fail the run with the previous value intact, not after erasing it.
  const metaKey = resolveSeedMetaKey(key, metaKeyOverride);
  await writeExtraKey(key, data, ttl);
  // The data TTL is right here, so the meta never has to be the shorter of the
  // two. seed-economy's four EIA weekly keys (21d data, 14d health budget) rode
  // the bare 7d default and went silent-OK for the 14 days in between.
  // `extra` carries the same optional producer diagnostics writeSeedMeta accepts
  // directly (see its contract note) — provenance a caller needs on the meta
  // record, not just inside the data payload.
  return writeSeedMeta(key, recordCount, metaKey, resolveSeedMetaTtl(metaTtlSeconds, ttl), coverage, extra);
}

// Some aggregate keys are both the data pointer and the provenance source for
// health. Publish that pair in one Redis transaction so readers cannot observe
// a new marker with the previous seed-meta record.
export async function writeExtraKeyWithMetaAtomically({
  key,
  data,
  ttlSeconds,
  recordCount,
  metaKey: metaKeyOverride,
  metaTtlSeconds,
  coverage,
  extra,
  fetchedAt = Date.now(),
}) {
  const { url, token } = getRedisCredentials();
  const dataTtl = Number(ttlSeconds);
  const metaTtl = Number(resolveSeedMetaTtl(metaTtlSeconds, dataTtl));
  if (!key || !Number.isInteger(dataTtl) || dataTtl <= 0) {
    throw new Error('Atomic extra-key publish requires a key and a positive integer TTL');
  }
  if (!Number.isInteger(metaTtl) || metaTtl <= 0) {
    throw new Error('Atomic seed-meta publish requires a positive integer TTL');
  }

  const metaKey = resolveSeedMetaKey(key, metaKeyOverride);
  const commands = [
    ['SET', key, JSON.stringify(data), 'EX', dataTtl],
    ['SET', metaKey, JSON.stringify(buildSeedMeta(recordCount, coverage, extra, fetchedAt)), 'EX', metaTtl],
  ];
  // This runs after the provider fetches have settled. Retrying this bounded
  // Redis transaction therefore recovers a transient publication failure
  // without replaying the provider requests or exposing half the pair.
  return withRetry(async () => {
    const resp = await fetch(`${url}/multi-exec`, {
      method: 'POST',
      headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json', 'User-Agent': CHROME_UA },
      body: JSON.stringify(commands),
      signal: AbortSignal.timeout(5_000),
    });
    if (!resp.ok) {
      const err = httpRetryError(resp);
      err.message = `Atomic extra-key publish failed: HTTP ${resp.status}`;
      err.httpStatus = resp.status;
      throw err;
    }

    let results;
    try {
      results = await resp.json();
    } catch (cause) {
      throw Object.assign(new Error('Atomic extra-key publish failed: invalid transaction response'), {
        cause,
        nonRetryable: true,
      });
    }
    if (!Array.isArray(results)) {
      throw Object.assign(
        new Error(`Atomic extra-key publish failed: ${results?.error || 'invalid transaction response'}`),
        { nonRetryable: true },
      );
    }
    const failures = results.filter((result) => result?.error || result?.result === 'ERR');
    if (failures.length > 0 || results.length !== commands.length) {
      throw Object.assign(
        new Error(`Atomic extra-key publish failed: ${failures.length || 'missing'} command result(s)`),
        { nonRetryable: true },
      );
    }
    return true;
  }, SEED_REDIS_RETRY_ATTEMPTS - 1, SEED_REDIS_RETRY_BASE_MS);
}

// Detailed counterpart to extendExistingTtl. Results stay aligned to the input
// keys so callers that publish per-key health can distinguish a confirmed
// EXPIRE no-op from a successful extension and from a pipeline result that
// could not be confirmed at all.
//
// `options.allowMissingKeys` names keys whose absence is EXPECTED (an optional
// marker that has not been written yet). For those keys a confirmed EXPIRE
// no-op stops being a failure: the "manual seed required" warning is suppressed
// and `allExtended` forgives them. That second half is load-bearing --
// `allExtended` gates runSeed's RETRY exit(1) and its preservationSucceeded
// diagnostic, so treating an expected-absent marker as a preservation failure
// would crash-loop a seeder over a key that is not supposed to exist yet.
//
// The #5364 contract is otherwise intact: a no-op on any key NOT in this list is
// still a real data condition, an unconfirmed result is still a failure even for
// a listed key, and `missingKeys` still reports every no-op for callers that
// need per-key truth.
export async function extendExistingTtlDetailed(keys, ttlSeconds = 600, options = {}) {
  const requestedKeys = Array.isArray(keys) ? keys : [];
  const allowMissingKeys = new Set(
    Array.isArray(options?.allowMissingKeys) ? options.allowMissingKeys : [],
  );
  if (requestedKeys.length === 0) {
    return {
      allExtended: true,
      extendedKeys: [],
      missingKeys: [],
      unconfirmedKeys: [],
    };
  }
  const url = process.env.UPSTASH_REDIS_REST_URL;
  const token = process.env.UPSTASH_REDIS_REST_TOKEN;
  if (!url || !token) {
    console.error('  Cannot extend TTL: missing Redis credentials');
    return {
      allExtended: false,
      extendedKeys: [],
      missingKeys: [],
      unconfirmedKeys: [...requestedKeys],
    };
  }
  try {
    // EXPIRE only refreshes TTL when key already exists (returns 0 on missing keys — no-op).
    // Check each result: keys that returned 0 are missing/expired and cannot be extended.
    const pipeline = requestedKeys.map(k => ['EXPIRE', k, ttlSeconds]);
    // Retry the pipeline call on transient Redis failures. A successful response
    // with some EXPIRE no-ops is a real missing-key condition, NOT a transient
    // error, so we only retry HTTP/network failures and return false for no-ops.
    const resp = await withRetry(async () => {
      const r = await fetch(`${url}/pipeline`, {
        method: 'POST',
        headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json' },
        body: JSON.stringify(pipeline),
        signal: AbortSignal.timeout(10_000),
      });
      if (!r.ok) {
        const err = new Error(`TTL extension pipeline failed (HTTP ${r.status})`);
        if (PERMANENT_4XX_STATUSES.has(r.status)) {
          err.nonRetryable = true;
        } else if (r.status === 429) {
          const retryAfterMs = parseRetryAfterUncappedMs(getResponseHeader(r.headers, 'Retry-After'));
          if (retryAfterMs != null) {
            err.retryAfterMs = retryAfterMs;
            // A bounded Redis operation cannot wait out a longer cooldown.
            if (retryAfterMs > MAX_RETRY_AFTER_MS) err.nonRetryable = true;
          }
        }
        throw err;
      }
      return r;
    }, 2, 1000);
    const results = await resp.json();
    const extendedKeys = [];
    const missingKeys = [];
    const unconfirmedKeys = [];
    for (let i = 0; i < requestedKeys.length; i++) {
      if (results?.[i]?.result === 1) {
        extendedKeys.push(requestedKeys[i]);
      } else if (results?.[i]?.result === 0) {
        missingKeys.push(requestedKeys[i]);
      } else {
        unconfirmedKeys.push(requestedKeys[i]);
      }
    }
    if (extendedKeys.length > 0) console.log(`  Extended TTL on ${extendedKeys.length} key(s) (${ttlSeconds}s)`);
    const strictMissingKeys = missingKeys.filter((key) => !allowMissingKeys.has(key));
    if (strictMissingKeys.length > 0) console.warn(`  WARNING: ${strictMissingKeys.length} key(s) were expired/missing — EXPIRE was a no-op; manual seed required`);
    if (unconfirmedKeys.length > 0) console.warn(`  WARNING: TTL extension result was unconfirmed for ${unconfirmedKeys.length} key(s)`);
    return {
      // An allowed-missing key is EXCLUDED from this verdict, not just from the
      // warning: `allExtended` gates runSeed's RETRY exit(1) and its
      // preservationSucceeded diagnostic, so counting an expected-absent marker
      // as a preservation failure would crash-loop a seeder over a key that is
      // not supposed to exist yet. The second clause requires a CONFIRMED no-op
      // -- an allowed-missing key whose EXPIRE result could not be read is still
      // a failure, because "we could not tell" is not "expectedly absent".
      allExtended: requestedKeys.every((key) => (
        extendedKeys.includes(key) || (allowMissingKeys.has(key) && missingKeys.includes(key))
      )),
      extendedKeys,
      missingKeys,
      unconfirmedKeys,
    };
  } catch (e) {
    console.error(`  TTL extension failed: ${e.message}`);
    return {
      allExtended: false,
      extendedKeys: [],
      missingKeys: [],
      unconfirmedKeys: [...requestedKeys],
    };
  }
}

// Returns true only when EVERY requested key was actually extended (EXPIRE
// returned 1 for all). Returns false on missing creds, network/HTTP failure,
// or any key that was missing/expired (EXPIRE no-op). Existing fire-and-forget
// callers ignore the return; a caller that treats a successful extension as
// proof the data is still alive (e.g. a market-closed skip that then reports
// fresh) MUST gate on this boolean and fall back to a real fetch on false —
// otherwise a silent extension failure looks green while the key expires.
export async function extendExistingTtl(keys, ttlSeconds = 600, options = {}) {
  const result = await extendExistingTtlDetailed(keys, ttlSeconds, options);
  return result.allExtended;
}

export function sleep(ms) {
  return new Promise((r) => setTimeout(r, ms));
}

// ─── Proxy helpers for sources that block Railway container IPs ───
const { resolveProxyString, resolveProxyStringConnect } = createRequire(import.meta.url)('./_proxy-utils.cjs');

export function resolveProxy(raw = process.env.PROXY_URL || '') {
  return resolveProxyString(raw);
}

// For HTTP CONNECT tunneling (httpsProxyFetchJson); keeps gate.decodo.com, not us.decodo.com.
export function resolveProxyForConnect() {
  return resolveProxyStringConnect();
}

// Scrub `scheme://user:pass@host` credentials out of anything we surface. Proxy auth
// strings are the only place seeders carry inline credentials, and they end up embedded
// in curl argv — see the execFileSync catch in curlFetch below.
export function redactProxyCredentials(text) {
  return String(text ?? '').replace(/(\w+:\/\/)[^/\s:@]+:[^/\s@]+@/g, '$1***:***@');
}

// curl-based fetch; throws on non-2xx. Returns response body as string.
// NOTE: requires curl binary — available in Dockerfile.relay (apk add curl) and Railway.
// Prefer httpsProxyFetchJson (pure Node.js) when possible; use curlFetch when curl-specific
// features are needed (e.g. --compressed, -L redirect following with proxy).
//
// `exec` is an injection seam for tests ONLY — the credential scrubbing below lives in a
// catch around execFileSync, and there is no other way to drive that branch deterministically.
export function curlFetch(
  url,
  proxyAuth,
  headers = {},
  { exec = execFileSync, timeoutMs = 15_000 } = {},
) {
  const curlTimeoutSeconds = Math.max(1, Math.ceil(timeoutMs / 1000));
  const args = ['-sS', '--compressed', '--max-time', String(curlTimeoutSeconds), '-L'];
  if (proxyAuth) {
    const proxyUrl = /^https?:\/\//i.test(proxyAuth) ? proxyAuth : `https://${proxyAuth}`;
    if (/^http:\/\//i.test(proxyUrl) && process.env.PROXY_ALLOW_INSECURE_HTTP !== '1') {
      throw new Error('Plain HTTP curl proxies require PROXY_ALLOW_INSECURE_HTTP=1');
    }
    args.push('-x', proxyUrl);
  }
  for (const [k, v] of Object.entries(headers)) args.push('-H', `${k}: ${v}`);
  args.push('-w', '\n%{http_code}');
  args.push(url);
  let raw;
  try {
    raw = exec('curl', args, {
      encoding: 'utf8',
      timeout: timeoutMs + 5_000,
      stdio: ['pipe', 'pipe', 'pipe'],
    });
  } catch (err) {
    // SECURITY: when curl itself exits non-zero (SSL_ERROR_SYSCALL, "CONNECT tunnel
    // failed", DNS), execFileSync builds an Error whose message is the ENTIRE argv —
    // including `-x http://user:pass@proxy-host`. Seeders log that message, so the proxy
    // credentials were being written verbatim into Railway logs on every curl-level
    // failure (observed continuously during the 2026-07-13 GDELT 429 storm). Re-throw
    // with curl's own stderr, which names the failure without echoing the command.
    //
    // Dropping `.status` is load-bearing, not incidental: execFileSync sets it to curl's
    // EXIT CODE (35, 56, 7…). _gdelt-fetch.mjs uses `.status` to distinguish upstream
    // HTTP responses from transport failures, so leaking the exit code would classify
    // curl exit 35 as an HTTP response and publish the wrong route diagnostic. Only a
    // genuine HTTP status may carry `.status`.
    const stderr = redactProxyCredentials(err?.stderr || '').trim().split('\n').filter(Boolean).pop();
    throw Object.assign(
      new Error(`curl failed: ${stderr || redactProxyCredentials(err?.message) || 'unknown error'}`),
      { curlFailed: true },
    );
  }
  const nl = raw.lastIndexOf('\n');
  const status = parseInt(raw.slice(nl + 1).trim(), 10);
  if (status < 200 || status >= 300) throw Object.assign(new Error(`HTTP ${status}`), { status });
  return raw.slice(0, nl);
}

// Pure Node.js HTTPS-through-proxy (CONNECT tunnel).
// proxyAuth format: "user:pass@host:port" (bare/Decodo → TLS) OR
//                  "https://user:pass@host:port" (explicit TLS) OR
//                  "http://user:pass@host:port"  (explicit plain TCP)
// Bare/undeclared-scheme proxies always use TLS (Decodo gate.decodo.com requires it).
// Explicit http:// proxies use plain TCP to avoid breaking non-TLS setups.
async function httpsProxyFetchJson(url, proxyAuth, proxyAttempt = 0) {
  const { buffer } = await httpsProxyFetchRaw(url, proxyAuth, { accept: 'application/json', proxyAttempt });
  return JSON.parse(buffer.toString('utf8'));
}

// `headers` are forwarded to the origin (e.g. an API subscription key); the
// shared User-Agent is applied unless a caller sets its own.
export async function httpsProxyFetchRaw(url, proxyAuth, {
  accept = '*/*', timeoutMs = 20_000, signal, proxyAttempt = 0, headers = {}, onErrorResponse, maxResponseBytes,
} = {}) {
  const { proxyFetch, parseProxyConfigForAttempt } = createRequire(import.meta.url)('./_proxy-utils.cjs');
  const proxyConfig = parseProxyConfigForAttempt(proxyAuth, proxyAttempt);
  if (!proxyConfig) throw new Error('Invalid proxy auth string');
  const result = await proxyFetch(url, proxyConfig, { accept, timeoutMs, signal, maxResponseBytes, headers: { 'User-Agent': CHROME_UA, ...headers } });
  if (!result.ok) {
    onErrorResponse?.(result);
    throw Object.assign(new Error(`HTTP ${result.status}`), { status: result.status });
  }
  return { buffer: result.buffer, contentType: result.contentType };
}

// Whether a proxy error should be retried. A retry reaches a DIFFERENT exit IP
// only because the caller advances its attempt index (see fredFetchJson) — a
// Decodo sticky port pins one exit for the life of the session and never
// rotates on its own. Reading it the other way round is what let three retries
// pile onto one dead exit during the 2026-09-10 outage (#7963).
// Covers 5xx/522, DNS/socket errors, AND mid-handshake TLS tears — the
// last group is load-bearing: if a TLS-tear isn't classified transient, the
// retry loop breaks on attempt 1 and falls to a direct FRED fetch, which a
// datacenter IP gets rate-limited/blocked on → the whole batch fails. Exported
// for unit testing (the proxy fetch itself is network-bound and not injectable).
export function isTransientProxyError(message) {
  return /HTTP(?:\/1\.[01])? 5\d{2}|522|timeout|ECONNRESET|ECONNREFUSED|ETIMEDOUT|EAI_AGAIN|EPIPE|socket (disconnected|hang up)|TLS connection|tls_get_more_records|packet length too long|SSL routines|secure TLS connection/i.test(message || '');
}

// Whether the ORIGIN refused this particular egress IP — a failure that a
// different sticky exit can actually fix. FRED blocks datacenter IPs, which is
// why the proxy leg exists at all (#2911), so a 403 served through a healthy
// tunnel says "this exit is unwelcome" and the next sticky exit may be fine.
//
// 403 ONLY, deliberately. 429 was in the first draft and came out under review.
// Nothing in this repo establishes that FRED's rate limit is scoped to the
// source IP — #2911 cites direct-fetch TIMEOUTS as the observed motivation, not
// IP-keyed 429s — and `api_key` travels in the query string (_fred-seeder.mjs),
// which is how quota is conventionally scoped. If the limit is per-key,
// rotating exits cannot clear it and merely triples the request count against a
// quota that is already exhausted, while the Retry-After FRED sends is
// discarded anyway because httpsProxyFetchRaw drops `result.headers` when it
// throws. Widen to 429 only with evidence that FRED's 429 is IP-scoped, and
// plumb Retry-After first — _proxy-utils.cjs already preserves those headers
// through the tunnel for exactly this reason (#6241).
//
// Deliberately separate from isTransientProxyError rather than folded into it:
// that predicate is shared by other seeders whose retry budgets are tuned to
// their own upstreams, and widening it would change their behaviour too. Kept
// status-based rather than message-based because proxyConnectTunnel and
// httpsProxyFetchRaw both collapse to `HTTP <status>` text, and only the
// structured fields tell the two apart.
//
// Gateway-layer rejections are excluded: proxyConnectTunnel marks its own
// failures `proxyConnect: true` for exactly this decision — see its comment in
// _proxy-utils.cjs, "only the origin case can be helped by a different exit". A
// 407, or a gateway 403 for a port outside the account's allocation, means the
// credentials or plan are wrong and no exit fixes that. Other origin 4xx are
// excluded too: every exit answers a bad series id identically, so rotating on
// one would just burn the proxy budget before the direct leg gets its turn.
//
// Takes the ERROR OBJECT, not a message string — it reads structured fields, so
// a mistaken isExitRefusalError(err.message) would silently return false
// forever and quietly disable rotation. The typeof guard makes that loud-ish
// rather than accidental, and a regression test pins it.
export function isExitRefusalError(error) {
  if (!error || typeof error !== 'object' || error.proxyConnect) return false;
  return error.status === 403;
}

const FRED_JSON_HEADERS = { Accept: 'application/json', 'User-Agent': CHROME_UA };

function throwIfFredMaintenance(body, contentType) {
  if (/\btext\/html\b/i.test(contentType || '')
    && /<h1[^>]*>\s*We(?:['\u2019]ll| will) be back soon!?\s*<\/h1>/i.test(body)) {
    throw markPreservableFetchFailure(new Error('FRED API is under maintenance'));
  }
}

function parseFredJson(body, contentType) {
  throwIfFredMaintenance(body, contentType);
  return JSON.parse(body);
}

// FRED's own edge returns sporadic 5xx on individual series. Observed
// 2026-08-26: four consecutive 24/24 runs, then `FRED T10Y2Y: fetch failed —
// direct: HTTP 502` and the same for UNRATE, publishing 22/24 — enough to trip
// health's minRecordCount of 24 for the whole hour. Both series answered 200
// when queried directly minutes later, and adjacent runs fetched them fine.
//
// Deliberately status-only, and deliberately NOT reusing isTransientProxyError:
// that predicate also matches timeouts and socket tears, and a direct leg that
// timed out has already burned its 20s budget — retrying it would double the
// worst case inside runSeed's fetch-phase deadline for a leg that is plainly
// broken. A 5xx fails fast, so this retry costs a few hundred milliseconds.
const FRED_DIRECT_ATTEMPTS = 2;
function isRetriableFredStatus(status) {
  return Number.isInteger(status) && status >= 500 && status <= 599;
}

// Direct FRED fetch with a bounded retry on a fast-failing 5xx. Shared by the
// proxy-fallback path and the no-proxy path: a transient 502 is transient
// regardless of which leg reached it, and having only one of the two retry is
// how the asymmetry below went unnoticed — the proxy leg already retried three
// times while its own fallback got a single attempt.
async function fredDirectFetchJson(url) {
  let lastError;
  for (let attempt = 1; attempt <= FRED_DIRECT_ATTEMPTS; attempt += 1) {
    try {
      const r = await fetch(url, { headers: FRED_JSON_HEADERS, signal: AbortSignal.timeout(20_000) });
      if (r.ok) {
        if (/\btext\/html\b/i.test(r.headers?.get?.('content-type') || '')) {
          return parseFredJson(await r.text(), r.headers.get('content-type'));
        }
        return await r.json();
      }
      throw Object.assign(new Error(`HTTP ${r.status}`), { status: r.status });
    } catch (error) {
      lastError = error;
      if (attempt < FRED_DIRECT_ATTEMPTS && isRetriableFredStatus(error?.status)) {
        await new Promise((resolve) => setTimeout(resolve, 350 * attempt + Math.random() * 250));
        continue;
      }
      if (isRetriableFredStatus(error?.status) || error?.status === 429) markPreservableFetchFailure(error);
      throw error;
    }
  }
  throw lastError;
}

// Fetch JSON from a FRED URL, routing through proxy when available.
// Proxy-first: FRED consistently blocks/throttles Railway datacenter IPs,
// so try proxy first to avoid 20s timeout on every direct attempt.
export async function fredFetchJson(url, proxyAuth) {
  if (proxyAuth) {
    // Advance Decodo sticky ports before falling back direct. Reusing port
    // 10001 kept all retries on the failed exit during the 2026-09-10 outage.
    // isTransientProxyError covers TLS-handshake tears — see its doc comment.
    //
    // The exits are recorded so the warning below can name them. Rotation
    // no-ops silently for any host outside parseProxyConfigForAttempt's sticky
    // map (us.decodo.com, an ISP or city-targeted endpoint) or any port outside
    // its range, and a healthy run looks identical whether rotation engaged or
    // the configured exit simply recovered. Three identical ports in that line
    // is the operator's one-line proof the rotation is inert for the deployed
    // PROXY_URL — without it the next outage reads exactly like the last one.
    const { parseProxyConfigForAttempt } = createRequire(import.meta.url)('./_proxy-utils.cjs');
    const triedExits = [];
    let lastProxyErr;
    for (let attempt = 1; attempt <= 3; attempt++) {
      triedExits.push(parseProxyConfigForAttempt(proxyAuth, attempt - 1)?.port ?? '?');
      try {
        const { buffer, contentType } = await httpsProxyFetchRaw(url, proxyAuth, {
          accept: 'application/json', proxyAttempt: attempt - 1,
          onErrorResponse: ({ buffer: errorBody, contentType: errorContentType }) => {
            throwIfFredMaintenance(errorBody.toString('utf8'), errorContentType);
          },
        });
        return parseFredJson(buffer.toString('utf8'), contentType);
      } catch (proxyErr) {
        lastProxyErr = proxyErr;
        // Two different reasons to try the next exit: the hop broke (transient),
        // or this exit's IP is the thing FRED is refusing (403/429). The second
        // is what the rotation above is FOR, and it used to break the loop after
        // one attempt because the transient predicate matches no 4xx.
        const rotatable = isTransientProxyError(proxyErr.message) || isExitRefusalError(proxyErr);
        if (attempt < 3 && rotatable) {
          await new Promise((r) => setTimeout(r, 400 * attempt + Math.random() * 300));
          continue;
        }
        break;
      }
    }
    console.warn(`  [fredFetch] proxy failed after retries on exits [${triedExits.join(', ')}] (${lastProxyErr?.message}) — retrying direct`);
    try {
      return await fredDirectFetchJson(url);
    } catch (directErr) {
      const proxyMaintenance = lastProxyErr?.preservableFetchFailure === true;
      const connectTimeout = directErr.cause?.code === 'UND_ERR_CONNECT_TIMEOUT';
      const directMessage = `${directErr.message}${connectTimeout ? ' (UND_ERR_CONNECT_TIMEOUT)' : ''}`;
      const message = `${proxyMaintenance ? `proxy: ${lastProxyErr.message}; ` : ''}direct: ${directMessage}`;
      throw Object.assign(new Error(message), {
        cause: directErr,
        preservableFetchFailure: directErr.preservableFetchFailure === true
          || (proxyMaintenance && (directErr.status === 403 || directErr.name === 'TimeoutError' || connectTimeout)),
      });
    }
  }
  return fredDirectFetchJson(url);
}

// Fetch JSON from an IMF DataMapper URL, direct-first with proxy fallback.
// Direct timeout is short (10s) since IMF blocks Railway IPs with 403 quickly.
export async function imfFetchJson(url, proxyAuth) {
  try {
    const r = await fetch(url, {
      headers: { 'User-Agent': CHROME_UA, Accept: 'application/json' },
      signal: AbortSignal.timeout(10_000),
    });
    if (!r.ok) throw new Error(`HTTP ${r.status}`);
    return await r.json();
  } catch (directErr) {
    if (!proxyAuth) throw directErr;
    console.warn(`  [IMF] Direct fetch failed (${directErr.message}); retrying via proxy`);
    return httpsProxyFetchJson(url, proxyAuth);
  }
}

// ---------------------------------------------------------------------------
// IMF SDMX 3.0 API (api.imf.org) — replaces blocked DataMapper API.
//
// Auth status (2026-05): currently allows unauthenticated requests, but the
// gateway has been observed flipping to hard 401 enforcement intermittently
// (one ~hours-long window on 2026-05-09 SIGTERM'd seed-bundle-market-backup
// in production). Set IMF_API_KEY to send `Ocp-Apim-Subscription-Key` for
// forward-compatibility — get a key at https://portal.api.imf.org/ (Sign in
// → Products → IMF Data SDMX API → Subscribe → Profile → Subscriptions →
// Primary key). One subscription unlocks both SDMX 2.1 and 3.0. The gateway
// returns a misleading `WWW-Authenticate: Bearer` 401; the actual scheme is
// APIM subscription key, NOT Bearer.
// ---------------------------------------------------------------------------
const IMF_SDMX_BASE = 'https://api.imf.org/external/sdmx/3.0';

// Build auth headers for api.imf.org. Returns {} when IMF_API_KEY is unset —
// IMF currently allows unauthenticated requests through, but flipped to hard
// 401 enforcement for ~hours on 2026-05-09 (captured in seed-bundle-market-
// backup logs). Sending the key when we have it is forward-compatible with
// permanent enforcement; the omit-on-empty path preserves today's working
// state. The `nonRetryable` 4xx guard in the fetcher (paired with this
// helper) prevents a 180s SIGTERM the next time enforcement lands.
export function imfAuthHeaders() {
  const key = process.env.IMF_API_KEY;
  return key ? { 'Ocp-Apim-Subscription-Key': key } : {};
}

/**
 * Normalize an SDMX 3.0 monthly period from `YYYY-MMM` (the on-the-wire
 * shape, e.g. `2026-M03`) to ISO `YYYY-MM` so downstream date math —
 * `period.split('-')`, `parseInt(month, 10)`, key comparisons — works
 * without special-casing. The M-prefix silently corrupts these
 * computations: `parseInt("M03", 10)` returns NaN, so 12-month delta
 * lookups against `byMonth[priorMonth]` always miss.
 *
 * Other SDMX 3.0 frequencies pass through unchanged:
 *   - Annual:    `YYYY`               (e.g. `2024`) — used by WEO/FM
 *   - Quarterly: `YYYY-Q1..Q4`        (e.g. `2024-Q3`) — sortable as-is
 *   - Daily:     `YYYY-MM-DD`         (e.g. `2024-03-15`) — used by ECB
 *   - Monthly:   `YYYY-MMM` → YYYY-MM (e.g. `2026-M03` → `2026-03`)
 *
 * Future monthly/quarterly SDMX consumers MUST call this at ingest
 * (right after reading `timeValues[parseInt(obsKey, 10)]`) so callers
 * downstream can keep using simple string comparisons and ISO splits.
 *
 * @param {string|null|undefined} period
 * @returns {string|null|undefined} Normalized period (or input unchanged for falsy/non-string)
 */
export function normalizeSdmxPeriod(period) {
  if (typeof period !== 'string') return period;
  return period.replace(/-M(\d{2})$/, '-$1');
}

/**
 * IMF WEO/FM annual indicator fetcher. Hardcoded to annual frequency by URL
 * construction (`*.${indicator}.A`) — period values come back as bare year
 * strings (`"2024"`), so no SDMX-period normalization is required here.
 *
 * NOTE for future extensions: if you need IMF monthly or quarterly data
 * (e.g. IRFCL, IFS, BOP), do NOT bolt frequency onto this helper — the
 * dimension layout differs (e.g. IRFCL is 4-dim COUNTRY.INDICATOR.SECTOR.FREQUENCY,
 * not WEO's 2-dim COUNTRY.INDICATOR). Roll a custom fetch and call
 * `normalizeSdmxPeriod()` on every period before storing it as a key.
 * See `scripts/seed-gold-cb-reserves.mjs::fetchIrfclMonthlySeries` for the
 * canonical monthly pattern.
 *
 * `release` holds the dataset-level PUBLICATION_DATE and UPDATE_DATE of this
 * one response, as IMF supplied them, or null when absent, malformed or
 * ambiguous. They are reference metadata, so the request uses
 * `attributes=all`: `dsd` omits them and `msd` drops the series attributes.
 * Neither date names a WEO edition.
 *
 * @returns {Promise<{ values: Record<string, Record<string, number>>, release: { publishedAt: string|null, updatedAt: string|null } }>}
 */
export async function imfSdmxFetchIndicatorWithRelease(indicator, { database = 'WEO', years } = {}) {
  const agencyMap = { WEO: 'IMF.RES', FM: 'IMF.FAD' };
  const agency = agencyMap[database] || 'IMF.RES';
  const url = `${IMF_SDMX_BASE}/data/dataflow/${agency}/${database}/+/*.${indicator}.A?dimensionAtObservation=TIME_PERIOD&attributes=all&measures=all`;

  const json = await withRetry(async () => {
    const r = await fetch(url, {
      headers: { 'User-Agent': CHROME_UA, Accept: 'application/json', ...imfAuthHeaders() },
      signal: AbortSignal.timeout(60_000),
    });
    if (!r.ok) {
      const err = new Error(`IMF SDMX ${indicator}: HTTP ${r.status}`);
      if (PERMANENT_4XX_STATUSES.has(r.status)) err.nonRetryable = true;
      // 429 (rate limit) and 503 (overloaded) typically carry Retry-After;
      // attach so withRetry can honor the upstream hint.
      if (r.status === 429 || r.status === 503) {
        err.retryAfterMs = parseRetryAfterUncappedMs(r.headers.get('retry-after'));
        if (err.retryAfterMs > MAX_RETRY_AFTER_MS) err.nonRetryable = true;
      }
      throw err;
    }
    return r.json();
  }, 2, 2000);

  const ds = json?.data?.dataSets?.[0];
  const struct = json?.data?.structures?.[Number.isInteger(ds?.structure) ? ds.structure : 0];
  const release = {
    publishedAt: imfDatasetDate(struct, ds, 'PUBLICATION_DATE'),
    updatedAt: imfDatasetDate(struct, ds, 'UPDATE_DATE'),
  };
  if (!struct || !ds?.series) return { values: {}, release };

  const countryDim = struct.dimensions.series.find(d => d.id === 'COUNTRY');
  const countryDimPos = struct.dimensions.series.findIndex(d => d.id === 'COUNTRY');
  const timeDim = struct.dimensions.observation.find(d => d.id === 'TIME_PERIOD');
  if (!countryDim || countryDimPos === -1 || !timeDim) return { values: {}, release };

  const countryValues = countryDim.values.map(v => v.id);
  const timeValues = timeDim.values.map(v => v.value || v.id);
  const yearSet = years ? new Set(years.map(String)) : null;

  const result = {};
  for (const [seriesKey, seriesData] of Object.entries(ds.series)) {
    const keyParts = seriesKey.split(':');
    const countryIdx = parseInt(keyParts[countryDimPos], 10);
    const iso3 = countryValues[countryIdx];
    if (!iso3) continue;

    const byYear = {};
    for (const [obsKey, obsVal] of Object.entries(seriesData.observations || {})) {
      const year = timeValues[parseInt(obsKey, 10)];
      if (!year || (yearSet && !yearSet.has(year))) continue;
      const v = obsVal?.[0];
      if (v != null) byYear[year] = parseFloat(v);
    }
    if (Object.keys(byYear).length > 0) result[iso3] = byYear;
  }
  return { values: result, release };
}

export async function imfSdmxFetchIndicator(indicator, options) {
  return (await imfSdmxFetchIndicatorWithRelease(indicator, options)).values;
}

const IMF_RELEASE_DATE_RE = /^(\d{4})-(\d{2})-(\d{2})(?:T(?:[01]\d|2[0-3]):[0-5]\d(?::[0-5]\d(?:\.\d+)?)?(?:Z|[+-](?:[01]\d|2[0-3]):[0-5]\d)?)?$/;

function imfDatasetDate(struct, ds, id) {
  const descriptors = struct?.attributes?.dataSet;
  if (!Array.isArray(descriptors)) return null;
  const positions = descriptors.flatMap((d, i) => (d?.id === id ? [i] : []));
  if (positions.length !== 1) return null;
  let value = ds?.attributes?.[positions[0]];
  if (Array.isArray(value)) {
    if (value.length !== 1) return null;
    [value] = value;
  }
  if (Number.isInteger(value)) {
    const coded = descriptors[positions[0]].values?.[value];
    value = coded?.value ?? coded?.id;
  }
  if (typeof value !== 'string') return null;
  const match = IMF_RELEASE_DATE_RE.exec(value);
  if (!match) return null;
  const [y, m, d] = match.slice(1, 4).map(Number);
  const day = new Date(Date.UTC(y, m - 1, d));
  return day.getUTCFullYear() === y && day.getUTCMonth() === m - 1 && day.getUTCDate() === d ? value : null;
}

// ---------------------------------------------------------------------------
// Learned Routes — persist successful scrape URLs across seed runs
// ---------------------------------------------------------------------------

// Validate a URL's hostname against a list of allowed domains (same list used
// for EXA includeDomains). Prevents stored-SSRF from Redis-persisted URLs.
export function isAllowedRouteHost(url, allowedHosts) {
  try {
    const hostname = new URL(url).hostname.replace(/^www\./, '');
    return allowedHosts.some(h => hostname === h || hostname.endsWith('.' + h));
  } catch {
    return false;
  }
}

// Batch-read all learned routes for a scope via single Upstash pipeline request.
// Returns Map<key → routeData>. Non-fatal: throws on HTTP error (caller catches).
export async function bulkReadLearnedRoutes(scope, keys) {
  if (!keys.length) return new Map();
  const { url, token } = getRedisCredentials();
  const pipeline = keys.map(k => ['GET', `seed-routes:${scope}:${k}`]);
  const resp = await fetch(`${url}/pipeline`, {
    method: 'POST',
    headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json' },
    body: JSON.stringify(pipeline),
    signal: AbortSignal.timeout(10_000),
  });
  if (!resp.ok) throw new Error(`bulkReadLearnedRoutes HTTP ${resp.status}`);
  const results = await resp.json();
  const map = new Map();
  for (let i = 0; i < keys.length; i++) {
    const raw = results[i]?.result;
    if (!raw) continue;
    try { map.set(keys[i], JSON.parse(raw)); }
    catch { console.warn(`  [routes] malformed JSON for ${keys[i]} — skipping`); }
  }
  return map;
}

// Batch-write route updates and hard-delete evicted routes via single pipeline.
// Keys in updates always win over deletes (SET/DEL conflict resolution).
// DELs are sent before SETs to ensure correct ordering.
export async function bulkWriteLearnedRoutes(scope, updates, deletes = new Set()) {
  const { url, token } = getRedisCredentials();
  const ROUTE_TTL = 14 * 24 * 3600; // 14 days
  const effectiveDeletes = [...deletes].filter(k => !updates.has(k));
  const pipeline = [];
  for (const k of effectiveDeletes)
    pipeline.push(['DEL', `seed-routes:${scope}:${k}`]);
  for (const [k, v] of updates)
    pipeline.push(['SET', `seed-routes:${scope}:${k}`, JSON.stringify(v), 'EX', ROUTE_TTL]);
  if (!pipeline.length) return;
  const resp = await fetch(`${url}/pipeline`, {
    method: 'POST',
    headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json' },
    body: JSON.stringify(pipeline),
    signal: AbortSignal.timeout(15_000),
  });
  if (!resp.ok) throw new Error(`bulkWriteLearnedRoutes HTTP ${resp.status}`);
  console.log(`  [routes] written: ${updates.size} updated, ${effectiveDeletes.length} deleted`);
}

// Decision tree for a single seed item: try learned route first, fall back to EXA.
// All external I/O is injected so this function can be unit-tested without Redis or HTTP.
//
// Returns: { localPrice, sourceSite, routeUpdate, routeDelete }
//   routeUpdate — route object to persist (null = nothing to write)
//   routeDelete — true if the Redis key should be hard-deleted
export async function processItemRoute({
  learned,           // route object from Redis, or undefined/null on first run
  allowedHosts,      // string[] — normalised (no www.), same as EXA includeDomains
  currency,          // e.g. 'AED'
  itemId,            // e.g. 'sugar' — used only for log messages
  fxRate,            // number | null
  itemUsdMax = null, // per-item bulk cap in USD (ITEM_USD_MAX[itemId])
  tryDirectFetch,    // async (url, currency, itemId, fxRate) => number | null
  scrapeFirecrawl,   // async (url, currency) => { price, source } | null
  fetchViaExa,       // async () => { localPrice, sourceSite } | null  (caller owns EXA+FC logic)
  sleep: sleepFn,    // async ms => void
  firecrawlDelayMs = 0,
}) {
  let localPrice = null;
  let sourceSite = '';
  let routeUpdate = null;
  let routeDelete = false;

  if (learned) {
    if (learned.failsSinceSuccess >= 2 || !isAllowedRouteHost(learned.url, allowedHosts)) {
      routeDelete = true;
      console.log(`    [learned✗] ${itemId}: evicting (${learned.failsSinceSuccess >= 2 ? '2 failures' : 'invalid host'})`);
    } else {
      localPrice = await tryDirectFetch(learned.url, currency, itemId, fxRate);
      if (localPrice !== null) {
        sourceSite = learned.url;
        routeUpdate = { ...learned, hits: learned.hits + 1, failsSinceSuccess: 0, lastSuccessAt: Date.now() };
        console.log(`    [learned✓] ${itemId}: ${localPrice} ${currency}`);
      } else {
        await sleepFn(firecrawlDelayMs);
        const fc = await scrapeFirecrawl(learned.url, currency);
        const fcSkip = fc && fxRate && itemUsdMax && (fc.price * fxRate) > itemUsdMax;
        if (fc && !fcSkip) {
          localPrice = fc.price;
          sourceSite = fc.source;
          routeUpdate = { ...learned, hits: learned.hits + 1, failsSinceSuccess: 0, lastSuccessAt: Date.now() };
          console.log(`    [learned-FC✓] ${itemId}: ${localPrice} ${currency}`);
        } else {
          const newFails = learned.failsSinceSuccess + 1;
          if (newFails >= 2) {
            routeDelete = true;
            console.log(`    [learned✗→EXA] ${itemId}: 2 failures — evicting, retrying via EXA`);
          } else {
            routeUpdate = { ...learned, failsSinceSuccess: newFails };
            console.log(`    [learned✗→EXA] ${itemId}: failed (${newFails}/2), retrying via EXA`);
          }
        }
      }
    }
  }

  if (localPrice === null) {
    const exaResult = await fetchViaExa();
    if (exaResult?.localPrice != null) {
      localPrice = exaResult.localPrice;
      sourceSite = exaResult.sourceSite || '';
      if (sourceSite && isAllowedRouteHost(sourceSite, allowedHosts)) {
        routeUpdate = { url: sourceSite, lastSuccessAt: Date.now(), hits: 1, failsSinceSuccess: 0, currency };
        console.log(`    [EXA->learned] ${itemId}: saved ${sourceSite.slice(0, 55)}`);
      }
    }
  }

  return { localPrice, sourceSite, routeUpdate, routeDelete };
}

// Five 8s timeouts plus four 150ms gaps take at most 40.6s, leaving ample
// headroom inside downstream seeds' 240s fetch-phase deadline.
const MAX_POINT_OF_USE_FX_REQUESTS = 5;

function isFinitePositiveRate(value) {
  return Number.isFinite(value) && value > 0;
}

/**
 * Cap Yahoo recovery work for conversion seeds. Currencies beyond the budget are
 * not attempted: prefer a finite prior/cache rate when present (still marked as
 * fallback provenance — unattempted is not live), else the caller fallback table.
 * @param {Record<string, string>} fxSymbols
 * @param {Record<string, number>} [fallbacks]
 * @param {Record<string, unknown>} [priorRates] rates already on hand (e.g. cache)
 */
async function fetchPointOfUseFxRates(fxSymbols, fallbacks = {}, priorRates = {}) {
  const attemptedSymbols = {};
  const deferredCurrencies = [];
  let requestCount = 0;

  for (const [currency, symbol] of Object.entries(fxSymbols)) {
    if (currency === 'USD' || requestCount < MAX_POINT_OF_USE_FX_REQUESTS) {
      attemptedSymbols[currency] = symbol;
      if (currency !== 'USD') requestCount += 1;
    } else {
      deferredCurrencies.push(currency);
    }
  }

  const result = await fetchYahooFxRatesWithProvenance(attemptedSymbols, fallbacks);
  for (const currency of deferredCurrencies) {
    const prior = priorRates[currency];
    result.rates[currency] = isFinitePositiveRate(prior)
      ? prior
      : (fallbacks[currency] ?? null);
  }
  result.fallbackCurrencies.push(...deferredCurrencies);
  return result;
}

/**
 * Shared FX rates cache — reads from Redis `shared:fx-rates:v1` (4h TTL).
 * Falls back to fetching from Yahoo Finance if the key is missing/expired.
 * All seeds needing currency conversion should call this instead of their own fetchFxRates().
 *
 * @param {Record<string, string>} fxSymbols  - map of { CCY: 'CCYUSD=X' }
 * @param {Record<string, number>} fallbacks  - hardcoded rates to use if Yahoo fails
 */
export async function getSharedFxRates(fxSymbols, fallbacks) {
  const SHARED_KEY = 'shared:fx-rates:v1';
  const { url, token } = getRedisCredentials();

  // Try reading cached rates first (read-only — only seed-fx-rates.mjs writes this key)
  try {
    const cached = await redisGet(url, token, SHARED_KEY);
    if (cached && typeof cached === 'object' && Object.keys(cached).length > 0) {
      console.log('  FX rates: loaded from shared cache');
      const hasValidFallbackProvenance = (
        Array.isArray(cached.fallbackCurrencies)
        && cached.fallbackCurrencies.every(c => typeof c === 'string')
      );
      const cachedFallbacks = new Set(hasValidFallbackProvenance ? cached.fallbackCurrencies : []);
      // The canonical seed publishes failed Yahoo quotes as null. Retry only
      // the gaps this consumer needs, then use its fallback table at the point
      // of use so direct readers never mistake constants for live quotes. A
      // markerless legacy snapshot may itself contain fallback constants, so
      // none of its requested non-USD values are trusted as live.
      const missing = Object.keys(fxSymbols).filter(c => (
        c !== 'USD'
        && (!hasValidFallbackProvenance || cached[c] == null || cachedFallbacks.has(c))
      ));
      if (missing.length === 0) return cached;
      console.log(`  FX rates: fetching ${missing.length} missing currencies from Yahoo`);
      const extra = await fetchPointOfUseFxRates(
        Object.fromEntries(missing.map(c => [c, fxSymbols[c]])),
        fallbacks,
        cached,
      );
      const retried = new Set(missing);
      const fallbackCurrencies = [
        ...[...cachedFallbacks].filter(c => !retried.has(c)),
        ...extra.fallbackCurrencies,
      ];
      return {
        ...cached,
        ...extra.rates,
        fallbackCurrencies: [...new Set(fallbackCurrencies)],
      };
    }
  } catch {
    // Cache read failed — fall through to live fetch
  }

  console.log('  FX rates: cache miss — fetching from Yahoo Finance');
  const result = await fetchPointOfUseFxRates(fxSymbols, fallbacks);
  return { ...result.rates, fallbackCurrencies: result.fallbackCurrencies };
}

/**
 * Fetch USD-per-unit Yahoo FX quotes and record which currencies used the
 * caller-provided fallback table. The provenance is based on the actual fetch
 * outcome, never float equality with the fallback constant.
 */
export async function fetchYahooFxRatesWithProvenance(fxSymbols, fallbacks = {}) {
  const rates = {};
  const fallbackCurrencies = [];
  let requestsRemaining = Object.keys(fxSymbols).filter(currency => currency !== 'USD').length;
  for (const [currency, symbol] of Object.entries(fxSymbols)) {
    if (currency === 'USD') { rates['USD'] = 1.0; continue; }
    let price = null;
    try {
      const url = `https://query1.finance.yahoo.com/v8/finance/chart/${encodeURIComponent(symbol)}`;
      const resp = await fetch(url, {
        headers: { 'User-Agent': CHROME_UA },
        signal: AbortSignal.timeout(8_000),
      });
      if (resp.ok) {
        const data = await resp.json();
        price = data?.chart?.result?.[0]?.meta?.regularMarketPrice;
      }
    } catch {}
    if (isFinitePositiveRate(price)) {
      rates[currency] = price;
    } else {
      rates[currency] = fallbacks[currency] ?? null;
      fallbackCurrencies.push(currency);
    }
    requestsRemaining -= 1;
    if (requestsRemaining > 0) await sleep(150);
  }
  console.log('  FX rates fetched:', JSON.stringify(rates));
  if (fallbackCurrencies.length > 0) {
    console.warn(`  FX rates using fallbacks (${fallbackCurrencies.length}): ${fallbackCurrencies.join(', ')}`);
  }
  return { rates, fallbackCurrencies };
}

/**
 * Read the current canonical snapshot from Redis before a seed run overwrites it.
 * Used by seed scripts that compute WoW deltas (bigmac, grocery-basket).
 * Returns null on any error by default — scripts must handle first-run (no prev
 * data). Pass strict:true when overwriting without the prior snapshot would lose
 * accumulated state; missing keys still return null, while read failures throw.
 * Pass includeEnvelopeMeta:true when a cross-seed calculation must bind the
 * payload and its fetchedAt clock to the same atomic Redis GET.
 * timeoutMs bounds the whole read, body included: raise it for multi-MB keys.
 */
export async function readSeedSnapshot(
  canonicalKey,
  { strict = false, includeEnvelopeMeta = false, timeoutMs = 5_000 } = {},
) {
  const url = process.env.UPSTASH_REDIS_REST_URL;
  const token = process.env.UPSTASH_REDIS_REST_TOKEN;
  if (!url || !token) return null;
  try {
    const resp = await fetch(`${url}/get/${encodeURIComponent(canonicalKey)}`, {
      headers: { Authorization: `Bearer ${token}`, 'User-Agent': CHROME_UA },
      signal: AbortSignal.timeout(timeoutMs),
    });
    if (!resp.ok) {
      if (strict) throw new Error(`Redis snapshot read failed: HTTP ${resp.status}`);
      return null;
    }
    // Upstash's GET contract has one unambiguous miss: an HTTP-200 JSON
    // envelope with an explicitly-null `result`. In strict mode, every other
    // malformed envelope is a read failure, not a cold start. This matters for
    // rolling/baseline seeders: degrading `{}` or invalid JSON to null can make
    // two ambiguous reads look like a missing key and overwrite last-good
    // state. Non-strict callers retain their best-effort null degradation.
    const body = strict
      ? await parseRedisCommandResponse(resp, 'Redis snapshot read')
      : await resp.json();
    const result = body?.result;
    if (result === null) return null;
    if (strict && typeof result !== 'string') {
      throw new Error('Redis snapshot read returned a non-string result');
    }
    if (!strict && !result) return null;
    let parsed;
    try {
      parsed = parseStoredSeedJson(result);
    } catch (cause) {
      if (strict) {
        throw Object.assign(new Error('Redis snapshot read returned malformed stored JSON'), { cause });
      }
      return null;
    }
    if (strict && parsed === null) {
      throw new Error('Redis snapshot read returned a null stored snapshot');
    }
    // Envelope-aware: WoW/prev baselines (bigmac, grocery-basket, fear-greed)
    // must see bare legacy-shape data whether the last write was pre- or post-
    // contract-migration. unwrapEnvelope is a no-op on legacy values.
    const envelope = unwrapEnvelope(parsed);
    return includeEnvelopeMeta ? { data: envelope.data, meta: envelope._seed } : envelope.data;
  } catch (error) {
    if (strict) throw error;
    return null;
  }
}

/**
 * Resolve recordCount for runSeed's freshness metadata write.
 *
 * Resolution order:
 *   1. opts.recordCount (function or number) — the seeder declared it explicitly
 *   2. Auto-detect from a known shape (Array.isArray, .predictions, .events, ...)
 *   3. payloadBytes > 0 → 1 (proven-payload fallback) + warn so the seeder author
 *      adds an explicit opts.recordCount
 *   4. 0
 *
 * The fallback exists because seeders publishing custom shapes would otherwise
 * trigger phantom EMPTY_DATA in /api/health even though the payload is fully
 * populated. See ~/.claude/skills/seed-recordcount-autodetect-phantom-empty.
 *
 * Pure function — extracted from runSeed for unit testing.
 */
export function computeRecordCount({ opts = {}, data, payloadBytes = 0, topicArticleCount, onPhantomFallback }) {
  if (opts.recordCount != null) {
    return typeof opts.recordCount === 'function' ? opts.recordCount(data) : opts.recordCount;
  }
  const detectedFromShape = Array.isArray(data)
    ? data.length
    : (topicArticleCount
      ?? data?.predictions?.length
      ?? data?.events?.length ?? data?.earthquakes?.length ?? data?.outages?.length
      ?? data?.fireDetections?.length ?? data?.anomalies?.length ?? data?.threats?.length
      ?? data?.quotes?.length ?? data?.stablecoins?.length
      ?? data?.cables?.length);
  if (detectedFromShape != null) return detectedFromShape;
  if (payloadBytes > 0) {
    if (typeof onPhantomFallback === 'function') onPhantomFallback();
    return 1;
  }
  return 0;
}

/**
 * Significant digits kept in a seeded sparkline series.
 *
 * Yahoo returns closes as float32 values widened to float64, so JSON.stringify emits the
 * conversion noise verbatim: `17.209999084472656` — 18 characters to express 17.21. It is
 * ~50% of every quote payload we seed, and those payloads sit in the bootstrap FAST tier,
 * which takes ~5x more CDN origin misses than the slow tier. Measured 2026-07-14:
 *
 *   market:commodities-bootstrap:v1  241,869 B  — 12,238 noisy floats, 53% of the key
 *   market:stocks-bootstrap:v1       187,834 B  —  7,858 noisy floats, 42% of the key
 *   market:gulf-quotes:v1             57,289 B  —  2,783 noisy floats, 51% of the key
 *
 * SIGNIFICANT digits, not decimal places: commodities carries FX pairs (AUDUSD=X at 0.69),
 * which a fixed 2dp round would flatten into a straight line.
 *
 * 7 chosen by sweeping every live series through the REAL renderer (src/utils/sparkline.ts)
 * and diffing the SVG it emits:
 *
 *   sig │ commodities │ stocks │ worst shift
 *    4  │     36%     │  37%   │  1.90px   <- visibly wrong
 *    6  │     43%     │  46%   │  0.10px
 *    7  │     44%     │  47%   │  0.00px (commodities 31/31 SVG byte-identical)
 *
 * At 7 digits the worst deviation anywhere is 0.10px — exactly ONE unit of the renderer's
 * own `toFixed(1)` coordinate quantum, i.e. the smallest difference the SVG can express, on
 * an 18px-tall chart. Dropping to 6 saves only ~2% more bytes and byte-matches fewer series,
 * so 7 is the better trade.
 *
 * Precision is the ONLY safe lever here. Downsampling was measured and REJECTED: miniSparkline
 * autoscales each series to its own min/max, so dropping any extreme rescales the whole curve
 * — 96-point resampling moved the median series 3-4px and cost USDTRY=X 40% of its vertical
 * range. See tests/sparkline-precision.test.mjs. Commodity sparklines use 5-minute bars to
 * fix their time window (scripts/shared/quote-sparkline.cjs), not to save bytes.
 */
export const SPARKLINE_SIGNIFICANT_DIGITS = 7;

/** Round one value to `sig` significant digits, preserving magnitude across price scales. */
function toSignificantDigits(value, sig) {
  if (typeof value !== 'number' || !Number.isFinite(value) || value === 0) return value;
  // toPrecision returns a string (possibly exponential); Number() normalises it back and
  // drops the trailing zeros, so JSON.stringify emits the shortest form.
  return Number(value.toPrecision(sig));
}

/**
 * Strip float64 conversion noise from a sparkline series. Non-arrays and non-finite entries
 * pass through untouched, so a malformed upstream response degrades exactly as it does today.
 */
export function roundSparkline(values, sig = SPARKLINE_SIGNIFICANT_DIGITS) {
  if (!Array.isArray(values)) return values;
  return values.map((v) => toSignificantDigits(v, sig));
}

/** Decimal places kept for published geographic coordinates. 5 dp is ~1.1m at the equator. */
export const GEO_COORDINATE_DECIMALS = 5;

/**
 * Round one lat/lon to `decimals` places for the PUBLISHED payload.
 *
 * Apply this at the serialization boundary, never at the parse boundary. Rounded
 * coordinates that reach comparison logic shift its decisions: the earthquake
 * cross-agency dedup gates on `haversineDistanceKm(...) <= 10`, and rounding both
 * sides first can move a pair across that threshold (verified: pairs at 9.99977km
 * become 10.00054km, so a duplicate publishes twice — or two distinct events merge).
 * Non-finite values pass through untouched, matching roundSparkline's contract.
 */
export function roundGeoCoordinate(value, decimals = GEO_COORDINATE_DECIMALS) {
  return Number.isFinite(value) ? Number(value.toFixed(decimals)) : value;
}

/**
 * A measured observation from an upstream feed, or null when there isn't one.
 *
 * Statistical and market APIs spell "suppressed", "not yet released" and "no
 * quote" as null, '' or false. Number() turns all three into 0, and 0 is a
 * publishable measurement, so a bare Number() converts missing data into a
 * confident reading of zero. Numeric strings stay valid because several feeds
 * quote their values.
 */
export function finiteObservation(value) {
  if (typeof value !== 'number' && (typeof value !== 'string' || !value.trim())) return null;
  const n = Number(value);
  return Number.isFinite(n) ? n : null;
}

/**
 * `previousClose` replaces the chart's own. fetchYahooQuoteChart passes the
 * last completed session's when Yahoo has rolled the chart to a session that
 * has not traded, where the chart's own equals the price.
 */
export function parseYahooChart(data, symbol, { previousClose = null } = {}) {
  const result = data?.chart?.result?.[0];
  const meta = result?.meta;
  if (!meta) return null;

  // A quote with no price is not a quote. Publishing it produced a market row
  // carrying an undefined price and a change of exactly 0.00%.
  const price = finiteObservation(meta.regularMarketPrice);
  if (price == null) return null;
  // A zero previous close makes the percentage change infinite, so it is
  // treated as unusable here exactly as the previous `||` chain did.
  const prevClose = [previousClose, meta.chartPreviousClose, meta.previousClose]
    .map(finiteObservation)
    .find(value => value != null && value !== 0) ?? price;
  const change = prevClose ? ((price - prevClose) / prevClose) * 100 : 0;
  const closes = result.indicators?.quote?.[0]?.close;
  // NaN survives a != null filter and serializes as null in the published
  // sparkline, leaving a hole in the chart rather than a shorter series.
  const sparkline = roundSparkline(
    Array.isArray(closes) ? closes.map(finiteObservation).filter(v => v != null) : [],
  );

  return { symbol, name: symbol, display: symbol, price, change: +change.toFixed(2), sparkline };
}

/**
 * Decide whether an extra-key write should be skipped to preserve last-good
 * cached data. Opt-in per extra-key via `skipWhenEmpty: true` — when set and the
 * resolved recordCount is 0, runSeed skips the write (and extends the key's TTL)
 * instead of clobbering a good cached payload with an empty recordCount=0 one on
 * a partial upstream fetch. The canonical key is already guarded by validateFn;
 * this closes the same gap for extra keys. Pure function — extracted for tests.
 *
 * A companion extra-key option, `allowMissingOnSkip: true`, marks a key whose
 * ABSENCE is expected rather than alarming — a completion marker that is not
 * written until the final tick of a multi-tick sweep. It downgrades the
 * "manual seed required" warning to an info line on every preservation path,
 * and is meaningless without `skipWhenEmpty`.
 */
export function shouldSkipEmptyExtraKey(ek, recordCount) {
  return Boolean(ek && ek.skipWhenEmpty) && recordCount === 0;
}

/**
 * Hard ceiling for a single seeded extra-key value — a blunt catastrophe backstop,
 * not the primary guard (that is findLeakedPrePublishFields below).
 *
 * Calibrated against a full scan of production (2026-07-14): the largest seeded value
 * in Redis is health:vpd-tracker:realtime:v1 at 3.14 MB, and only three exceed 2 MB.
 * 8 MB therefore leaves ~2.5x headroom over the largest legitimate payload — so ordinary
 * growth cannot crash a healthy seeder — while still refusing the 11.5 MB
 * forecast:predictions-bootstrap:v1 that triggered this guard.
 */
export const MAX_SEEDED_VALUE_BYTES = 8 * 1024 * 1024;

/**
 * Detect an extra key that is re-exporting the seeder's PRE-PUBLISH internals.
 *
 * The trap (production incident, forecast:predictions-bootstrap:v1): runSeed feeds
 * `publishTransform(data)` to the canonical key but feeds RAW `data` to every extraKey
 * transform. A transform written as `{ ...data, <tweak> }` therefore ships the entire
 * internal pipeline state — for seed-forecasts that was fullRunPredictions, inputs,
 * publishSelectionPool, situationClusters, stateUnits, telemetry: an 11.5 MB key, 66x
 * larger than the 172 KB canonical key it was meant to compact.
 *
 * The invariant: `publishTransform` exists precisely to STRIP pre-publish internals from
 * the canonical payload. Any field it dropped, resurfacing in an extra key, is that same
 * internal state escaping through a side door.
 *
 * So we flag exactly the intersection:
 *     present in raw `data`  AND  present in `ekData`  AND  absent from `publishData`
 *
 * Fields an extra key legitimately ADDS (markers like `detailStripped`) never appear in
 * raw `data`, so they are not flagged. Fields the canonical key keeps are not flagged.
 * A seeder that genuinely must re-export a pre-publish field can opt out per-key with
 * `allowPrePublishFields: ['inputs']`.
 *
 * Pure function — extracted for tests.
 *
 * @returns {string[]} leaked top-level field names (empty when clean)
 */
export function findLeakedPrePublishFields(rawData, publishData, ekData, ek = {}) {
  const isPlain = (v) => v && typeof v === 'object' && !Array.isArray(v);
  // No publishTransform → publishData IS rawData → nothing was stripped → nothing to leak.
  if (!isPlain(rawData) || !isPlain(publishData) || !isPlain(ekData)) return [];
  if (rawData === publishData) return [];

  const allowed = new Set(ek.allowPrePublishFields || []);
  const canonicalFields = new Set(Object.keys(publishData));

  return Object.keys(ekData).filter((field) => (
    !canonicalFields.has(field)
    && Object.hasOwn(rawData, field)
    && !allowed.has(field)
  ));
}

// Fleet-wide graceful-degradation backstop (issue #4786). A non-settling
// await inside a seeder's fetchFn — e.g. an unguarded R2/S3 body stream whose
// socket is silently reaped — never rejects, so the Phase-1 try/catch can't
// catch it; the event loop drains and Node exits 13 ("Detected unsettled
// top-level await"), painting a red Railway badge that is neither a graceful
// skip nor a catchable failure. Racing the fetch against a wall-clock deadline
// converts that hang into a normal rejection, which the existing graceful path
// turns into exit 75 (TTL extended, last-good served, no data lost).
//
// The standalone deadline is tied to lockTtlMs — never a fixed value — because
// seeders legitimately run from ~1min to 40min. A healthy seeder is designed
// never to outlive its own lock, so lockTtlMs + margin exceeds any legitimate
// run; the only thing that trips it is a genuine hang. A false trip is itself
// graceful (exit 75), so the margin errs generous.
//
// When spawned as a bundle section, that lock-derived ceiling can outlast the
// runner's section timeoutMs (#8479). resolveFetchDeadlineMs then clamps the
// fetch deadline to leave FETCH_PHASE_PUBLISH_RESERVE_MS for publish or
// graceful cleanup before the runner SIGTERMs.
export const FETCH_PHASE_DEADLINE_MARGIN_MS = 120_000;

// Time left between the fetch-phase deadline and the bundle section timeout so
// publish (success) or releaseLock + TTL extend (graceful fetch failure) can
// finish before `_bundle-runner` sends SIGTERM. Matches the 40s headroom used
// by education-attainment and cross-strait activity seeders.
export const FETCH_PHASE_PUBLISH_RESERVE_MS = 40_000;

export function raceFetchDeadline(promise, ms, label) {
  let timer;
  const guard = new Promise((_, reject) => {
    timer = setTimeout(
      () => reject(new Error(`${label} fetch phase exceeded ${ms}ms deadline (likely a non-settling await — see issue #4786)`)),
      ms,
    );
  });
  return Promise.race([promise, guard]).finally(() => clearTimeout(timer));
}

/**
 * Resolve the fetch-phase wall-clock budget for runSeed.
 *
 * Standalone: explicit `fetchPhaseTimeoutMs`, else `lockTtlMs + margin`.
 * Bundle section: also clamp to `sectionTimeoutMs - publish reserve` so the
 * graceful path is reachable before the runner's SIGTERM (#8479).
 *
 * @param {{
 *   fetchPhaseTimeoutMs?: number | null,
 *   lockTtlMs: number,
 *   sectionTimeoutMs?: number | null,
 * }} opts
 * @returns {number}
 */
export function resolveFetchDeadlineMs({
  fetchPhaseTimeoutMs,
  lockTtlMs,
  sectionTimeoutMs = null,
}) {
  const configured = Number.isFinite(fetchPhaseTimeoutMs) && fetchPhaseTimeoutMs > 0
    ? fetchPhaseTimeoutMs
    : lockTtlMs + FETCH_PHASE_DEADLINE_MARGIN_MS;
  if (!Number.isFinite(sectionTimeoutMs) || sectionTimeoutMs <= 0) {
    return configured;
  }
  // A section shorter than the reserve still needs a positive race target so
  // hang detection fires rather than waiting forever for SIGTERM.
  const sectionCap = Math.max(1, sectionTimeoutMs - FETCH_PHASE_PUBLISH_RESERVE_MS);
  return Math.min(configured, sectionCap);
}

// Set by _bundle-runner for canonical-clock members that need proof that every
// publish side effect completed. Standalone seed runs leave it unset.
export const BUNDLE_COMPLETION_META_KEY_ENV = 'WM_BUNDLE_COMPLETION_META_KEY';

// Set by _bundle-runner to the section's timeoutMs so runSeed can clamp its
// fetch deadline inside the wall clock that will SIGTERM the child (#8479).
export const BUNDLE_SECTION_TIMEOUT_MS_ENV = 'BUNDLE_SECTION_TIMEOUT_MS';

/**
 * Section timeoutMs injected by `_bundle-runner` for the current child.
 * Standalone seed runs leave it unset.
 *
 * @returns {number | null}
 */
export function getBundleSectionTimeoutMs() {
  const raw = Number(process.env[BUNDLE_SECTION_TIMEOUT_MS_ENV]);
  return Number.isFinite(raw) && raw > 0 ? raw : null;
}

export async function runSeed(domain, resource, canonicalKey, fetchFn, opts = {}) {
  const {
    validateFn,
    ttlSeconds,
    maxPayloadBytes,
    lockTtlMs = 120_000,
    lockAcquireRetries = 2,
    extraKeys,
    // Keys written outside runSeed's normal extra-key phase that still need
    // last-good TTL protection when the primary fetch fails or is skipped.
    preserveKeys = [],
    // Some sources must keep their original absolute expiry after a failed
    // refresh. Opt out of renewal without changing the stored value or clock.
    extendTtlOnFailure = true,
    // Opt-in companion keys whose write TTL differs from the canonical key.
    // Each declaration is preserved at its own TTL; preserveKeys keeps its
    // existing canonical-TTL behavior for backward compatibility.
    preserveKeyTtls = [],
    // Standalone-cron exit policy. Railway shows exit 75 as a crash, so a
    // seeder that can serve last-good through a short upstream outage may opt
    // in to exit 0 on a fetch failure. All of these must hold, else it exits 75:
    //   - the error is tagged with markPreservableFetchFailure() (an upstream
    //     outage, never a missing credential or a code bug);
    //   - Redis confirms the canonical key and seed-meta TTLs were extended;
    //   - seed-meta fetchedAt is within maxStaleMin, so a chronic outage goes
    //     back to exit 75 once health would call the data stale;
    //   - the run is standalone. Under _bundle-runner exit 0 means "keys were
    //     written" (#6396), so bundle members always keep exit 75.
    // Requires maxStaleMin. seed-meta is never refreshed on this path.
    allowPreservedFetchFailure = false,
    beforePublish,
    publishAtomically,
    compressPayload = false, // store gzip+base64 (see _seed-payload-codec.mjs) when JSON outgrows MAX_PAYLOAD_BYTES
    afterPublish,
    afterValidationSkip,
    afterPreservedValidationSkip,
    afterFreshness,
    publishTransform,
    declareRecords,        // new — contract opt-in. When present, runSeed enters
                           // envelope-dual-write path: writes `{_seed, data}` to
                           // canonicalKey alongside legacy `seed-meta:*` key.
    sourceVersion,         // new — required when declareRecords is passed
    schemaVersion,         // new — required when declareRecords is passed
    zeroIsValid = false,   // new — when true, recordCount=0 is OK_ZERO, not RETRY
    contentMeta,           // (rawData, runStartedAtMs) => {newestItemAt, oldestItemAt} | null
    maxContentAgeMin,      // positive integer minutes — opts in together with contentMeta
    fetchPhaseTimeoutMs,   // hard ceiling on the fetch phase; defaults to lockTtlMs + margin (#4786)
  } = opts;
  const contractMode = typeof declareRecords === 'function';
  if (extraKeys && !Array.isArray(extraKeys)) {
    console.error(`  CONTRACT VIOLATION: ${domain}:${resource} extraKeys must be an array`);
    process.exit(1);
  }
  // A colliding or un-namespaced extraKey meta key would be caught by
  // writeSeedMeta, but only after the provider fetches and the data write.
  // Refuse it at config time instead, before any upstream call is spent.
  for (const ek of Array.isArray(extraKeys) ? extraKeys : []) {
    if (ek?.metaKey === undefined || ek?.metaKey === null) continue;
    try {
      resolveSeedMetaKey(ek.key, ek.metaKey);
    } catch (err) {
      console.error(`  CONTRACT VIOLATION: ${domain}:${resource} ${err.message}`);
      process.exit(1);
    }
  }
  if (afterPublish && typeof afterPublish !== 'function') {
    console.error(`  CONTRACT VIOLATION: ${domain}:${resource} afterPublish must be a function`);
    process.exit(1);
  }
  if (publishAtomically && typeof publishAtomically !== 'function') {
    console.error(`  CONTRACT VIOLATION: ${domain}:${resource} publishAtomically must be a function`);
    process.exit(1);
  }
  if (afterFreshness && typeof afterFreshness !== 'function') {
    console.error(`  CONTRACT VIOLATION: ${domain}:${resource} afterFreshness must be a function`);
    process.exit(1);
  }
  if (allowPreservedFetchFailure && !(Number.isFinite(opts.maxStaleMin) && opts.maxStaleMin > 0)) {
    console.error(
      `  CONTRACT VIOLATION: ${domain}:${resource} allowPreservedFetchFailure requires a positive maxStaleMin`,
    );
    process.exit(1);
  }
  const bundleCompletionMetaKey = String(process.env[BUNDLE_COMPLETION_META_KEY_ENV] ?? '').trim();
  if (bundleCompletionMetaKey) {
    if (!contractMode) {
      console.error(`  CONTRACT VIOLATION: ${domain}:${resource} bundle completion attestation requires contract mode`);
      process.exit(1);
    }
    if (!bundleCompletionMetaKey.startsWith('seed-completion:')) {
      console.error(
        `  CONTRACT VIOLATION: ${domain}:${resource} bundle completion key must use the dedicated `
        + `seed-completion: namespace, got ${bundleCompletionMetaKey}`,
      );
      process.exit(1);
    }
  }
  if (contractMode) {
    // Soft-warn (PR 2) on other mandatory contract fields; PR 3 hard-aborts.
    const missing = [];
    if (typeof sourceVersion !== 'string' || sourceVersion.trim() === '') missing.push('sourceVersion');
    if (!Number.isInteger(schemaVersion) || schemaVersion < 1) missing.push('schemaVersion');
    if (typeof opts.maxStaleMin !== 'number') missing.push('maxStaleMin');
    if (missing.length) {
      console.warn(`  [seed-contract] ${domain}:${resource} missing fields: ${missing.join(', ')} — required in PR 3`);
    }
  }
  // Content-age contract validation (2026-05-04 health-readiness plan).
  // contentMeta and maxContentAgeMin opt in TOGETHER. Hard-fail at config time
  // on misconfig — silently disabling the check would defeat the alarm.
  const contentAgeOptedIn = contentMeta != null || maxContentAgeMin != null;
  if (contentAgeOptedIn) {
    if (typeof contentMeta !== 'function') {
      console.error(`  CONTRACT VIOLATION: ${domain}:${resource} declares maxContentAgeMin without contentMeta function`);
      process.exit(1);
    }
    if (!Number.isInteger(maxContentAgeMin) || maxContentAgeMin <= 0) {
      console.error(`  CONTRACT VIOLATION: ${domain}:${resource} maxContentAgeMin must be a positive integer (minutes), got ${JSON.stringify(maxContentAgeMin)}`);
      process.exit(1);
    }
  }
  if (!Array.isArray(preserveKeyTtls)) {
    console.error(`  CONTRACT VIOLATION: ${domain}:${resource} preserveKeyTtls must be an array`);
    process.exit(1);
  }
  const normalizedPreserveKeyTtls = [];
  const declaredTtlByKey = new Map();
  for (const declaration of preserveKeyTtls) {
    const key = declaration?.key;
    const declaredTtl = declaration?.ttlSeconds;
    if (
      !declaration
      || typeof declaration !== 'object'
      || Array.isArray(declaration)
      || typeof key !== 'string'
      || key.trim().length === 0
      || !Number.isInteger(declaredTtl)
      || declaredTtl <= 0
    ) {
      console.error(
        `  CONTRACT VIOLATION: ${domain}:${resource} preserveKeyTtls entries must be `
        + `{ key: non-empty string, ttlSeconds: positive integer }`,
      );
      process.exit(1);
    }
    if (declaredTtlByKey.has(key) && declaredTtlByKey.get(key) !== declaredTtl) {
      console.error(
        `  CONTRACT VIOLATION: ${domain}:${resource} preserveKeyTtls declares conflicting TTLs for ${key}`,
      );
      process.exit(1);
    }
    declaredTtlByKey.set(key, declaredTtl);
    normalizedPreserveKeyTtls.push({ key, ttlSeconds: declaredTtl });
  }
  const runId = `${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
  const startMs = Date.now();
  const defaultPreservationTtl = ttlSeconds || 600;
  const defaultPreservationKeys = [...new Set([
    canonicalKey,
    `seed-meta:${domain}:${resource}`,
    ...(extraKeys || []).map((ek) => ek.key),
    ...preserveKeys,
  ].filter((key) => typeof key === 'string' && key.length > 0))];

  // Extra keys whose absence is expected rather than alarming (see
  // shouldSkipEmptyExtraKey). Threaded into EVERY preservation call, not just
  // the empty-skip branch: the fetch-failure, SIGTERM, contract-RETRY,
  // atomic-publish-failure and validation-skip paths all preserve the same key
  // set, and warning "manual seed required" for a marker that is not supposed
  // to exist yet is the false alarm allowMissingOnSkip exists to remove.
  const optionalPreservationKeys = (extraKeys || [])
    .filter((ek) => ek && ek.allowMissingOnSkip)
    .map((ek) => ek.key)
    .filter((key) => typeof key === 'string' && key.length > 0);

  // Single preservation seam for fetch failure, fetch-phase SIGTERM, contract
  // RETRY, validation skip, and per-extra-key empty skips. Grouping keys by TTL
  // keeps one Redis pipeline per TTL while allowing explicit declarations to
  // override a key that also appears in the default canonical-TTL cohort.
  const preserveExistingKeys = async (targets) => {
    if (extendTtlOnFailure === false) return false;
    const ttlByKey = new Map();
    if (targets) {
      for (const target of targets) ttlByKey.set(target.key, target.ttlSeconds);
    } else {
      for (const key of defaultPreservationKeys) ttlByKey.set(key, defaultPreservationTtl);
      for (const target of normalizedPreserveKeyTtls) ttlByKey.set(target.key, target.ttlSeconds);
    }

    const keysByTtl = new Map();
    for (const [key, targetTtl] of ttlByKey) {
      if (!keysByTtl.has(targetTtl)) keysByTtl.set(targetTtl, []);
      keysByTtl.get(targetTtl).push(key);
    }
    const results = await Promise.all(
      [...keysByTtl].map(([targetTtl, keys]) => extendExistingTtl(keys, targetTtl, {
        allowMissingKeys: optionalPreservationKeys,
      })),
    );
    return results.every(Boolean);
  };

  console.log(`=== ${domain}:${resource} Seed ===`);
  console.log(`  Run ID:  ${runId}`);
  console.log(`  Key:     ${canonicalKey}`);
  if (contractMode) console.log(`  Mode:    contract (envelope dual-write)`);

  // Acquire lock
  const lockResult = await acquireLockSafely(`${domain}:${resource}`, runId, lockTtlMs, {
    label: `${domain}:${resource}`,
    maxRetries: lockAcquireRetries,
  });
  if (lockResult.skipped) {
    process.exit(0);
  }
  if (!lockResult.locked) {
    console.log('  SKIPPED: another seed run in progress');
    process.exit(0);
  }

  // SM-122: the fetch deadline is lockTtlMs + 120s, so a run can outlive its
  // lock, and another run can then take it and publish newer data. Renew the
  // lock while this run works, and fence every publish point on ownership:
  // the canonical SET and seed-meta SET run inside Redis only while this run
  // still holds the lock (SEED_LOCK_FENCED_SET_SCRIPT), and the publish phase
  // re-checks ownership before its first write. A run that lost the lock
  // exits without publishing and without touching seed-meta.
  const lockDomain = `${domain}:${resource}`;
  const lockHeartbeat = startLockHeartbeat(lockDomain, runId, lockTtlMs);
  const lockFence = {
    lockKey: `seed-lock:${lockDomain}`,
    runId,
    ttlMs: lockTtlMs,
    renew: () => renewLockIfHeld(lockDomain, runId, lockTtlMs),
  };
  // Cheap local check between Redis-fenced writes: the heartbeat already
  // learned the lock is gone, so stop before the next write group.
  const assertLockStillHeld = () => {
    if (lockHeartbeat.state.lost) throw lockHeartbeat.state.lost;
  };

  // SIGTERM handler — installed BEFORE fetch and KEPT installed through
  // publish. _bundle-runner.mjs sends SIGTERM when a section's timeout
  // fires, then SIGKILL after KILL_GRACE_MS (5s). Without a publish-phase
  // handler, a timeout that fires during atomicPublish or extendExistingTtl
  // leaves seed-lock:<domain>:<resource> dangling for the full lockTtlMs
  // (default 120s). For seeders bundled in fast-firing crons (e.g.
  // seed-bis-lbs.mjs in seed-bundle-macro.mjs) the next tick can collide
  // with that orphaned lock and SKIP repeatedly — the canonical key never
  // gets published and /api/health reports `EMPTY`.
  //
  // The handler is phase-aware so it preserves the strict-floor invariant
  // (emptyDataIsFailure seeders MUST NOT refresh seed-meta on validation
  // reject — see imf-external Railway log 2026-04-13). During fetch we
  // release lock + extend existing-data TTL so consumers keep seeing
  // last-good. During publish we release lock ONLY: data was fetched but
  // not yet stored; refreshing TTL here would silently re-anchor stale
  // data and corrupt the strict-floor retry path.
  //
  // Releases run in parallel (disjoint keys; serializing compounds Upstash
  // latency during the exact failure mode this handler exists to handle).
  // Exit 143 = POSIX convention for SIGTERM-terminated process.
  let currentPhase = 'fetch';
  const sigTermHandler = async () => {
    console.error(`  [${domain}:${resource}] SIGTERM received during ${currentPhase} phase — releasing lock runId=${runId}`);
    lockHeartbeat.stop();
    try {
      if (currentPhase === 'fetch') {
        await Promise.allSettled([
          releaseLock(`${domain}:${resource}`, runId),
          preserveExistingKeys(),
        ]);
      } else {
        await releaseLock(`${domain}:${resource}`, runId);
      }
    } catch (err) {
      console.error(`  [${domain}:${resource}] SIGTERM cleanup error: ${err?.message || err}`);
    } finally {
      // process.exit does not drain in-flight promises — flush any
      // fire-and-forget llm_call telemetry (bounded by its 1.5s fetch
      // timeout; the runner's 5s SIGKILL grace leaves room).
      try { await flushPendingLlmEvents(); } catch { /* never block exit */ }
      process.exit(143);
    }
  };
  process.once('SIGTERM', sigTermHandler);

  // Phase 1: Fetch data (graceful on failure — extend TTL on stale data).
  // Raced against a wall-clock deadline so a non-settling await inside fetchFn
  // (see raceFetchDeadline above, issue #4786) surfaces as a catchable
  // rejection instead of hanging the process into an exit-13 red badge.
  // When spawned by the bundle runner, also clamp to the section timeout so
  // this graceful path fires before the runner's SIGTERM (#8479).
  const sectionTimeoutMs = getBundleSectionTimeoutMs();
  const unconstrainedDeadlineMs = resolveFetchDeadlineMs({
    fetchPhaseTimeoutMs,
    lockTtlMs,
    sectionTimeoutMs: null,
  });
  const fetchDeadlineMs = resolveFetchDeadlineMs({
    fetchPhaseTimeoutMs,
    lockTtlMs,
    sectionTimeoutMs,
  });
  if (sectionTimeoutMs != null && fetchDeadlineMs < unconstrainedDeadlineMs) {
    console.warn(
      `  [${domain}:${resource}] fetch deadline clamped ${unconstrainedDeadlineMs}ms → ${fetchDeadlineMs}ms `
      + `to fit bundle section timeout ${sectionTimeoutMs}ms (issue #8479)`,
    );
  }
  let data;
  // Absolute wall-clock deadline of the fetch race, for fetchers with optional
  // work that must stop in time to publish what they already have.
  const fetchDeadlineAtMs = Date.now() + fetchDeadlineMs;
  try {
    data = await raceFetchDeadline(
      withRetry(() => fetchFn({ runId, runStartedAtMs: startMs, fetchDeadlineAtMs })),
      fetchDeadlineMs,
      `${domain}:${resource}`,
    );
  } catch (err) {
    // Keep the SIGTERM handler installed across the fetch-failure
    // cleanup. Earlier code did `process.off('SIGTERM', sigTermHandler)`
    // here, which opened a new leak window: SIGTERM during the
    // releaseLock + extendExistingTtl awaits below would fall through
    // to Node's default termination and could strand seed-lock or skip
    // the TTL extension. Both paths (this catch's manual ops and the
    // handler's parallel ops) are idempotent — the LUA verify-and-DEL
    // releases at most once for a given runId, and EXPIRE pipelines on
    // existing keys are safely re-runnable — so a race between the
    // catch path and the handler converges on the correct end state.
    // process.exit below terminates before any pending SIGTERM can fire
    // on the success path of cleanup.
    await releaseLock(`${domain}:${resource}`, runId);
    const durationMs = Date.now() - startMs;
    const cause = err.cause ? ` (cause: ${err.cause.message || err.cause.code || err.cause})` : '';
    console.error(`  FETCH FAILED: ${err.message || err}${cause}`);

    if (opts.recordInitialFetchFailure === true) {
      try {
        const { url, token } = getRedisCredentials();
        await redisCommand(url, token, ['SET', `seed-meta:${domain}:${resource}`, JSON.stringify({
          status: 'error', firstFailureAt: startMs, recordCount: 0, runId,
        }), 'NX']);
      } catch {
        console.warn(`  [${domain}:${resource}] initial failure receipt could not be stored`);
      }
    }

    const preserved = await preserveExistingKeys();

    if (
      allowPreservedFetchFailure === true
      && err?.preservableFetchFailure === true
      && preserved
      && sectionTimeoutMs == null
    ) {
      const ageMin = await readSeedMetaAgeMin(`seed-meta:${domain}:${resource}`);
      if (ageMin != null && ageMin <= opts.maxStaleMin) {
        // Distinct marker: this tick failed, but it is not the exit-75
        // "Failed gracefully" outcome that log triage counts as a crash.
        console.log(`\n=== Refresh failed; last-good retained (${Math.round(durationMs)}ms) ===`);
        console.warn(
          `  [${domain}:${resource}] last-good is ${Math.round(ageMin)}min old (maxStaleMin ${opts.maxStaleMin}); `
          + 'exiting 0 without refreshing seed-meta',
        );
        await exitAfterTelemetryFlush(0);
      }
      console.warn(
        `  [${domain}:${resource}] last-good age ${ageMin == null ? 'unknown' : `${Math.round(ageMin)}min`} `
        + `is not within maxStaleMin ${opts.maxStaleMin}; exiting ${GRACEFUL_FETCH_FAILURE_EXIT_CODE}`,
      );
    }

    console.log(`\n=== Failed gracefully (${Math.round(durationMs)}ms) ===`);
    await exitAfterTelemetryFlush(GRACEFUL_FETCH_FAILURE_EXIT_CODE);
  }
  // Transition to publish phase — handler stays installed but switches
  // behavior via the phase tracker.
  currentPhase = 'publish';

  // Phase 2: Publish to Redis (rethrow on failure — data was fetched but not stored)
  try {
    // Ownership gate before any publish-phase write (canonical, extra keys,
    // hooks, seed-meta, TTL preservation). Renews the lease on success.
    await lockFence.renew();

    // Content-age contract: invoke contentMeta on RAW fetcher output BEFORE
    // publishTransform runs. This lets seeders carry pre-publish helper fields
    // (e.g. _publishedAtIsSynthetic) on items that contentMeta reads, then
    // strip them via publishTransform before they reach the canonical key
    // and downstream clients. See the 2026-05-04 health-readiness plan,
    // Sprint 1 / Sprint 2 disease-outbreaks pilot.
    //
    // contentMeta returning null OR throwing both signal "no usable item
    // timestamps" → write newestItemAt: null in the envelope, which the
    // health classifier reads as STALE_CONTENT.
    let contentNewestAt = null;
    let contentOldestAt = null;
    if (contentAgeOptedIn) {
      try {
        const result = contentMeta(data, startMs);
        if (result && typeof result === 'object'
            && Number.isFinite(result.newestItemAt) && result.newestItemAt > 0
            && Number.isFinite(result.oldestItemAt) && result.oldestItemAt > 0) {
          contentNewestAt = result.newestItemAt;
          contentOldestAt = result.oldestItemAt;
        }
      } catch (err) {
        console.warn(`  [content-age] ${domain}:${resource}: contentMeta threw, treating as null: ${err?.message || err}`);
      }
    }

    const publishData = publishTransform ? publishTransform(data) : data;

    // In contract mode, resolve recordCount from declareRecords BEFORE publish so
    // the envelope carries the correct state. RETRY-on-empty paths skip the
    // publish entirely (leaving the previous envelope in place).
    let contractState = null;   // 'OK' | 'OK_ZERO' | 'RETRY'
    let contractRecordCount = null;
    let envelopeMeta = null;
    if (contractMode) {
      try {
        contractRecordCount = resolveRecordCount(declareRecords, publishData);
      } catch (err) {
        // Contract violation — declareRecords returned non-int / threw. HARD FAIL.
        await releaseLock(`${domain}:${resource}`, runId);
        console.error(`  CONTRACT VIOLATION: ${err.message || err}`);
        await exitAfterTelemetryFlush(1);
      }
      if (contractRecordCount > 0) {
        contractState = 'OK';
      } else if (zeroIsValid) {
        contractState = 'OK_ZERO';
      } else {
        contractState = 'RETRY';
      }
      if (contractState !== 'RETRY') {
        envelopeMeta = {
          fetchedAt: Date.now(),
          recordCount: contractRecordCount,
          sourceVersion: sourceVersion || '',
          schemaVersion: schemaVersion || 1,
          state: contractState,
        };
        // Carry content-age fields when seeder opted in. Presence of
        // maxContentAgeMin in the envelope is the opt-in signal for the
        // health classifier. newestItemAt/oldestItemAt may be explicit null
        // when contentMeta returned null OR all items lacked usable
        // timestamps — classifier reads those as STALE_CONTENT.
        if (contentAgeOptedIn) {
          envelopeMeta.newestItemAt = contentNewestAt;
          envelopeMeta.oldestItemAt = contentOldestAt;
          envelopeMeta.maxContentAgeMin = maxContentAgeMin;
        }
      }
    }

    // Contract RETRY on empty (no zeroIsValid) — skip publish and preserve the
    // last-good keys. Exit 0 when Redis confirms every key was actually extended;
    // otherwise the data is already gone (or preservation could not be verified), so a
    // green process would hide a live outage indefinitely — EXCEPT when the seeder
    // declared `sourceUnavailable`, which also exits 0 (see #5256 below: with no source
    // configured, no retry can ever restore the data, so exiting 1 crash-loops forever
    // and /api/health already carries the alarm).
    if (contractState === 'RETRY') {
      const durationMs = Date.now() - startMs;
      const preserved = await preserveExistingKeys();

      // #5256: the RETRY-FAILED exit below assumes A LATER TICK CAN RESTORE THE DATA. When
      // the seeder reports it had no usable source at all (primary unconfigured AND every
      // fallback down), no tick ever can — seed-conflict-intel crash-looped every ~15min
      // forever, firing "Deploy Crashed!" each time while /api/health already reported the
      // domain EMPTY/crit. The crash added nothing over health; it only trained us to ignore
      // the crash channel. Stay green, publish NOTHING (an empty envelope would overwrite
      // last-good the moment the source blips), and leave the data alarm to /api/health.
      //
      // This is deliberately NOT `zeroIsValid`: a seeder must opt in per-run, on the exact
      // code path where it knows it has no source. A zero-yield run that does not declare
      // sourceUnavailable is still a dead feed and still exits 1 — #5258 stands.
      //
      // Checked BEFORE `preserved`, not inside `!preserved`: the outcome is exit 0 either
      // way, but an operator must see the REAL reason on every tick. Falling through to the
      // generic "TTL extended, bundle will retry next cycle" message while last-good is
      // still alive would hide the no-source condition for however many cycles the keys
      // survive, and only surface it once they expire.
      if (data?.sourceUnavailable) {
        console.warn(
          `  NO SOURCE: declareRecords returned 0 and no usable upstream was available — published nothing, `
          + (preserved
            ? `last-good TTL extended (data still served, but it is no longer being refreshed).`
            : `and last-good has already expired — /api/health reports ${domain}:${resource} EMPTY.`),
        );
        console.log(`\n=== Done (${Math.round(durationMs)}ms, NO SOURCE) ===`);
        await releaseLock(`${domain}:${resource}`, runId);
        await exitAfterTelemetryFlush(0);
      }

      if (!preserved) {
        console.error(`  FAILURE: declareRecords returned 0 and last-good preservation failed — one or more keys are missing or could not be extended`);
        console.log(`\n=== Done (${Math.round(durationMs)}ms, RETRY FAILED) ===`);
        await releaseLock(`${domain}:${resource}`, runId);
        await exitAfterTelemetryFlush(1);
      }
      console.log(`  RETRY: declareRecords returned 0 (zeroIsValid=false) — envelope unchanged, TTL extended, bundle will retry next cycle`);
      console.log(`\n=== Done (${Math.round(durationMs)}ms, RETRY) ===`);
      await releaseLock(`${domain}:${resource}`, runId);
      await exitAfterTelemetryFlush(0);
    }

    let publishResult;
    try {
      publishResult = await atomicPublish(canonicalKey, publishData, validateFn, ttlSeconds, {
        envelopeMeta,
        maxPayloadBytes,
        beforePublish: beforePublish
          ? () => beforePublish(data, { canonicalKey, ttlSeconds, runId, lockFence })
          : undefined,
        publishAtomically: publishAtomically
          ? (publishContext) => publishAtomically(data, { ...publishContext, runId })
          : undefined,
        compressPayload,
        lockFence,
      });
    } catch (error) {
      // An atomic publisher either switches its complete key cohort or leaves
      // the prior cohort untouched. If staging or the final switch fails,
      // extend that untouched last-good cohort before surfacing the failure.
      // Validation skips are handled below and retain their stricter policy.
      // A thrown `beforePublish` is the same shape: it runs ahead of every
      // canonical/extra write, so the prior cohort is likewise untouched and the
      // deep coverage rejections that live there must not cost last-good TTLs.
      // A lost seed lock goes straight to the lock-lost exit below: the keys
      // now belong to the run that holds the lock.
      if (isSeedLockLostError(error)) throw error;
      if (publishAtomically || beforePublish) {
        const preserved = await preserveExistingKeys().catch(() => false);
        if (!preserved) {
          console.error(`  FAILURE: atomic publish failed and last-good preservation was incomplete`);
        }
      }
      throw error;
    }
    if (publishResult.skipped) {
      const durationMs = Date.now() - startMs;
      const preserved = await preserveExistingKeys();
      const strictFailure = Boolean(opts.emptyDataIsFailure);
      const validationSkipContext = {
        canonicalKey,
        ttlSeconds,
        recordCount: contractRecordCount,
        runId,
        preservationSucceeded: preserved,
      };
      // Some rejected snapshots carry failure evidence that must survive even
      // when there is no complete last-good cohort to preserve. Keep this hook
      // in the publish phase so its persistence cannot make withRetry(fetchFn)
      // repeat upstream source requests. A hook may return
      // `{ freshnessMetaPatch }`; reserved freshness fields are stripped before
      // the shared writer applies the patch.
      let validationSkipResult = null;
      let validationSkipMetaRead = false;
      let validationSkipExistingMeta = null;
      if (!strictFailure && afterValidationSkip) {
        validationSkipExistingMeta = await readExistingSeedMeta(domain, resource);
        validationSkipMetaRead = true;
        assertLockStillHeld();
        validationSkipResult = await afterValidationSkip(data, {
          ...validationSkipContext,
          existingSeedMeta: validationSkipExistingMeta,
        });
      }
      if (strictFailure) {
        // Strict-floor seeders (e.g. IMF-External, floor=180 countries) treat
        // empty data as a real upstream failure. Do NOT refresh seed-meta —
        // letting fetchedAt stay stale lets bundles retry on their next cron
        // fire and lets health flip to STALE_SEED. Writing fresh meta here
        // caused imf-external to skip for the full 30-day interval after a
        // single transient failure (Railway log 2026-04-13).
        console.error(`  FAILURE: validation failed (empty data) — seed-meta NOT refreshed; bundle will retry next cycle`);
      } else {
        // Write seed-meta even when data is empty so health can distinguish
        // "seeder ran but nothing to publish" from "seeder stopped" (quiet-
        // period feeds: news, events, sparse indicators).
        //
        // BUT — when the canonical key still holds last-good contract-mode
        // data with recordCount > 0, mirror its (fetchedAt, recordCount)
        // into seed-meta instead of writing zero. This keeps /api/health
        // reporting an accurate count when validateFn rejects a transient
        // upstream blip (e.g. WB late-reporter variation that drops a
        // resilience indicator from 153 → 149 countries when the floor was
        // 150 — production incident 2026-05-03 for resilience:power-losses
        // where canonical had 216 countries but seed-meta got overwritten
        // with 0 → EMPTY_DATA). The mirrored fetchedAt is canonical's
        // ORIGINAL fetch time, NOT now, so STALE_SEED still fires naturally
        // once the canonical data ages past maxStaleMin — preserving the
        // strict-floor honesty WITHOUT punishing a transient blip with a
        // misleading zero.
        //
        // Falls back to writing 0 (legacy quiet-period behavior) when:
        //   - canonical key is missing
        //   - canonical envelope is malformed / legacy bare shape
        //   - canonical envelope has recordCount <= 0
        const canonicalMeta = await readCanonicalEnvelopeMeta(canonicalKey);
        // Preserve non-reserved diagnostics (poolCounts, GDELT errorReason, …)
        // from the previous seed-meta write. The SET below replaces the whole
        // key; without this merge a validate-skip after a healthy publish wipes
        // afterPublish patches and fail-closed consumers false-alarm.
        const currentSkipDiagnostics =
          freshnessMetaDiagnosticsPatch(validationSkipResult?.freshnessMetaPatch) || {};
        const preservedDiagnostics = {
          ...(freshnessMetaDiagnosticsPatch(
            validationSkipMetaRead
              ? validationSkipExistingMeta
              : await readExistingSeedMeta(domain, resource),
          ) || {}),
          ...currentSkipDiagnostics,
        };
        if (canonicalMeta) {
          // Pass-through canonical's contentAge so health doesn't lose the
          // STALE_CONTENT signal exactly when last-good-with-stale-content
          // data is being served (Codex round 1 P0b).
          await writeFreshnessMetadataSafely(
            domain, resource, canonicalMeta.recordCount,
            canonicalMeta.sourceVersion || opts.sourceVersion,
            ttlSeconds,
            canonicalMeta.fetchedAt,
            canonicalMeta.contentAge,
            Object.keys(preservedDiagnostics).length > 0 ? preservedDiagnostics : null,
            { lockFence },
          );
          console.log(
            `  SKIPPED: validation failed (empty/partial fetch) — seed-meta mirrors canonical ` +
            `(fetchedAt=${new Date(canonicalMeta.fetchedAt).toISOString()}, recordCount=${canonicalMeta.recordCount}` +
            `${canonicalMeta.contentAge ? `, newestItemAt=${canonicalMeta.contentAge.newestItemAt == null ? 'null' : new Date(canonicalMeta.contentAge.newestItemAt).toISOString()}` : ''}); ` +
            `existing cache TTL extended`,
          );
        } else {
          // No non-empty last-good envelope: drop prior diagnostics because
          // they described a different cohort, but retain diagnostics emitted
          // by this rejected attempt. A valid zero-record predecessor can
          // still carry bounded source-failure evidence.
          await writeFreshnessMetadataSafely(
            domain, resource, 0, opts.sourceVersion, ttlSeconds,
            undefined, undefined,
            Object.keys(currentSkipDiagnostics).length > 0 ? currentSkipDiagnostics : null,
            { lockFence },
          );
          console.log(`  SKIPPED: validation failed (empty data) — seed-meta refreshed (recordCount=0), existing cache TTL extended`);
        }
      }
      if (!strictFailure && preserved && afterPreservedValidationSkip) {
        await afterPreservedValidationSkip(data, validationSkipContext);
      }
      console.log(`\n=== Done (${Math.round(durationMs)}ms, no write) ===`);
      await releaseLock(`${domain}:${resource}`, runId);
      // Strict path exits non-zero so _bundle-runner counts it as failed++
      // (otherwise the bundle summary hides upstream outages behind ran++).
      await exitAfterTelemetryFlush(strictFailure ? 1 : 0);
    }
    const { payloadBytes } = publishResult;
    const topicArticleCount = Array.isArray(data?.topics)
      ? data.topics.reduce((n, t) => n + (t?.articles?.length || t?.events?.length || 0), 0)
      : undefined;
    const recordCount = contractMode
      ? contractRecordCount
      : computeRecordCount({
          opts, data, payloadBytes, topicArticleCount,
          onPhantomFallback: () => console.warn(
            `  [recordCount] auto-detect did not match a known shape (payloadBytes=${payloadBytes}); falling back to 1. Add opts.recordCount to ${domain}:${resource} for accurate health metrics.`
          ),
        });

    // Write extra keys (e.g., bootstrap hydration keys). In contract mode each
    // extra key gets its own envelope; declareRecords may be per-key or reuse
    // the canonical one.
    if (extraKeys) {
      for (const ek of extraKeys) {
        // Stop at the first key once the heartbeat has seen the lock go; each
        // write below is also fenced inside Redis.
        assertLockStillHeld();
        // skipWhenEmpty needs a resolved recordCount, which only exists in
        // contract mode (declareRecords). Warn loudly on misconfig instead of
        // silently writing the empty payload the flag was meant to guard against.
        if (ek.skipWhenEmpty && !contractMode) {
          console.warn(`  [extraKey] ${ek.key} declares skipWhenEmpty but ${domain}:${resource} is not in contract mode (no declareRecords) — guard inactive`);
        }
        const ekData = ek.transform ? ek.transform(data) : data;

        // Guard 1 — pre-publish internals escaping through an extra key. `data` here is
        // the RAW fetcher output, NOT `publishData`: a transform written as `{ ...data }`
        // re-exports everything publishTransform deliberately stripped from the canonical
        // key. That shipped an 11.5 MB forecast:predictions-bootstrap:v1 (66x its own
        // canonical key) and every bootstrap origin miss paid for it. Fail loudly: the
        // canonical key is already published at this point, so refusing to write the extra
        // key leaves last-good data in place rather than serving a monstrous payload.
        const leaked = findLeakedPrePublishFields(data, publishData, ekData, ek);
        if (leaked.length > 0) {
          await releaseLock(`${domain}:${resource}`, runId);
          console.error(
            `  CONTRACT VIOLATION on extraKey ${ek.key}: re-exports ${leaked.length} pre-publish field(s) `
            + `that publishTransform strips from the canonical key: ${leaked.join(', ')}. `
            + `extraKey transforms receive the RAW fetcher output, not the published payload — `
            + `project it first (e.g. compose with your publishTransform) or list the fields in `
            + `allowPrePublishFields if they are genuinely intended.`,
          );
          await exitAfterTelemetryFlush(1);
        }

        let ekEnvelope = null;
        if (contractMode) {
          const ekDeclare = typeof ek.declareRecords === 'function' ? ek.declareRecords : declareRecords;
          let ekCount;
          try {
            ekCount = resolveRecordCount(ekDeclare, ekData);
          } catch (err) {
            await releaseLock(`${domain}:${resource}`, runId);
            console.error(`  CONTRACT VIOLATION on extraKey ${ek.key}: ${err.message || err}`);
            await exitAfterTelemetryFlush(1);
          }
          // Opt-in skip-empty: don't overwrite a good cached extra-key payload
          // with a recordCount=0 write on a partial fetch (e.g. a token panel
          // whose IDs the upstream dropped this cycle). Preserve last-good by
          // extending the existing key's TTL instead. Three outcomes, reported
          // distinctly because they mean different things to an operator:
          // preserved (EXPIRE confirmed), expected-absent (an
          // allowMissingOnSkip key that has not been written yet), or a real
          // preservation failure / unconfirmed pipeline result.
          if (shouldSkipEmptyExtraKey(ek, ekCount)) {
            assertLockStillHeld();
            const preservation = await extendExistingTtlDetailed(
              [ek.key],
              ek.ttl || ttlSeconds || 600,
              { allowMissingKeys: ek.allowMissingOnSkip ? [ek.key] : [] },
            );
            // Per-key truth, not allExtended: that verdict now forgives an
            // allowed-missing key, so it would report a preserved TTL for a key
            // that is simply absent.
            if (preservation.extendedKeys.includes(ek.key)) {
              console.log(`  [extraKey] ${ek.key} empty (recordCount=0) — skipped write, extended TTL to preserve last-good`);
            } else if (ek.allowMissingOnSkip && preservation.missingKeys.includes(ek.key)) {
              console.log(`  [extraKey] ${ek.key} empty (recordCount=0) — skipped write, optional last-good key was absent`);
            } else {
              console.warn(`  [extraKey] ${ek.key} empty (recordCount=0) — skipped write, TTL preservation failed or was unconfirmed`);
            }
            continue;
          }
          ekEnvelope = {
            fetchedAt: envelopeMeta.fetchedAt,
            recordCount: ekCount,
            sourceVersion: sourceVersion || '',
            schemaVersion: schemaVersion || 1,
            state: ekCount > 0 ? 'OK' : (zeroIsValid ? 'OK_ZERO' : 'OK'),
          };
        }

        // Guard 2 — blunt backstop. Catches a monstrous payload whatever the cause,
        // including one this codebase has not seen yet. Measure the serialized UTF-8
        // value (including a contract envelope), which is what Redis actually stores.
        const ekBytes = extraKeyPayloadBytes(ek.key, ekData, ekEnvelope);
        if (ekBytes > MAX_SEEDED_VALUE_BYTES) {
          await releaseLock(`${domain}:${resource}`, runId);
          console.error(
            `  CONTRACT VIOLATION on extraKey ${ek.key}: payload is ${ekBytes} bytes, above the `
            + `${MAX_SEEDED_VALUE_BYTES}-byte ceiling for a seeded value. Refusing to publish.`,
          );
          await exitAfterTelemetryFlush(1);
        }
        await writeExtraKey(ek.key, ekData, ek.ttl || ttlSeconds, ekEnvelope, { lockFence });
        if (contractMode && ek.metaKey) {
          const metaExtra = typeof ek.metaExtra === 'function'
            ? ek.metaExtra(ekData, data)
            : ek.metaExtra;
          let wroteMeta = false;
          try {
            wroteMeta = await writeSeedMeta(
              ek.key,
              ekEnvelope?.recordCount ?? 0,
              ek.metaKey,
              // Same data TTL the writeExtraKey above just used, so a long-lived
              // extra key can't outlive the meta that reports on it.
              resolveSeedMetaTtl(ek.metaTtlSeconds, ek.ttl || ttlSeconds),
              ek.coverage,
              metaExtra,
              { lockFence },
            );
          } catch (err) {
            if (ek.metaCritical || isSeedLockLostError(err)) throw err;
            console.warn(`  Extra key ${ek.key}: seed-meta ${ek.metaKey} write failed; freshness will age: ${err.message}`);
          }
          if (!wroteMeta && ek.metaCritical) throw new Error(`Extra key ${ek.key}: seed-meta ${ek.metaKey} write failed`);
        }
      }
    }

    assertLockStillHeld();
    const afterPublishResult = afterPublish
      ? await afterPublish(data, { canonicalKey, ttlSeconds, recordCount, runId })
      : null;

    // Mirror content-age fields into seed-meta when the seeder opted in.
    //
    // Read content-age from the LOCAL `contentNewestAt`/`contentOldestAt`
    // computed back at line ~1088 — NOT from `envelopeMeta`. The local
    // values are populated whenever the seeder opted in (`contentAgeOptedIn`
    // === true); `envelopeMeta` is null for non-contract-mode seeders, so
    // gating on `envelopeMeta` silently dropped the content-age signal for
    // every seeder that hadn't migrated to contract mode yet — defeating
    // the opt-in for the majority of the cohort.
    //
    // Both branches publish the same trio (envelopeMeta carries the same
    // values when contract mode populates it at line ~1141); reading from
    // the local source unifies the two paths and makes the seed-meta
    // mirror match the contract-mode envelope exactly.
    const successContentAge = contentAgeOptedIn ? {
      newestItemAt: contentNewestAt,
      oldestItemAt: contentOldestAt,
      maxContentAgeMin,
    } : undefined;
    const meta = await writeFreshnessMetadataSafely(
      domain, resource, recordCount, opts.sourceVersion, ttlSeconds,
      undefined,            // fetchedAtOverride — success path uses now
      successContentAge,
      afterPublishResult?.freshnessMetaPatch,
      { lockFence },
    );
    if (afterFreshness) {
      if (meta == null) {
        throw new Error(`${domain}:${resource} freshness metadata write failed before completion`);
      }
      await afterFreshness(data, {
        canonicalKey,
        ttlSeconds,
        recordCount,
        runId,
        freshnessMeta: meta,
      });
    }

    // This is the final required Redis write for an attested bundle member.
    // It deliberately does not run on fetch failure, contract retry, or
    // validation-skip paths. Bind it to the canonical envelope timestamp so a
    // marker from any other run cannot attest this publish.
    if (bundleCompletionMetaKey) {
      if (meta == null) {
        throw new Error(`${domain}:${resource} freshness metadata write failed before completion attestation`);
      }
      if (!Number.isFinite(envelopeMeta?.fetchedAt)) {
        throw new Error(`${domain}:${resource} canonical envelope timestamp missing before completion attestation`);
      }
      await writeExtraKey(
        bundleCompletionMetaKey,
        {
          fetchedAt: envelopeMeta.fetchedAt,
          completedAt: Date.now(),
          runId,
        },
        Math.max(7 * 24 * 60 * 60, ttlSeconds || 0),
        undefined,
        { lockFence },
      );
    }

    const durationMs = Date.now() - startMs;
    const completionState =
      typeof afterPublishResult?.completionState === 'string'
      && afterPublishResult.completionState.trim().length > 0
        ? afterPublishResult.completionState
        : (contractState || 'LEGACY');
    logSeedResult(domain, recordCount, durationMs, {
      runId,
      canonicalKey,
      fetchedAt: envelopeMeta?.fetchedAt ?? meta?.fetchedAt ?? null,
      ...(contentAgeOptedIn ? { newestItemAt: contentNewestAt, oldestItemAt: contentOldestAt, maxContentAgeMin } : {}),
      payloadBytes,
      contractMode,
      state: completionState,
    });

    // Verify (best-effort: write already succeeded, don't fail the job on transient read issues)
    let verified = false;
    for (let attempt = 0; attempt < SEED_VERIFY_ATTEMPTS; attempt++) {
      try {
        verified = !!(await verifySeedKey(canonicalKey));
        if (verified) break;
        if (attempt < SEED_VERIFY_ATTEMPTS - 1) {
          await new Promise(r => setTimeout(r, SEED_VERIFY_RETRY_DELAY_MS));
        }
      } catch {
        if (attempt < SEED_VERIFY_ATTEMPTS - 1) {
          await new Promise(r => setTimeout(r, SEED_VERIFY_RETRY_DELAY_MS));
        }
      }
    }
    if (verified) {
      console.log(`  Verified: data present in Redis`);
    } else {
      console.warn(`  WARNING: verification read returned null for ${canonicalKey} (write succeeded, may be transient)`);
    }

    console.log(`\n=== Done (${Math.round(durationMs)}ms) ===`);
    await releaseLock(`${domain}:${resource}`, runId);
    await exitAfterTelemetryFlush(0);
  } catch (err) {
    lockHeartbeat.stop();
    if (isSeedLockLostError(err)) {
      const durationMs = Date.now() - startMs;
      if (err.holder) {
        // Same outcome as the acquire-time lock-contention skip: the run that
        // now holds the lock publishes this tick and owns its seed-meta.
        console.warn(
          `  SKIPPED: seed lock ${err.lockKey} was taken over by run ${err.holder} before this run (${runId}) `
          + 'finished publishing — remaining writes and seed-meta left to that run',
        );
        console.log(`\n=== Done (${Math.round(durationMs)}ms, LOCK LOST) ===`);
        await exitAfterTelemetryFlush(0);
      }
      // No run owns the tick: keep last-good alive and report a graceful
      // failure so the next tick retries. seed-meta is not touched.
      console.warn(
        `  SKIPPED: seed lock ${err.lockKey} expired before this run (${runId}) finished publishing `
        + '— remaining writes and seed-meta skipped',
      );
      await preserveExistingKeys().catch(() => false);
      console.log(`\n=== Failed gracefully (${Math.round(durationMs)}ms, lock expired) ===`);
      await exitAfterTelemetryFlush(GRACEFUL_FETCH_FAILURE_EXIT_CODE);
    }
    await releaseLock(`${domain}:${resource}`, runId);
    throw err;
  }
}