Blame for scripts/us_local_refresh.py

#!/usr/bin/env python
"""Weekly LOCAL US registers refresh: LCFS (CARB) + EPA (Part 80 Renewable Fuel Producers).



Cloud egress unverified for both endpoints (INS precedent: GSE blocked HF Jobs + GitHub Actions

datacenter IPs); the files are tiny (1.7MB + 5.8MB) and weekly cadence suffices, so this refreshes

HERE, local, and publishes through the same private-dataset candidate/release flow the daily and

INS refreshes use -- never a direct serving-data write, never auto-activation.



    fetch CARB xlsx + EPA xls (browser UA) -> schema assert (refuse loudly on drift)

    -> normalize (lcfs.py / epa.py) -> corpus_watch row-count bands

       (data/us_registers_counts.json -- a DEDICATED manifest, see tests/test_us_registers_counts.py

       for why this never touches the shared data/corpus_counts.json)

    -> snapshot-diff for change events -> data_plane.publish



Any schema/band/rebuild failure leaves current serving data untouched (no publish call at all)

and is recorded loudly in the report. A DOWNLOAD failure is treated differently, because an

unreachable source says nothing about the data: that register keeps its previous snapshot and

the other register still publishes, provided a usable previous snapshot exists and at least one

register actually refreshed. Registered as a weekly Windows Scheduled Task

("Sinetti PC weekly US refresh"), weekend slot (see docs/ins-local-refresh-runbook.md for the sibling

INS task this mirrors).



Usage:

  python scripts/us_local_refresh.py            # full run: fetch + publish

  python scripts/us_local_refresh.py --dry-run  # config check only, no network

"""
from __future__ import annotations

import argparse
import json
import os
import shutil
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path

ROOT = Path(__file__).parent.parent
sys.path.insert(0, str(ROOT))
os.environ.setdefault("APP_BACKEND", "local")


def _devnull_headless_streams() -> None:
    """Under pythonw.exe (the Scheduled Task launcher) there is no console: sys.stdout/stderr

    are None and ANY print/progress-bar write dies with "'NoneType' object has no attribute

    'write'". Route them to devnull so the run only speaks via runner.log + the report file.

    Confirmed live 2026-08-07: the task's first real run failed inside data_plane.boot_sync's

    huggingface_hub progress-bar write for exactly this reason -- must run before that import."""
    if sys.stdout is None:
        sys.stdout = open(os.devnull, "w", encoding="utf-8")  # noqa: SIM115 — lives for the process
    if sys.stderr is None:
        sys.stderr = open(os.devnull, "w", encoding="utf-8")  # noqa: SIM115


_devnull_headless_streams()

import data_plane  # noqa: E402
import epa  # noqa: E402
import lcfs  # noqa: E402
import scripts.refresh_quality as refresh_quality  # noqa: E402
import state_paths  # noqa: E402
from sinetti_services import corpus_watch, job_activation  # noqa: E402
from sinetti_services import source_checks  # noqa: E402
from sinetti_services.us_local_email_job import run_us_local_email_step  # noqa: E402

REPORT_NAME = "last_us_local_refresh.json"
LOG_NAME = "runner.log"
DEFAULT_STAGING_DIR = ROOT / "data" / "refresh" / "us_local"


def _manifest_path() -> Path:
    return ROOT / "data" / "us_registers_counts.json"


def load_env_file(path, environ=os.environ) -> int:
    """Load KEY=VALUE lines from a .env file into `environ`. Existing keys are NEVER overwritten

    and values are NEVER printed. Returns the number of keys loaded. Missing file -> 0."""
    p = Path(path)
    if not p.exists():
        return 0
    loaded = 0
    for line in p.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        key = key.strip()
        value = value.strip().strip('"').strip("'")
        if key and key not in environ:
            environ[key] = value
            loaded += 1
    return loaded


def _cached_login_token() -> str:
    try:
        from huggingface_hub import get_token
        return get_token() or ""
    except Exception:
        return ""


def _log_line(staging: Path, text: str) -> None:
    staging.mkdir(parents=True, exist_ok=True)
    stamp = datetime.now(timezone.utc).isoformat(timespec="seconds")
    with open(staging / LOG_NAME, "a", encoding="utf-8") as fh:
        fh.write(f"{stamp} {text}\n")


def _write_report(staging: Path, report: dict) -> None:
    staging.mkdir(parents=True, exist_ok=True)
    (staging / REPORT_NAME).write_text(json.dumps(report, indent=2, default=str), encoding="utf-8")


_SERVING_PAIR = ("sinetti_app.sqlite.gz", "sinetti_data_release.json")


def _artifact_pathway_count(artifact_gz: Path) -> int:
    return _artifact_row_count(artifact_gz, "v_app_pathways")


def _artifact_row_count(artifact_gz: Path, table: str) -> int:
    import gzip as _gzip
    import sqlite3 as _sqlite3

    with tempfile.NamedTemporaryFile(suffix=".sqlite", delete=False) as tmp:
        tmp_path = Path(tmp.name)
    try:
        with _gzip.open(artifact_gz, "rb") as fi, open(tmp_path, "wb") as fo:
            shutil.copyfileobj(fi, fo)
        conn = _sqlite3.connect(tmp_path)
        try:
            return int(conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0])
        finally:
            conn.close()
    finally:
        tmp_path.unlink(missing_ok=True)


def _drip_register_site_geocode(artifact_gz: Path | None) -> dict:
    """Budgeted Google drip for newly projected EPA/LCFS facility sites. Fail-open always."""
    if artifact_gz is None or not Path(artifact_gz).exists():
        return {"status": "skipped", "reason": "no-artifact"}
    try:
        import scripts.daily_refresh as daily_refresh
        import scripts.geocode_budget as geocode_budget
        import scripts.geocode_holders_google as google_geocoder
        import scripts.geocode_sites as site_geocoder
        import state_paths
        from sinetti_data import load_app_sites

        rows = [
            row
            for row in load_app_sites(artifact_gz)
            if str(row.get("site_source") or "")
            in {"epa_facility", "lcfs_facility", "lcfs_facility_coarse"}
        ]
        if not rows:
            return {"status": "skipped", "reason": "no-register-facility-sites", "attempted": 0}
        # Straight to the site geocoder on the shared holder ledger: the nightly's
        # geocode_sites_with_budget now runs a site-accurate-only drip on its own ledger
        # (SIN-0809), which would drop coarse LCFS facilities and switch this drip off.
        ledger_path = state_paths.refresh_usage_path()
        daily_limit = daily_refresh._env_int("GEOCODE_DAILY_LIMIT", geocode_budget.DEFAULT_DAILY_LIMIT)
        monthly_limit = daily_refresh._env_int("GEOCODE_MONTHLY_LIMIT", geocode_budget.DEFAULT_MONTHLY_LIMIT)
        key = google_geocoder._key()
        if not key:
            return {"status": "skipped-no-google-key", "attempted": 0, "geocoded": 0}
        remaining = geocode_budget.remaining_calls(
            geocode_budget.load_usage(ledger_path), daily_limit=daily_limit, monthly_limit=monthly_limit,
        )
        # Cap per US refresh so one weekend run cannot burn the whole daily budget.
        limit = min(int(os.environ.get("US_SITE_GEOCODE_MAX_N", "400") or 400), remaining)
        if limit <= 0:
            return {"status": "skipped-budget-exhausted", "remaining_before": remaining,
                    "attempted": 0, "geocoded": 0}
        result = site_geocoder.run_geocode_sites(
            max_n=limit,
            key=key,
            dry_run=False,
            sites_rows=rows,
            budget_ledger_path=ledger_path,
            daily_limit=daily_limit,
            monthly_limit=monthly_limit,
        )
        return {
            "status": result.get("status") or "ok",
            "remaining_before": remaining,
            "attempted": int(result.get("attempted") or 0),
            "geocoded": int(result.get("geocoded") or 0),
            "failed": int(result.get("failed") or 0),
            "todo": int(result.get("todo") or 0),
            "detail": result,
        }
    except Exception as exc:  # noqa: BLE001 — geocode must never block publish
        return {
            "status": "error",
            "error": f"{type(exc).__name__}: {str(exc)[:300]}",
            "attempted": 0,
            "geocoded": 0,
        }


def _rebuild_candidate_serving_pair(candidate_root: Path, rebuild_fn=None) -> dict:
    """Rebuild the candidate's serving pair from its own data/ (which now holds the fresh

    LCFS/EPA snapshots beside the pulled register bundles), validate it, and swap only the

    verified pair back into the candidate. INS-refresh pattern; any failure leaves the

    candidate's previous pair untouched and refuses the publish."""
    if rebuild_fn is None:
        import scripts.daily_refresh as daily_refresh

        rebuild_fn = daily_refresh.rebuild_bundled_app_database
    import sinetti_data_release

    candidate_data = candidate_root / "data"
    try:
        with tempfile.TemporaryDirectory(prefix="sinetti-us-rebuild-") as temp_name:
            staged_root = Path(temp_name) / "candidate"
            staged_data = staged_root / "data"
            shutil.copytree(candidate_data, staged_data)
            app_db = rebuild_fn(root=staged_root, replace=True)
            if app_db.get("status") != "ok":
                raise RuntimeError(str(app_db.get("error") or "candidate rebuild did not complete"))
            validation = sinetti_data_release.validate_data_release(
                staged_data / "sinetti_app.sqlite.gz",
                manifest_path=staged_data / "sinetti_data_release.json",
                inspect_database=True,
            )
            if not validation.get("ok"):
                raise RuntimeError(str(validation.get("reason") or "rebuilt serving pair is invalid"))
            # A rebuild in a workspace missing cert_pdf_structured.sqlite.gz "succeeds" with ZERO
            # v_app_pathways rows and would silently gut every deep-dive downstream. Refuse it.
            pathway_count = _artifact_pathway_count(staged_data / "sinetti_app.sqlite.gz")
            if pathway_count <= 0:
                raise RuntimeError(
                    "rebuilt serving pair has zero v_app_pathways rows "
                    "(structured PDF layer missing from the candidate workspace?)"
                )
            # Same trap, second shape: a structured DB from before the sites layer imports with
            # a silent sites=[] fallback, gutting Certified Sites / Browse map annex sites
            # (release 79fc6880 shipped exactly this way on 2026-08-08). Refuse it too.
            sites_count = _artifact_row_count(staged_data / "sinetti_app.sqlite.gz", "sites")
            if sites_count <= 0:
                raise RuntimeError(
                    "rebuilt serving pair has zero sites rows "
                    "(structured DB predates the sites layer, or structured_sites is empty?)"
                )
            for name in _SERVING_PAIR:
                shutil.copy2(staged_data / name, candidate_data / name)
        return {
            "status": "ok",
            "certificates": app_db.get("certificates"),
            "lcfs_epa_address_fills": app_db.get("lcfs_epa_address_fills"),
        }
    except Exception as exc:  # noqa: BLE001 -- refuse the publish, never crash the run
        return {"status": "error", "error": f"{type(exc).__name__}: {str(exc)[:300]}"}


def _fetch_and_normalize_lcfs(fetch_fn) -> dict:
    """Fetch + schema-assert + normalize LCFS. Returns a refresh-leg report dict; never raises.



    A failure records WHICH stage failed, because the two mean opposite things: failure='fetch'

    is the source being unreachable (transient, the previous snapshot is still true), while

    failure='schema' is a data-shape change we do not understand yet and must always refuse."""
    try:
        header, rows = fetch_fn()
    except Exception as exc:  # noqa: BLE001 -- any failure here must refuse, never crash the run
        return {"status": "error", "decision": "kept_previous", "failure": "fetch",
                "error": f"{type(exc).__name__}: {exc}"}
    try:
        lcfs.assert_schema(header)
        certs, excluded = lcfs.normalize_pathways(rows)
    except Exception as exc:  # noqa: BLE001
        return {"status": "error", "decision": "kept_previous", "failure": "schema",
                "error": f"{type(exc).__name__}: {exc}"}
    return {"status": "ok", "decision": "accepted", "rows_total": len(rows),
            "certs": len(certs), "excluded": len(excluded), "raw_rows": rows, "certs_list": certs}


def _fetch_and_normalize_epa(fetch_fn) -> dict:
    try:
        header, rows = fetch_fn()
    except Exception as exc:  # noqa: BLE001
        return {"status": "error", "decision": "kept_previous", "failure": "fetch",
                "error": f"{type(exc).__name__}: {exc}"}
    try:
        epa.assert_schema(header)
        certs = epa.normalize_rows(rows)
    except Exception as exc:  # noqa: BLE001
        return {"status": "error", "decision": "kept_previous", "failure": "schema",
                "error": f"{type(exc).__name__}: {exc}"}
    return {"status": "ok", "decision": "accepted", "rows_total": len(rows),
            "certs": len(certs), "certs_list": certs}


# Which corpus-band probes each register contributes. A leg that did not refresh contributes
# none of its keys -- corpus_watch.check_counts requires the counts and the manifest to carry
# EXACTLY the same keys, so a partial refresh has to narrow the manifest, never the counts.
_LEG_BAND_KEYS = {
    "lcfs": ("lcfs.pathways_total", "lcfs.biofuel_subset"),
    "epa": ("epa.producer_rows",),
}


def _previous_snapshot_certs(candidate_data: Path, leg_name: str) -> list | None:
    """The previous snapshot for a register in the pulled candidate baseline, or None when there

    is nothing usable to keep. Never raises -- an unreadable snapshot counts as absent."""
    loaders = {
        "lcfs": (lcfs.load_lcfs_certs, candidate_data / "lcfs" / "pathways.json.gz"),
        "epa": (epa.load_epa_certs, candidate_data / "epa" / "producers.json.gz"),
    }
    loader, path = loaders[leg_name]
    if not path.is_file():
        return None
    try:
        certs = loader(path=path)
    except Exception:  # noqa: BLE001 -- an unreadable previous snapshot is not something to keep
        return None
    return certs or None


def _narrow_manifest(manifest: dict, keys) -> dict:
    """The band manifest restricted to `keys`, so a partial refresh band-checks exactly the

    counts it actually produced."""
    wanted = set(keys)
    narrowed = dict(manifest)
    narrowed["counts"] = {k: v for k, v in (manifest.get("counts") or {}).items() if k in wanted}
    return narrowed


def _register_publish_gate_review() -> dict:
    """LCFS/EPA have no PDF corpus and are not covered by the certificate-PDF Trust Review

    (sinetti_services/trust_review_service.py) -- its checks are scoped to ISCC/REDcert/INS

    certificate-format evidence (data_trust_control_service, cert_format_catalog.json), none of

    which exists for these registers. By the time upload succeeds below, this run's own hard

    gates have ALREADY passed -- schema assert, corpus_watch bands (this register family's

    coverage-check equivalent), and rebuild + validate_data_release -- because publish never

    happens otherwise (see `ok` above). That IS this register family's equivalent of a clean

    review. Shaped to satisfy job_activation.trust_review_clean() with a zero-finding

    "completed" review, honestly labeled as this run's own publish gate, not a PDF review."""
    return {"status": "completed", "checks": [], "source": "us_local_refresh_publish_gate"}


def _exception_http_status(exc: Exception) -> int | None:
    response = getattr(exc, "response", None)
    status = getattr(response, "status_code", None)
    return status if isinstance(status, int) else None


def _is_publish_conflict(exc: Exception) -> bool:
    return _exception_http_status(exc) == 412


def _finalize_report(staging: Path, report: dict, rc: int) -> None:
    finished = datetime.now(timezone.utc)
    started = datetime.fromisoformat(report["started_at"])
    if not (rc == 0 and report.get("status") == "unchanged"):
        report["status"] = "ok" if rc == 0 else "error"
    report["finished_at"] = finished.isoformat()
    report["duration_s"] = round((finished - started).total_seconds(), 2)
    _write_report(staging, report)
    refresh_quality.append_history_files(staging, report)
    # Operator brief last. Member flags already sit on the shared change log via
    # append_history_files, so this step does not send a second subscriber digest.
    try:
        report["email"] = run_us_local_email_step(report, history_dir=staging)
    except Exception as exc:  # noqa: BLE001
        report["email"] = {
            "status": f"error: {type(exc).__name__}: {str(exc)[:200]}",
            "admin": "not-attempted", "digest": "not-attempted", "red_flag_count": 0,
        }
    _write_report(staging, report)
    _log_line(staging, f"status={report['status']} rc={rc} upload={report['upload'].get('status')} "
                       f"activation={report.get('job_activation', {}).get('status', '?')}")


def main(argv: list[str] | None = None, *, fetch_lcfs_fn=None, fetch_epa_fn=None,

         fetch_lcfs_last_modified_fn=None, probe_lcfs_fn=None, probe_epa_fn=None,

         rebuild_fn=None, environ=os.environ, env_file=None,

         _candidate_root: Path | None = None, _defer_finalize: bool = False,

         _attempt_result: dict | None = None, _site_geocode_report: dict | None = None) -> int:
    from sinetti_operator.registration import register_live_providers
    register_live_providers()  # jobs never reach full_registration (SIN-0999)
    env_path = env_file if env_file is not None else ROOT / ".env"
    load_env_file(env_path, environ=environ)
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--dry-run", action="store_true", help="print config, no network calls")
    ap.add_argument("--staging-dir", default=str(DEFAULT_STAGING_DIR))
    args = ap.parse_args(argv)
    staging = Path(args.staging_dir)

    token = (environ.get("HF_TOKEN", "") or environ.get("HF_JOB_TOKEN", "")
             or environ.get("HF_CREATE_TOKEN", "") or _cached_login_token())
    data_repo_id = environ.get("SINETTI_DATA_REPO_ID", "")

    if args.dry_run:
        print(json.dumps({"dry_run": True, "data_repo_id": data_repo_id,
                          "staging_dir": str(staging), "hf_token_present": bool(token)}, indent=2))
        return 0

    if not data_repo_id:
        print("REFUSING: SINETTI_DATA_REPO_ID not set -- this pipeline publishes only through "
              "the private dataset candidate/release flow, no legacy fallback.")
        _log_line(staging, "refused: SINETTI_DATA_REPO_ID missing")
        return 2

    if not token:
        print("REFUSING: no HF token (HF_TOKEN / HF_JOB_TOKEN / HF_CREATE_TOKEN) in environment, "
              ".env, or a cached `huggingface-cli login`.")
        _log_line(staging, "refused: HF_TOKEN missing")
        return 2

    if _candidate_root is None:
        orchestration_started = datetime.now(timezone.utc).isoformat()
        previous_geocode_report = None
        final_report = None
        final_rc = 1
        attempts = 0
        conflicts = 0
        for attempt_index in range(2):
            attempts += 1
            attempt_result: dict = {}
            with tempfile.TemporaryDirectory(prefix="sinetti-us-candidate-") as workspace_name:
                final_rc = main(
                    argv, fetch_lcfs_fn=fetch_lcfs_fn, fetch_epa_fn=fetch_epa_fn,
                    fetch_lcfs_last_modified_fn=fetch_lcfs_last_modified_fn,
                    probe_lcfs_fn=probe_lcfs_fn, probe_epa_fn=probe_epa_fn,
                    rebuild_fn=rebuild_fn, environ=environ, env_file=env_file,
                    _candidate_root=Path(workspace_name), _defer_finalize=True,
                    _attempt_result=attempt_result, _site_geocode_report=previous_geocode_report,
                )
            final_report = attempt_result.get("report")
            if final_report is None:
                # Defensive fallback for an internal attempt that returned before making a report.
                return final_rc
            if not final_report.get("upload", {}).get("retryable_conflict") or attempt_index == 1:
                break
            conflicts += 1
            previous_geocode_report = final_report["refresh"].get("site_geocode")

        assert final_report is not None
        final_report["started_at"] = orchestration_started
        final_report["publish_attempts"] = attempts
        final_report["publish_conflicts_recovered"] = conflicts if final_rc == 0 else 0
        _finalize_report(staging, final_report, final_rc)
        return final_rc

    candidate_root = Path(_candidate_root)
    candidate_data = candidate_root / "data"

    baseline = data_plane.boot_sync(
        repo_id=data_repo_id, token=token, cache_dir=candidate_root / "data-cache",
        data_dir=candidate_data, materialize=True,
    )
    baseline_failed = baseline.get("status") != "ok" or not baseline.get("hub_revision")

    started = datetime.now(timezone.utc)
    run_id = started.isoformat()

    if baseline_failed:
        report = {
            "started_at": run_id, "runner": "us-local-refresh",
            "baseline": baseline,
            "refresh": {"lcfs": {"status": "error", "decision": "kept_previous",
                                 "error": "verified dataset baseline unavailable"},
                       "epa": {"status": "error", "decision": "kept_previous",
                              "error": "verified dataset baseline unavailable"}},
            "upload": {"status": "skipped-baseline-error"},
            "status": "error",
        }
        if _defer_finalize:
            if _attempt_result is not None:
                _attempt_result.update({"report": report, "rc": 1})
            return 1
        _write_report(staging, report)
        _log_line(staging, f"status=error rc=1 reason=baseline_failed")
        return 1

    fetch_lcfs_fn = fetch_lcfs_fn or lcfs.fetch_lcfs_rows
    fetch_epa_fn = fetch_epa_fn or epa.fetch_epa_rows
    fetch_lcfs_last_modified_fn = fetch_lcfs_last_modified_fn or lcfs.fetch_lcfs_source_last_modified

    # Probe both publishers before downloading. CARB serves Last-Modified (2026-08-08 when
    # measured) so its gate is real; EPA's CDX endpoint serves neither Last-Modified nor ETag,
    # so decide_skip falls back to Content-Length -- a weak signal, and EPA is the register
    # that moves most (8 added / 8 deregistered in the six days after 2026-08-08), so that
    # gate rarely fires. Both are advisory: a failed probe fetches as before.
    lcfs_gate = source_checks.gate(
        "LCFS", probe_fn=probe_lcfs_fn or lcfs.probe_lcfs_source,
        bundle_path=candidate_data / "lcfs" / "pathways.json.gz", checked_at=run_id,
    )
    epa_gate = source_checks.gate(
        "EPA", probe_fn=probe_epa_fn or epa.probe_epa_source,
        bundle_path=candidate_data / "epa" / "producers.json.gz", checked_at=run_id,
    )
    if lcfs_gate["skip"]:
        lcfs_leg = {"status": "unchanged", "decision": "kept_previous",
                    "reason": lcfs_gate["reason"], "signal": lcfs_gate["signal"],
                    "source_last_modified": lcfs_gate["source_last_modified"]}
    else:
        lcfs_leg = _fetch_and_normalize_lcfs(fetch_lcfs_fn)
        lcfs_leg["probe"] = lcfs_gate["reason"]
    if epa_gate["skip"]:
        epa_leg = {"status": "unchanged", "decision": "kept_previous",
                   "reason": epa_gate["reason"], "signal": epa_gate["signal"]}
    else:
        epa_leg = _fetch_and_normalize_epa(fetch_epa_fn)
        epa_leg["probe"] = epa_gate["reason"]
    # Advisory only -- CARB's own Last-Modified header, separate from the row fetch above and
    # never allowed to affect the schema/publish decision. A HEAD-request failure here must
    # not touch lcfs_leg["status"]; it only leaves source_last_modified blank.
    lcfs_source_last_modified = lcfs_gate["source_last_modified"] or ""
    if lcfs_leg["status"] == "ok" and not lcfs_source_last_modified:
        try:
            lcfs_source_last_modified = fetch_lcfs_last_modified_fn() or ""
        except Exception:  # noqa: BLE001 -- advisory only, never blocks the refresh
            lcfs_source_last_modified = ""
    if lcfs_leg["status"] == "ok":
        lcfs_leg["source_last_modified"] = lcfs_source_last_modified

    legs = {"lcfs": lcfs_leg, "epa": epa_leg}

    # A source being unreachable must not strand the OTHER source's fresh data, and must not
    # strand build-time normalisation fixes either -- the candidate is rebuilt from the
    # published snapshots, so a rebuild is worth doing even when only one register moved.
    # (CARB served HTTP 503 from its own origin on 2026-08-08 while EPA fetched fine.) The leg
    # that failed keeps its previous snapshot, and only if there IS one: republishing a
    # register as absent would look exactly like the register losing every row.
    # Schema drift is never tolerated here -- an unexplained data-shape change still refuses.
    blocking: list[str] = []
    for name, leg in legs.items():
        if leg["status"] == "ok":
            continue
        if leg["status"] == "unchanged":
            # Publisher reported nothing moved: nothing to accept, nothing to block.
            continue
        if leg.get("failure") != "fetch":
            blocking.append(name)
            continue
        if _previous_snapshot_certs(candidate_data, name) is None:
            leg["error"] = (
                f"{leg.get('error', '')} (no usable previous {name.upper()} snapshot in the "
                "candidate baseline to keep)"
            ).strip()
            blocking.append(name)

    accepted = [name for name, leg in legs.items() if leg["status"] == "ok"]
    all_unchanged = not blocking and not accepted and all(
        leg["status"] == "unchanged" for leg in legs.values()
    )
    # Nothing fresh anywhere means nothing to publish: a republish of unchanged snapshots would
    # burn a Hub revision and a release for no data change.
    legs_ok = not blocking and bool(accepted)

    bands_report: dict = {"status": "skipped", "reason": "refresh not accepted"}
    if legs_ok:
        counts: dict[str, int] = {}
        if "lcfs" in accepted:
            counts["lcfs.pathways_total"] = lcfs_leg["rows_total"]
            counts["lcfs.biofuel_subset"] = lcfs_leg["certs"]
        if "epa" in accepted:
            counts["epa.producer_rows"] = epa_leg["certs"]
        manifest = corpus_watch.load_manifest(_manifest_path())
        bands_report = corpus_watch.counts_report(counts, _narrow_manifest(manifest, counts))
        skipped_keys = sorted(
            key
            for name, leg_keys in _LEG_BAND_KEYS.items()
            if name not in accepted
            for key in leg_keys
        )
        bands_report["skipped_keys"] = skipped_keys
        if skipped_keys:
            bands_report["skipped_reason"] = "source unavailable; previous snapshot kept"

    ok = legs_ok and bands_report.get("status") == "ok"

    change_events = {"lcfs": [], "epa": []}
    if ok:
        if "lcfs" in accepted:
            old_lcfs = lcfs.load_lcfs_certs(path=candidate_data / "lcfs" / "pathways.json.gz")
            lcfs_diff = lcfs.diff_lcfs_snapshot(old_lcfs, lcfs_leg["certs_list"])
            change_events["lcfs"] = refresh_quality.build_register_diff_events(
                run_id, "LCFS", old_lcfs, lcfs_leg["certs_list"], lcfs_diff, detected_at=run_id,
            )
            lcfs.write_snapshot(
                lcfs_leg["certs_list"], path=candidate_data / "lcfs" / "pathways.json.gz",
                fetched_at=run_id, source="us_local_refresh",
                source_last_modified=lcfs_source_last_modified or None,
            )
        if "epa" in accepted:
            old_epa = epa.load_epa_certs(path=candidate_data / "epa" / "producers.json.gz")
            epa_diff = epa.diff_epa_snapshot(old_epa, epa_leg["certs_list"])
            change_events["epa"] = refresh_quality.build_register_diff_events(
                run_id, "EPA", old_epa, epa_leg["certs_list"], epa_diff, detected_at=run_id,
            )
            epa.write_snapshot(
                epa_leg["certs_list"], path=candidate_data / "epa" / "producers.json.gz",
                fetched_at=run_id, source="us_local_refresh",
            )

    # Rebuild the candidate serving pair so the published candidate actually CONTAINS the
    # US registers (INS-refresh pattern): stage a copy, rebuild the app DB there, validate,
    # then swap only the verified pair back. Snapshot-only publishes left LCFS/EPA invisible
    # to serving until some other refresh happened to rebuild.
    app_db_report: dict = {"status": "skipped", "reason": "refresh not accepted"}
    if ok:
        app_db_report = _rebuild_candidate_serving_pair(candidate_root, rebuild_fn)
        ok = app_db_report.get("status") == "ok"

    # Fail-open drip: new EPA/LCFS facility sites need lat/lng in data/site_coords.json.
    # Never refuse the publish for geocode budget / missing key.
    site_geocode_report = _site_geocode_report
    if site_geocode_report is None:
        site_geocode_report = _drip_register_site_geocode(
            candidate_root / "data" / "sinetti_app.sqlite.gz" if ok else None
        )

    def _leg_report(leg: dict) -> dict:
        out = {k: v for k, v in leg.items() if k not in ("raw_rows", "certs_list")}
        return out

    report: dict = {
        "started_at": run_id, "runner": "us-local-refresh",
        "baseline": baseline,
        "refresh": {
            "lcfs": {**_leg_report(lcfs_leg), "change_events": change_events["lcfs"]} if lcfs_leg["status"] == "ok" else lcfs_leg,
            "epa": {**_leg_report(epa_leg), "change_events": change_events["epa"]} if epa_leg["status"] == "ok" else epa_leg,
            "bands": bands_report,
            "app_db": app_db_report,
            "site_geocode": site_geocode_report,
        },
        "upload": {"status": "pending"},
        "job_activation": {"status": "not-attempted", "reason": "dataset release not published"},
    }

    rc = 0
    if all_unchanged:
        report["upload"] = {"status": "skipped-unchanged",
                            "reason": "every source reported unchanged; nothing to publish"}
        report["status"] = "unchanged"
    elif not legs_ok:
        report["upload"] = {"status": "skipped-refresh-error"}
        rc = 1
    elif bands_report.get("status") != "ok":
        report["upload"] = {"status": "skipped-band-violation"}
        rc = 1
    elif app_db_report.get("status") != "ok":
        report["upload"] = {"status": "skipped-rebuild-error"}
        rc = 1
    else:
        try:
            result = data_plane.publish(
                candidate_root, repo_id=data_repo_id, token=token,
                producer="us_local_refresh", expected_parent=baseline.get("hub_revision"),
            )
        except Exception as exc:  # noqa: BLE001
            report["upload"] = {"status": "error",
                                "exception": {"class": type(exc).__name__, "message": str(exc)}}
            http_status = _exception_http_status(exc)
            if http_status is not None:
                report["upload"]["http_status"] = http_status
            if _is_publish_conflict(exc):
                report["upload"]["retryable_conflict"] = True
            rc = 1
            report["job_activation"] = {
                "status": "skipped", "reason": "dataset publish raised an exception",
            }
        else:
            report["upload"] = result
            if result.get("status") != "ok":
                rc = 1
                report["job_activation"] = {
                    "status": "skipped", "reason": "dataset publish did not succeed",
                }
            else:
                try:
                    report["job_activation"] = job_activation.decide_and_activate(
                        enabled=job_activation.job_auto_activate_enabled(environ),
                        ok=True,
                        revision=result.get("commit_sha"),
                        candidate_data_dir=candidate_data,
                        review=_register_publish_gate_review(),
                        state_root=state_paths.state_root(),
                        by="us-local-refresh",
                    )
                except Exception as exc:  # noqa: BLE001 -- publish remains authoritative
                    report["job_activation"] = {
                        "status": "hold",
                        "reason": "dataset published; automatic job activation raised an exception",
                        "exception": {"class": type(exc).__name__, "message": str(exc)},
                    }

    if _defer_finalize:
        if _attempt_result is not None:
            _attempt_result.update({"report": report, "rc": rc})
        return rc
    _finalize_report(staging, report, rc)
    return rc


if __name__ == "__main__":
    raise SystemExit(main())