#!/usr/bin/env python
"""Daily Sinetti database refresh entrypoint for local runs or Hugging Face Scheduled Jobs.

The job refreshes the bundled CS datasets, then runs a capped Google geocode delta pass for only
missing holder locations. A persistent usage ledger prevents runaway Google Maps spend.
"""
from __future__ import annotations

import argparse
import contextlib
import copy
import gzip
import hashlib
import inspect
import json
import os
import re
import shutil
import sqlite3
import sys
import tempfile
import threading
import time
import traceback
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
from typing import Callable, Iterable

ROOT = Path(__file__).parent.parent
sys.path.insert(0, str(ROOT))

import data_plane  # noqa: E402
import iscc_fetch  # noqa: E402
import redcert  # noqa: E402
import schemes as _schemes  # noqa: E402
import redcert_pdf  # noqa: E402
import redcert_pdf_store  # noqa: E402
import rehearsal  # noqa: E402
import rehearsal_candidate_baseline  # noqa: E402
import rehearsal_candidate_registers  # noqa: E402
import rehearsal_recovery_contract  # noqa: E402
import scripts.geocode_budget as geocode_budget  # noqa: E402
import scripts.geocode_holders_google as google_geocoder  # noqa: E402
import scripts.geocode_holder_addresses_backfill as address_geocoder  # noqa: E402
import scripts.geocode_sites as site_geocoder  # noqa: E402
import scripts.refresh_quality as refresh_quality  # noqa: E402
import sinetti_data  # noqa: E402
import state_paths  # noqa: E402
from sinetti_services import company_overrides, geocode_points, geocode_workspace
from sinetti_services import trust_review_service, refresh_publication  # noqa: E402
from sinetti_services import notifications  # noqa: E402
from sinetti_services import soft_alert_service  # noqa: E402
from sinetti_services.refresh_job_lifecycle import (  # noqa: E402
    can_start_tail,
    flush_refresh_receipt,
    install_interrupt_flush,
    remaining_budget_s,
    TAIL_RESERVE_S,
)
from sinetti_services.returning_holder_priors import (  # noqa: E402
    load_change_events_jsonl,
    prior_coverage_for_refresh,
)
from scripts.rebuild_sinetti_app_artifact import rebuild_bundled_app_database  # noqa: E402
from scripts.run_cert_pdf_delta_refresh import (  # noqa: E402
    rebuild_downstream_from_raw_artifact,
    run_pdf_delta_refresh,
)
from scripts.run_iscc_audit_delta_refresh import run_audit_delta_refresh  # noqa: E402
from scripts.run_parallel_pdf_stages import StageLimits, run_parallel_pdf_delta_refresh  # noqa: E402

try:
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
except (AttributeError, ValueError):
    pass


# Raised 2026-07-23 (owner decision): Google-console-verified real usage showed the old
# 300/9,000 caps were far below both actual spend patterns and the current Geocoding free
# tier (10k free events/month, then ~$5/1k — see geocode_budget.estimate_google_geocoding_cost).
# The caps remain a circuit breaker against runaway loops, not a cost-minimization tool.
DEFAULT_DAILY_GEOCODE_LIMIT = geocode_budget.DEFAULT_DAILY_LIMIT
DEFAULT_MONTHLY_GEOCODE_LIMIT = geocode_budget.DEFAULT_MONTHLY_LIMIT
# Site drip cap (SIN-0825 A): off in code like every bulk-fetch cap; the job registration sets
# the owner's 200 (register_hf_daily_refresh_job.py). Changing the cap needs the owner.
DEFAULT_SITE_GEOCODE_DAILY_LIMIT = 0
DEFAULT_PDF_DELTA_MAX_PDFS = 0
DEFAULT_AUDIT_DELTA_MAX_PDFS = 0
# Same opt-in-off convention as the PDF-parsing delta above: an unmocked test run of main() must
# never make live calls to redcert.eu, so this stays 0 until explicitly configured (env
# REDCERT_PDF_MAX_FETCHES or --redcert-pdf-max-fetches) on the real scheduled job. ~3,000 in-force
# certs fill over several runs at a nonzero cap; already-hosted/no_document certs are free (index
# lookup only, no network) so the cap only bites on the genuinely-pending backlog.
DEFAULT_REDCERT_PDF_MAX_FETCHES = 0
# What the tail rebuild, its Trust Review and the second publish need after the REDcert PDF fill
# (25 min on 29 Oct 2026; SIN-1076), with margin.
TAIL_PUBLISH_RESERVE_S = 40 * 60
DEFAULT_REFRESH_SCHEMES = _schemes.CERTIFICATION_SLUGS
# Registers the address backfill has no target for. Empty since 2BSvs/RSB got theirs (2BSvs: street
# + postcode + city; RSB: the address string after the holder name). A scheme listed here is
# skipped by both geocode loops, because the backfill treats an unknown scheme as ISCC.
_ADDRESS_BACKFILL_PENDING: tuple[str, ...] = ()
DEFAULT_GEOCODE_SCHEMES = tuple(
    slug for slug in _schemes.CERTIFICATION_SLUGS if slug not in _ADDRESS_BACKFILL_PENDING
)
DEFAULT_LEDGER_PATH = state_paths.refresh_usage_path()
DEFAULT_REPORT_PATH = state_paths.refresh_report_path()
DEFAULT_HISTORY_DIR = state_paths.refresh_dir()
DEFAULT_HF_IGNORE_PATTERNS = [
    ".git/*",
    ".pytest_cache/*",
    "__pycache__/*",
    "*/__pycache__/*",
    "*.pyc",
    # huggingface_hub matches ignore patterns with fnmatchcase against the FULL RELATIVE path, so
    # a bare "intel.db" only ever matched a file at the repo root and never data/intel.db, which is
    # where the file actually lives in a non-mounted (local / no SINETTI_STATE_DIR) tree. intel.db
    # holds customer-private state — password hashes, billing and entitlements, watchlists, saved
    # views, activity history — and must never reach the Space repo. The leading "*" makes each
    # pattern depth-agnostic; the sidecars are listed so a journal-mode change cannot reopen this.
    # Covered by tests/test_daily_refresh.py::test_hf_ignore_patterns_exclude_private_state.
    "*intel.db",
    "*intel.db-wal",
    "*intel.db-shm",
    "*intel.db-journal",
    "*.google_maps_key",
]

_REVIEW_SCHEME_NAMES = {_schemes.SLUG_BY_KEY[key]: key for key in _schemes.CERTIFICATION_SCHEMES}


def _now_iso(now: datetime | None = None) -> str:
    value = now or datetime.now(timezone.utc)
    if value.tzinfo is None:
        value = value.replace(tzinfo=timezone.utc)
    return value.astimezone(timezone.utc).isoformat()


def _csv(value: str | None, default: tuple[str, ...]) -> list[str]:
    raw = value if value is not None and value.strip() else ",".join(default)
    return [part.strip().lower() for part in raw.split(",") if part.strip()]


def _env_int(name: str, default: int) -> int:
    try:
        return max(0, int(os.environ.get(name, default)))
    except (TypeError, ValueError):
        return default


def _env_bool(name: str, default: bool = False) -> bool:
    value = os.environ.get(name)
    if value is None:
        return default
    return value.strip().lower() in {"1", "true", "yes", "on"}


_BUSINESS_OVERRIDE_FILES = {
    "entity_identity_overrides": (
        "ENTITY_IDENTITY_OVERRIDES_PATH",
        Path("data/entity_identity_overrides.json"),
    ),
    "related_company_overrides": (
        "SINETTI_RELATED_COMPANY_OVERRIDES",
        Path("data/related_company_overrides.json"),
    ),
}


def _validate_bundled_business_overrides(root: Path) -> dict:
    metadata: dict = {"status": "validated"}
    for receipt_name, (_env_name, relative_path) in _BUSINESS_OVERRIDE_FILES.items():
        path = (root / relative_path).resolve()
        if not path.is_file():
            raise ValueError(f"required bundled business override is not a regular file: {path}")
        contents = path.read_bytes()
        if not contents:
            raise ValueError(f"required bundled business override is empty: {path}")
        try:
            json.loads(contents.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise ValueError(f"required bundled business override is not valid JSON: {path}") from exc
        metadata[receipt_name] = {
            "path": str(path),
            "sha256": hashlib.sha256(contents).hexdigest(),
        }
    return metadata


@contextlib.contextmanager
def _bind_bundled_business_overrides(metadata: dict):
    previous = {}
    try:
        for receipt_name, (env_name, _relative_path) in _BUSINESS_OVERRIDE_FILES.items():
            previous[env_name] = os.environ.get(env_name)
            os.environ[env_name] = metadata[receipt_name]["path"]
        yield
    finally:
        for env_name, value in previous.items():
            if value is None:
                os.environ.pop(env_name, None)
            else:
                os.environ[env_name] = value


def _memory_trace_sample() -> dict:
    """Best-effort RSS + system memory snapshot. Never raises -- a platform quirk just
    shrinks the sample, since the whole point is surviving an unexpected kill."""
    sample: dict = {}
    try:
        import resource

        # ru_maxrss is KB on Linux, bytes on macOS -- this job only ever runs on the Linux
        # HF Jobs image, so KB is the correct unit here.
        sample["rss_mb"] = round(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024, 1)
    except Exception:
        pass
    try:
        for line in Path("/proc/meminfo").read_text(encoding="utf-8").splitlines():
            if line.startswith("MemAvailable:"):
                sample["mem_available_mb"] = round(int(line.split()[1]) / 1024, 1)
            elif line.startswith("MemTotal:"):
                sample["mem_total_mb"] = round(int(line.split()[1]) / 1024, 1)
    except Exception:
        pass
    return sample


def start_memory_trace(*, interval_s: float = 30.0) -> threading.Event:
    """Print an RSS/meminfo line every interval_s so a SIGKILL (OOM) leaves evidence of how
    memory grew right up to the kill -- SIGKILL itself is unblockable, so this is the only
    way to see it coming. Requires the caller's stdout to be unbuffered (-u / PYTHONUNBUFFERED)
    or the lines die with the process exactly like everything else did on 2026-08-17."""
    stop = threading.Event()
    started = time.monotonic()

    def _loop() -> None:
        while not stop.wait(interval_s):
            sample = _memory_trace_sample()
            sample["elapsed_s"] = round(time.monotonic() - started, 1)
            print(f"MEMTRACE {json.dumps(sample, sort_keys=True)}", flush=True)

    threading.Thread(target=_loop, name="memtrace", daemon=True).start()
    return stop


def run_acquisition_queue_step(*, can_continue: bool, dry_run: bool, skip: bool = False,
                               runner=None) -> dict:
    """Pick up the admin-queued acquisition-retry batch, if one is waiting.

    Runs BEFORE the PDF delta so the delta's own rebuild carries the acquired documents. No
    queued request is the normal case. Fail-soft by construction: a broken batch is recorded on
    the request and reported here, and the refresh carries on.
    """
    if dry_run:
        return {"status": "skipped-dry-run"}
    if skip:
        return {"status": "skipped"}
    if not can_continue:
        return {"status": "skipped-refresh-error"}
    try:
        if runner is None:
            from scripts.run_acquisition_retry import run_from_queue as runner
        return runner()
    except Exception as exc:  # noqa: BLE001
        return {"status": "error",
                "exception": {"class": type(exc).__name__, "message": str(exc)}}


def finish_acquisition_queue_step(step: dict, *, pdf_delta: dict, rebuild=None) -> dict:
    """Rebuild structured/app artifacts when the queue promoted documents but the delta did not."""
    if not step.get("promoted_details_artifact"):
        return step
    if (pdf_delta.get("structured") or {}).get("status") == "ok":
        return {**step, "downstream": "rebuilt-by-pdf-delta"}
    try:
        outcome = (rebuild or rebuild_downstream_from_raw_artifact)()
        return {**step, "downstream": outcome.get("status", "ok")}
    except Exception as exc:  # noqa: BLE001
        return {**step, "status": "error", "downstream": "error",
                "exception": {"class": type(exc).__name__, "message": str(exc)}}


def _pdf_stage_limits(args) -> StageLimits:
    return StageLimits(total=args.pdf_stage_total_workers, certificates=args.pdf_stage_certificate_workers,
        audits=args.pdf_stage_audit_workers, iscc=args.pdf_stage_iscc_workers, redcert=args.pdf_stage_redcert_workers)


def _file_hash(path: Path) -> str | None:
    if not path.is_file():
        return None
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _pdf_stage_inputs_unchanged(proof: dict) -> bool:
    return all(_file_hash(Path(path)) == expected for path, expected in proof.items())


def _seed_pdf_stage_candidate(candidate_root: Path) -> dict:
    from scripts.rebuild_sinetti_app_artifact import bundled_paths
    import bundle_meta
    geocode_workspace.recover_journal(ROOT / "data")
    paths = [path for key, path in bundled_paths(ROOT).items()
             if key not in {"structured_db", "structured_artifact", "details_artifact", "work_db", "artifact"}]
    paths += [bundle_meta.meta_path_for(path) for path in paths]
    # INS loader reads these siblings explicitly; omitting them changes serving identities,
    # locations and PDF links even when the copied register snapshot is byte-identical.
    paths += [ROOT / "data/ins" / name for name in
              ("place_overrides.json", "holder_name_overrides.json", "cert_pdfs.json")]
    paths += [ROOT / "data/cfr/facility_place_overrides.json", ROOT / "data/epa/country_overrides.json",
              ROOT / "data/ins/certificate_identity_resolutions.json"]
    paths += [geocode_points.holder_baseline_path(ROOT / "data", scheme) for scheme in geocode_points.HOLDER_SCHEMES]
    paths += [geocode_points.site_coords_nightly_path(ROOT / "data"),
              geocode_points.provenance_path(ROOT / "data"), geocode_points.fallback_path(ROOT / "data")]
    proof = {}
    for source in dict.fromkeys(paths):
        expected = _file_hash(source)
        proof[str(source)] = expected
        if expected is not None:
            target = candidate_root / source.relative_to(ROOT)
            target.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(source, target)
            if _file_hash(target) != expected:
                raise ValueError(f"candidate input changed during copy: {source}")
    # Both layers of the company override files (SIN-0553): the live state paths and the bundled seeds.
    for source in (state_paths.entity_identity_overrides_path(), state_paths.related_company_overrides_path(),
                   company_overrides.entity_seed_path(), company_overrides.related_seed_path()):
        proof[str(source)] = _file_hash(Path(source))
    return proof


def _promote_pdf_stage_handoff(candidate_root: Path, handoff: dict) -> None:
    names = {"raw_work_db": "cert_pdf_details.sqlite", "raw_artifact": "cert_pdf_details.sqlite.gz",
        "structured_work_db": "cert_pdf_structured.sqlite", "structured_artifact": "cert_pdf_structured.sqlite.gz",
        "app_work_db": "sinetti_app.sqlite", "app_artifact": "sinetti_app.sqlite.gz",
        "data_release": "sinetti_data_release.json"}
    if Path(handoff["candidate_root"]).resolve() != candidate_root.resolve():
        raise ValueError("candidate handoff root mismatch")
    files = handoff["files"]
    for key, name in names.items():
        expected_path = candidate_root / "data" / name
        entry = files[key]
        if Path(entry["path"]).resolve() != expected_path.resolve() or _file_hash(expected_path) != entry["sha256"]:
            raise ValueError(f"invalid candidate handoff: {key}")
    with tempfile.TemporaryDirectory(prefix="pdf-promote-", dir=candidate_root) as backup_dir:
        backup = Path(backup_dir)
        existed = {}
        for name in names.values():
            target = ROOT / "data" / name
            existed[name] = target.is_file()
            if existed[name]:
                shutil.copy2(target, backup / name)
        promoted = []
        try:
            for key, name in names.items():
                target = ROOT / "data" / name
                target.parent.mkdir(parents=True, exist_ok=True)
                promoted.append(name)
                shutil.copy2(candidate_root / "data" / name, target)
                if _file_hash(target) != files[key]["sha256"]:
                    raise ValueError(f"promoted output hash mismatch: {key}")
        except Exception:
            for name in reversed(promoted):
                target = ROOT / "data" / name
                if existed[name]:
                    shutil.copy2(backup / name, target)
                else:
                    target.unlink(missing_ok=True)
            raise


def run_pdf_stage_overlap_step(args, *, can_continue: bool, dry_run: bool,
                               should_continue: Callable[[], bool], acquisition: dict,
                               skip_reason: str = "skipped-refresh-error", source_audit: dict | None = None) -> dict:
    certificate_cap = 0 if args.skip_pdf_delta else max(0, args.pdf_delta_max_pdfs)
    audit_cap = 0 if args.skip_audit_delta else max(0, args.audit_delta_max_pdfs)
    if dry_run or not can_continue or (not certificate_cap and not audit_cap and not acquisition.get("promoted_details_artifact")):
        status = "skipped-dry-run" if dry_run else skip_reason if not can_continue else "skipped"
        return {"status": status, "changed": False, "pdf_delta": {"status": status}, "audit_delta": {"status": status}}
    try:
        limits = _pdf_stage_limits(args)
        limits.validate()
        with tempfile.TemporaryDirectory(prefix="sinetti-pdf-stage-candidate-") as private:
            candidate = Path(private)
            inputs = _seed_pdf_stage_candidate(candidate)
            from scripts.build_cert_pdf_details import load_default_queue, DEFAULT_PARALLEL_ITEM_TIMEOUT_S
            audit_kwargs = {'source_audit': source_audit} if source_audit is not None else {}
            result = run_parallel_pdf_delta_refresh(candidate_root=candidate,
                baseline_sqlite=ROOT / "data/cert_pdf_details.sqlite",
                baseline_artifact=ROOT / "data/cert_pdf_details.sqlite.gz",
                max_certificates=certificate_cap, max_audits=audit_cap,
                queue_loader=lambda: load_default_queue(ROOT, fraction=1.0),
                audit_url_map=ROOT / "data/iscc/audit_reports.json", limits=limits,
                prefer_new_certificates=args.pdf_delta_prefer_new_documents,
                prefer_new_audits=args.audit_delta_prefer_new_documents,
                certificate_cache_dir=args.pdf_cache_dir or state_paths.state_root() / "cert_pdf_cache",
                audit_cache_dir=state_paths.state_root() / "audit_report_cache", cache_max_gb=10,
                should_continue=should_continue, certificate_timeout_s=DEFAULT_PARALLEL_ITEM_TIMEOUT_S,
                audit_timeout_s=float(os.environ.get("AUDIT_PARSE_IDLE_TIMEOUT_S", "180")),
                force_assembly_if_no_delta=bool(acquisition.get("promoted_details_artifact")), **audit_kwargs)
            if result.get("status") != "ok":
                return result
            if result.get("changed"):
                if not _pdf_stage_inputs_unchanged(inputs):
                    raise ValueError("app inputs changed during PDF candidate assembly")
                _promote_pdf_stage_handoff(candidate, result["handoff"])
                rebuild = result["assembly"]["rebuild"]
                result["pdf_delta"].update(structured=rebuild["structured"], app_db=result["app_db"],
                    timing={**(result["pdf_delta"].get("raw_parse", {}).get("timing") or {}), **rebuild.get("timing", {})})
                result["app_input_proof"] = inputs
                result["promoted"] = True
            if args.skip_pdf_delta or certificate_cap == 0:
                result["pdf_delta"]["status"] = "skipped"
            if args.skip_audit_delta or audit_cap == 0:
                result["audit_delta"]["status"] = "skipped"
            return result
    except Exception as exc:
        return {"status": "error", "changed": False, "handoff": None,
                "phase": "daily-handoff", "error": str(exc)}


def _pdf_stage_overlap_failed(report: dict) -> bool:
    return (report.get("pdf_stage_overlap") or {}).get("status") == "error"


def run_pdf_source_audit_step(*, data_root: Path, can_continue: bool, dry_run: bool,
                            should_continue: Callable[[], bool], environ=None,
                            remaining_s: float | None = None) -> dict:
    """Use the same job tail admission/deadline; independent audit caps default zero."""
    from sinetti_services import pdf_source_audit
    if not can_continue:
        return pdf_source_audit._empty('skipped-no-time')
    from scripts.build_cert_pdf_details import DEFAULT_PARALLEL_ITEM_TIMEOUT_S
    timeout = (DEFAULT_PARALLEL_ITEM_TIMEOUT_S if remaining_s is None else
               min(DEFAULT_PARALLEL_ITEM_TIMEOUT_S, max(0.0, remaining_s)))
    return pdf_source_audit.run_source_audit(data_root=data_root,
        state_root=state_paths.state_root(), environ=environ, dry_run=dry_run,
        should_continue=should_continue, weekly_local=False,
        source_timeout_s=timeout)


def run_audit_delta_step(
    args,
    *,
    can_continue: bool,
    dry_run: bool,
    should_continue: Callable[[], bool] | None = None,
    source_audit: dict | None = None,
) -> dict:
    if dry_run:
        return {"status": "skipped-dry-run", "max_pdfs": args.audit_delta_max_pdfs}
    if args.skip_audit_delta or args.audit_delta_max_pdfs <= 0:
        return {"status": "skipped", "max_pdfs": args.audit_delta_max_pdfs}
    if not can_continue:
        return {"status": "skipped-refresh-error", "max_pdfs": args.audit_delta_max_pdfs}
    try:
        kwargs = dict(
            max_pdfs=args.audit_delta_max_pdfs,
            prefer_new_documents=args.audit_delta_prefer_new_documents,
        )
        if should_continue is not None:
            kwargs["should_continue"] = should_continue
        if source_audit is not None:
            kwargs['source_audit'] = source_audit
        workers = int(getattr(args, "audit_delta_workers", 1) or 1)
        if workers > 1:
            kwargs["workers"] = workers
        return run_audit_delta_refresh(**kwargs)
    except Exception as exc:
        return {
            "status": "error",
            "max_pdfs": args.audit_delta_max_pdfs,
            "exception": {"class": type(exc).__name__, "message": str(exc)},
        }


def _accepted_review_schemes(refresh_report: dict) -> tuple[str, ...]:
    return tuple(
        display
        for key, display in _REVIEW_SCHEME_NAMES.items()
        if (refresh_report.get(key) or {}).get("status") == "ok"
        and (refresh_report.get(key) or {}).get("decision") == "accepted"
    )


def _post_refresh_review(*, candidate_root: Path, schemes: tuple[str, ...]) -> dict:
    try:
        data = candidate_root / "data"
        enabled = refresh_publication.production_publication_enabled()
        options = {}
        identity = None
        if rehearsal.is_rehearsal():
            options["store_path"] = data / "refresh" / "trust_review.sqlite"
        elif enabled:
            if not os.environ.get("SINETTI_STATE_DIR", "").strip():
                raise ValueError("production publication requires explicit shared state")
            shared = state_paths.state_root().resolve()
            candidate = candidate_root.resolve()
            if (shared == state_paths.BUNDLED_DATA_DIR.resolve()
                    or shared == candidate or shared in candidate.parents or candidate in shared.parents):
                raise ValueError("production review requires external shared state outside the candidate")
            options["store_path"] = shared / "trust_review.sqlite"
        if enabled:
            from scripts.build_cert_format_catalog import main as build_catalog
            if build_catalog(["--artifact", str(data / "cert_pdf_details.sqlite.gz"),
                              "--out", str(data / "cert_format_catalog.json"),
                              "--ins-index", str(data / "ins" / "cert_pdfs.json"),
                              "--skip-thumbs"]) != 0:
                raise ValueError("candidate catalog rebuild failed")
            identity = refresh_publication.capture_review_identity(data)
            groups = sinetti_data.load_app_certificate_groups(data / "sinetti_app.sqlite.gz")
            schemes = tuple(scheme for scheme in _schemes.CERTIFICATION_SCHEMES
                            if scheme in _schemes.LOADED_CERTIFICATION_SCHEMES or groups.get(scheme))
        review = trust_review_service.safe_post_refresh_review(
            candidate_artifact=candidate_root / "data" / "sinetti_app.sqlite.gz",
            details_artifact=candidate_root / "data" / "cert_pdf_details.sqlite.gz",
            schemes=schemes,
            **options,
        )
        if identity is not None:
            if refresh_publication.capture_review_identity(data) != identity:
                raise ValueError("candidate artifacts changed during Trust Review")
            if review.get("candidate_revision") != identity["candidate_revision"]:
                raise ValueError("Trust Review candidate identity does not match artifacts")
            review["publication_artifact_ids"] = identity["artifact_ids"]
        return review
    except Exception as exc:  # noqa: BLE001 - even a broken safety wrapper cannot fail refresh
        return {"status": "failed", "error": f"{type(exc).__name__}: {str(exc)[:300]}"}


def _parse_status_by_cert_for_soft_alerts() -> dict[tuple[str, str], str] | None:
    """Map (scheme, certificate_id) → parse_status for pathway terminal abandon.

    Fail-open: missing/corrupt PDF document reads must not block Soft processing.
    """
    try:
        docs = sinetti_data.load_app_pdf_documents()
    except Exception:
        return None
    out: dict[tuple[str, str], str] = {}
    for doc in docs or ():
        scheme = str(doc.get("scheme") or "").strip()
        cert_id = str(doc.get("certificate_id") or "").strip()
        if not scheme or not cert_id:
            continue
        out[(scheme, cert_id)] = str(doc.get("parse_status") or "").strip()
    return out


# Events whose certificate may carry a register suspension window. The window lives on the
# certificate's OWN register record and survives reinstatement, so an episode's real dates are
# recoverable even after ISCC has put the certificate back in the valid list.
_WINDOW_EVENT_TYPES = {
    "certificate_suspended",
    "certificate_reinstated",
    "certificate_withdrawn",
    "certificate_revoked_or_terminated",
}


def _enrich_iscc_suspension_windows(events: list[dict], run_id: str | None) -> dict:
    """Stamp the register's suspension window onto this run's ISCC compliance events, and keep it
    in the durable window store.

    Bounded by construction: only certificates that already produced a compliance event this run
    are probed (2-7 on observed nights), one request each. Fail-visible but never fatal — an
    enrichment lookup must not be able to fail a refresh.
    """
    try:
        from sinetti_services import iscc_suspension_windows as isw

        wanted = [
            str(event.get("certificate_id") or "").strip()
            for event in events
            if str(event.get("scheme") or "") == "ISCC"
            and str(event.get("event_type") or "") in _WINDOW_EVENT_TYPES
        ]
        wanted = [cid for cid in wanted if cid]
        if not wanted:
            return {"status": "skipped", "reason": "no compliance events", "probes": 0}

        dropped: list[list[str]] = []
        observed_at = run_id or _now_iso()
        found = isw.resolve_windows(
            wanted,
            fetch=iscc_fetch.fetch_certificate_record,
            observed_at=observed_at,
            max_probes=isw.max_probes_from_env(),
            on_drop=dropped.append,
        )
        for event in events:
            if str(event.get("scheme") or "") != "ISCC":
                continue
            window = found.get(str(event.get("certificate_id") or "").strip())
            if not window:
                continue
            for key in isw.WINDOW_FIELDS:
                if window.get(key):
                    event[key] = window[key]

        report = isw.merge_windows(
            state_paths.refresh_dir() / "iscc_suspension_windows.json", found,
        )
        report.update({"status": "ok", "probes": len(found), "requested": len(set(wanted))})
        if dropped:
            report["skipped_over_cap"] = sorted({cid for batch in dropped for cid in batch})
        return report
    except Exception as exc:  # noqa: BLE001 — reported, refresh continues
        return {"status": "error", "error": f"{type(exc).__name__}: {exc}"}


def _build_change_events_with_soft_alerts(
    run_id: str,
    scheme: str,
    old_certs: list[dict],
    new_certs: list[dict],
    **kwargs,
) -> list[dict]:
    """Build durable change events and fail-open the sibling Soft inbox hook."""
    events = refresh_quality.build_change_events(
        run_id, scheme, old_certs, new_certs, **kwargs
    )
    try:
        emissions = soft_alert_service.process_soft_deltas_for_refresh(
            old_certs=old_certs,
            new_certs=new_certs,
            detected_at=kwargs.get("detected_at"),
            scheme=scheme,
            parse_status_by_cert=_parse_status_by_cert_for_soft_alerts(),
        )
        notifications.dispatch_soft_immediate_emissions(
            emissions,
            scheme=scheme,
            occurred_at=kwargs.get("detected_at"),
        )
    except Exception:
        # Alerts are advisory: state-store or membership failures cannot reject a
        # successfully validated register snapshot.
        pass
    return events


def _serving_pair_revision(candidate_root: Path) -> str:
    data_dir = Path(candidate_root) / "data"
    identity = trust_review_service.serving_pair_identity(
        data_dir / "sinetti_app.sqlite.gz",
        data_dir / "sinetti_data_release.json",
        allow_missing_manifest=True,
    )
    return str(identity["candidate_revision"])


def _candidate_changed_upload_report(review: dict, candidate_root: Path) -> dict | None:
    reviewed = str(review.get("candidate_revision") or "")
    if not reviewed:
        return None
    current = _serving_pair_revision(candidate_root)
    if current == reviewed:
        return None
    return {
        "status": "error-candidate-changed-after-review",
        "reviewed_candidate_revision": reviewed,
        "current_candidate_revision": current,
        "error": "candidate serving pair changed after Trust Review and before upload",
    }


def _snapshot_reviewed_candidate(
    review: dict, candidate_root: Path
) -> tuple[tempfile.TemporaryDirectory | None, Path | None, dict | None]:
    reviewed = str(review.get("candidate_revision") or "")
    if not reviewed:
        return None, None, None
    temp = tempfile.TemporaryDirectory(prefix="sinetti-reviewed-release-")
    snapshot_root = Path(temp.name)
    for rel, source in data_plane.iter_artifact_files(Path(candidate_root)):
        destination = snapshot_root / rel
        destination.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(source, destination)
    proof = review.get("publication_artifact_ids")
    if proof:
        catalog = snapshot_root / "data" / "cert_format_catalog.json"
        shutil.copy2(candidate_root / "data" / catalog.name, catalog)
        if refresh_publication.capture_review_identity(snapshot_root / "data")["artifact_ids"] != proof:
            temp.cleanup()
            return None, None, {"status": "error-candidate-changed-after-review",
                                "error": "candidate artifacts changed after Trust Review"}
    current = _serving_pair_revision(snapshot_root)
    if current != reviewed:
        temp.cleanup()
        return None, None, {
            "status": "error-candidate-changed-after-review",
            "reviewed_candidate_revision": reviewed,
            "current_candidate_revision": current,
            "error": "candidate serving pair changed after Trust Review and before upload",
        }
    return temp, snapshot_root, None


def _copy_legacy_space_tree(source_root: Path, snapshot_root: Path) -> None:
    """Copy the legacy Space upload root so upload reads immutable reviewed bytes."""
    ignored_names = {
        ".git",
        ".pytest_cache",
        ".mypy_cache",
        ".ruff_cache",
        ".venv",
        "__pycache__",
        "node_modules",
        # Local-only state, never Space content. .worktrees holds whole parked nested
        # checkouts — copying them ships stale code copies, and their transient dirs can be
        # access-denied on Windows, failing the entire copytree (observed 2026-07-23:
        # .worktrees/*/.superpowers/sdd/pytest-tmp* -> WinError 5 -> upload step error).
        ".worktrees",
        ".claude",
        ".superpowers",
        # Gitignored local data-plane state: the pull cache (current/previous/merge-backups
        # plus stray tool sandboxes — data-cache/codex-pytest/* are access-denied on this
        # machine and fail the copytree, observed 2026-07-23) and archive publication
        # workspaces. Never Space content.
        "data-cache",
        "data-backfill",
    }

    def ignore(_directory: str, names: list[str]) -> set[str]:
        return {
            name
            for name in names
            if name in ignored_names or name.endswith(".pyc")
        }

    shutil.copytree(source_root, snapshot_root, dirs_exist_ok=True, ignore=ignore)


def _snapshot_reviewed_legacy_space(
    review: dict, candidate_root: Path
) -> tuple[tempfile.TemporaryDirectory | None, Path | None, dict | None]:
    reviewed = str(review.get("candidate_revision") or "")
    if not reviewed:
        return None, None, None
    temp = tempfile.TemporaryDirectory(prefix="sinetti-reviewed-space-")
    snapshot_root = Path(temp.name)
    _copy_legacy_space_tree(Path(candidate_root), snapshot_root)
    current = _serving_pair_revision(snapshot_root)
    if current != reviewed:
        temp.cleanup()
        return None, None, {
            "status": "error-candidate-changed-after-review",
            "reviewed_candidate_revision": reviewed,
            "current_candidate_revision": current,
            "error": "candidate serving pair changed after Trust Review and before upload",
        }
    return temp, snapshot_root, None


def _gzip_json_len(path: Path) -> int:
    if not path.exists():
        return 0
    with gzip.open(path, "rt", encoding="utf-8") as fh:
        data = json.load(fh)
    return len(data) if isinstance(data, list) else 0


def _read_gzip_json_list(path: Path) -> list[dict]:
    if not path.exists():
        return []
    try:
        with gzip.open(path, "rt", encoding="utf-8") as fh:
            data = json.load(fh)
        return data if isinstance(data, list) else []
    except Exception:
        return []


def _write_gzip_json(path: Path, rows: list[dict]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with gzip.open(path, "wt", encoding="utf-8") as fh:
        json.dump(rows, fh, ensure_ascii=False)


def _iscc_bundle_rows(path: Path) -> list[dict]:
    if not path.exists():
        return []
    try:
        return iscc_fetch.load_bundle(path)
    except Exception:
        return []


def _duration_s(start: float) -> float:
    return round(time.monotonic() - start, 3)


def _failure_report(*, decision: str, validation: dict | None = None,
                    exc: BaseException | None = None, duration_s: float = 0.0) -> dict:
    out = {
        "status": "error",
        "decision": decision,
        "duration_s": duration_s,
        "validation": validation or {"status": "critical", "checks": []},
    }
    if exc is not None:
        out["exception"] = {"class": type(exc).__name__, "message": str(exc)}
    return out


# Fixed 6-request nightly probe of ISCC certificate archive per-status totals -- trend evidence
# for the archive-growth check, not an unbounded capped fetch (owner-approved deviation from the
# bulk-fetch opt-in-zero convention, 2026-08-13). None means "all statuses".
_ARCHIVE_PROBE_STATUSES: tuple[str | None, ...] = (
    None, "expired", "valid", "withdrawn", "terminated", "suspended",
)


def _archive_probe_key(status: str | None) -> str:
    return status or "all"


def _prior_archive_totals(history_dir: str | Path) -> dict[str, int]:
    path = Path(history_dir) / "compliance_history.jsonl"
    result = refresh_quality.read_jsonl(path)
    prior: dict[str, int] = {}
    for record in result.get("records", []):
        if not isinstance(record, dict) or record.get("scheme") != "ISCC":
            continue
        kind = str(record.get("kind") or "")
        if not kind.startswith("archive:"):
            continue
        try:
            prior[kind] = int(record.get("count"))
        except (TypeError, ValueError):
            continue
    return prior


def _run_iscc_archive_totals_probe(*, run_id: str, history_dir: str | Path) -> dict:
    """Advisory nightly probe of ISCC certificate archive per-status totals.

    Fixed 6 requests. Never blocks refresh -- caller wraps this in try/except too, but any
    exception here is already caught and reported as {"status": "error", "error": ...}.
    Kill switch: env ISCC_ARCHIVE_PROBE (default on; "0" disables).
    """
    if not _env_bool("ISCC_ARCHIVE_PROBE", True):
        return {"status": "skipped-disabled"}
    try:
        prior = _prior_archive_totals(history_dir)
        totals: dict[str, int] = {}
        history_rows: list[dict] = []
        for index, status in enumerate(_ARCHIVE_PROBE_STATUSES):
            page = iscc_fetch.fetch_certificate_archive_page(1, status=status)
            key = _archive_probe_key(status)
            count = int(page["total_count"])
            totals[key] = count
            history_rows.append({
                "run_id": run_id,
                "scheme": "ISCC",
                "kind": f"archive:{key}",
                "count": count,
                "source": "ISCC certificate archive API",
            })
            if index < len(_ARCHIVE_PROBE_STATUSES) - 1:
                time.sleep(0.5)
        warnings = []
        for key, count in totals.items():
            prior_count = prior.get(f"archive:{key}")
            if prior_count is not None and count < prior_count:
                warnings.append({"kind": f"archive:{key}", "previous": prior_count, "current": count})
        return {"status": "ok", "totals": totals, "history_rows": history_rows, "warnings": warnings}
    except Exception as exc:
        return {"status": "error", "error": f"{type(exc).__name__}: {exc}"}


def _refresh_iscc(steps: list[str] | None = None, *, run_id: str = "") -> dict:
    selected = steps or ["certs", "compliance", "blocklist"]
    started = time.monotonic()
    out: dict = {
        "status": "ok",
        "decision": "accepted",
        "duration_s": 0.0,
        "validation": {"status": "ok", "checks": []},
        "fingerprints": [],
        "change_events": [],
        "compliance_history": [],
    }
    data_dir = ROOT / "data" / "iscc"
    validations: list[dict] = []
    certs_diff: tuple[list[dict], list[dict]] | None = None
    authority_rows: list[dict] | None = None
    previous_compliance: list[dict] | None = None
    prior_coverage = prior_coverage_for_refresh(
        "ISCC", refresh_dir=state_paths.refresh_dir(), archive_sqlite=None,
    )
    try:
        if "certs" in selected:
            path = data_dir / "snapshot.csv.gz"
            old_rows = _iscc_bundle_rows(path)
            text = iscc_fetch.fetch_certs_csv_text()
            validation = refresh_quality.validate_iscc_csv_text(text, previous_count=len(old_rows))
            validations.append(validation)
            if refresh_quality.validation_is_critical(validation):
                out.update(_failure_report(
                    decision="kept_previous",
                    validation=validation,
                    duration_s=_duration_s(started),
                ))
                return out
            out["certs"] = iscc_fetch.write_certs_bundle(
                text, path, fetched_at=(run_id or None), source="daily-refresh")
            new_rows = _iscc_bundle_rows(path)
            out["fingerprints"].append(refresh_quality.fingerprint_record(
                run_id=run_id, scheme="ISCC", kind="certs", path=path, count=out["certs"],
            ))
            # Defer change events until after compliance so removals can be upgraded.
            certs_diff = (old_rows, new_rows)
            # History retention: the valid-only feed drops lapsed/removed certs, so keep their
            # full last-seen rows in the append-only disappearance ledger (merged into the
            # historical ISCC archive table at app-db build). Fail-visible, never blocks refresh.
            try:
                from sinetti_services import iscc_disappearance_ledger
                out["disappearance_ledger"] = iscc_disappearance_ledger.append_disappearances(
                    data_dir / "disappeared.json.gz", old_rows, new_rows,
                    detected_at=(run_id or _now_iso()),
                )
            except Exception as exc:  # noqa: BLE001 — reported, refresh continues
                out["disappearance_ledger"] = {
                    "status": "error", "error": f"{type(exc).__name__}: {exc}",
                }
        if "compliance" in selected:
            path = data_dir / "compliance.json.gz"
            previous_compliance = _read_gzip_json_list(path)  # noqa: F841 — reused below
            old_count = len(previous_compliance) if previous_compliance else _gzip_json_len(path)
            flat: list[dict] = []
            for status in iscc_fetch.COMPLIANCE_STATUSES:
                flat.extend(iscc_fetch.fetch_compliance_certs(status))
            validation = refresh_quality.validate_compliance_records(flat, previous_count=old_count)
            validations.append(validation)
            if refresh_quality.validation_is_critical(validation):
                if certs_diff is not None:
                    old_rows, new_rows = certs_diff
                    out["change_events"].extend(_build_change_events_with_soft_alerts(
                        run_id or _now_iso(), "ISCC", old_rows, new_rows,
                        detected_at=run_id or None,
                        status_authority=previous_compliance,
                        authority_source="ISCC compliance register",
                        prior_coverage=prior_coverage,
                    ))
                out.update(_failure_report(
                    decision="kept_previous",
                    validation=validation,
                    duration_s=_duration_s(started),
                ))
                return out
            _write_gzip_json(path, flat)
            import bundle_meta  # noqa: E402

            # The release records each register's fetch date from this sidecar; without it
            # ISCC Compliance shipped undated and the promotion guard could not compare it.
            bundle_meta.write_meta(
                path, rows=len(flat), source="ISCC compliance register",
                fetched_at=(run_id or None),
            )
            out["compliance"] = len(flat)
            out["fingerprints"].append(refresh_quality.fingerprint_record(
                run_id=run_id, scheme="ISCC", kind="compliance", path=path, count=len(flat),
            ))
            authority_rows = flat
            counts = Counter(str(c.get("status") or c.get("raw_status") or "").lower() for c in flat)
            for kind, count in sorted(counts.items()):
                out["compliance_history"].append({
                    "run_id": run_id,
                    "scheme": "ISCC",
                    "kind": kind or "unknown",
                    "count": int(count),
                    "source": "ISCC compliance register",
                })
            probe = _run_iscc_archive_totals_probe(
                run_id=run_id, history_dir=state_paths.refresh_dir(),
            )
            if probe.get("status") == "ok":
                out["archive_totals"] = probe["totals"]
                out["compliance_history"].extend(probe["history_rows"])
                if probe["warnings"]:
                    out["archive_totals_warnings"] = probe["warnings"]
            elif probe.get("status") == "error":
                out["archive_totals"] = {"status": "error", "error": probe.get("error")}
        if certs_diff is not None:
            old_rows, new_rows = certs_diff
            if authority_rows is None:
                authority_rows = _read_gzip_json_list(data_dir / "compliance.json.gz")
            out["change_events"].extend(_build_change_events_with_soft_alerts(
                run_id or _now_iso(), "ISCC", old_rows, new_rows,
                detected_at=run_id or None,
                status_authority=authority_rows,
                authority_source="ISCC compliance register",
                prior_coverage=prior_coverage,
            ))
            # A lifted ISCC suspension is invisible in the valid-only CSV (its Status column is
            # always "1"), so read reinstatements from the compliance register's own diff. Only
            # meaningful when this run actually refetched the register: without a fresh copy,
            # previous_compliance IS authority_rows and the diff is empty by construction.
            if previous_compliance is not None and authority_rows is not None:
                out["change_events"].extend(refresh_quality.build_compliance_reversal_events(
                    run_id or _now_iso(), "ISCC", previous_compliance, authority_rows, new_rows,
                    detected_at=run_id or None,
                    authority_source="ISCC compliance register",
                ))
            # That diff is blind to an episode that opened and closed while the accepted baseline
            # was frozen (a run of failed publishes). The append-only history is not. Only with a
            # FRESHLY fetched register: a stale copy may still omit or list anything.
            if previous_compliance is not None and authority_rows is not None:
                try:
                    out["change_events"].extend(
                        refresh_quality.build_orphaned_suspension_reinstatements(
                            run_id or _now_iso(), "ISCC",
                            load_change_events_jsonl(
                                state_paths.refresh_dir() / "change_events.jsonl"),
                            authority_rows, new_rows,
                            detected_at=run_id or None,
                            authority_source="ISCC compliance register",
                            skip_certificate_ids={
                                str(event.get("certificate_id") or "")
                                for event in out["change_events"]
                                if event.get("event_type") == "certificate_reinstated"
                            },
                        ))
                except Exception as exc:  # noqa: BLE001 — reported, refresh continues
                    out["orphaned_reinstatements"] = {
                        "status": "error", "error": f"{type(exc).__name__}: {exc}",
                    }
            # The compliance register lists only a RUNNING suspension, but the certificate's own
            # record keeps both ends afterwards — so a finished episode still has real dates.
            out["suspension_windows"] = _enrich_iscc_suspension_windows(
                out["change_events"], run_id,
            )
        if "blocklist" in selected:
            path = data_dir / "blocklist.json.gz"
            old_count = _gzip_json_len(path)
            blocklist = iscc_fetch.fetch_all_blocklists()
            validation = refresh_quality.validate_blocklist_records(blocklist, previous_count=old_count)
            validations.append(validation)
            if refresh_quality.validation_is_critical(validation):
                out.update(_failure_report(
                    decision="kept_previous",
                    validation=validation,
                    duration_s=_duration_s(started),
                ))
                return out
            _write_gzip_json(path, blocklist)
            out["blocklist"] = len(blocklist)
            out["fingerprints"].append(refresh_quality.fingerprint_record(
                run_id=run_id, scheme="ISCC", kind="blocklist", path=path, count=len(blocklist),
            ))
            counts = Counter(str(e.get("category") or "unknown") for e in blocklist)
            for kind, count in sorted(counts.items()):
                out["compliance_history"].append({
                    "run_id": run_id,
                    "scheme": "ISCC",
                    "kind": f"blocklist:{kind}",
                    "count": int(count),
                    "source": "ISCC integrity blocklist",
                })
    except Exception as exc:
        out.update(_failure_report(
            decision="kept_previous",
            exc=exc,
            duration_s=_duration_s(started),
        ))
        return out
    if "certs" in selected:
        out["pdf_links"] = _refresh_iscc_pdf_links(data_dir)
    out["validation"] = refresh_quality.combine_validations(*validations)
    out["duration_s"] = _duration_s(started)
    return out


def _refresh_iscc_pdf_links(data_dir: Path) -> dict:
    """Give every ISCC certificate its hub PDF link, capped per night.

    Before 2026-09-21 the link map was written once by hand and never again, so every certificate
    issued after 27 Jun 2026 had no PDF link in the app — and, since the PDF queue only takes
    certificates that have one, was never parsed. Withdrawn and suspended certificates bring
    their link in the compliance register's own block; the rest cost one register request each,
    up to ISCC_PDF_LINK_MAX_LOOKUPS. The map rides the data release and the app-database build
    copies it into the serving table. Fail-visible, never fatal.
    """
    try:
        from sinetti_services import iscc_pdf_links

        path = data_dir / "cert_pdfs.json"
        compliance = _read_gzip_json_list(data_dir / "compliance.json.gz") or []
        certs = _iscc_bundle_rows(data_dir / "snapshot.csv.gz") + compliance
        known = {
            str(row.get("certificate_id") or "").strip(): str(row.get("certificate_pdf_url") or "")
            for row in compliance
        }
        links, report = iscc_pdf_links.fill_missing_links(
            certs,
            iscc_pdf_links.load_links(path),
            lookup=iscc_fetch.fetch_certificate_pdf_url,
            max_lookups=iscc_pdf_links.max_lookups_from_env(),
            known_links=known,
            sleep_s=0.1,
        )
        if report["links_after"] != report["links_before"]:
            iscc_pdf_links.write_links(path, links)
        report["missed_ids"] = report["missed_ids"][:20]  # a sample; the count is in "misses"
        return report
    except Exception as exc:  # noqa: BLE001 — reported, refresh continues
        return {"status": "error", "error": f"{type(exc).__name__}: {exc}"}


def _refresh_redcert(*, run_id: str = "") -> dict:
    started = time.monotonic()
    import redcert_fetch  # noqa: E402
    import redcert  # noqa: E402
    try:
        path = ROOT / "data" / "redcert" / "certs.json.gz"
        old_rows = redcert.load_redcert_certs()
        content = redcert_fetch.fetch_xlsx()
        raw = redcert_fetch.parse_xlsx(content)
        raw_validation = refresh_quality.validate_tabular_headers(
            "REDcert XLSX",
            raw,
            {
                "certificate_id": {"identifikator", "identifier"},
                "holder_name": {"namedesinhabers", "holder"},
                "country": {"land", "country"},
                "status": {"status"},
                "valid_until": {"gueltigbis", "gultigbis", "gültigbis", "validuntil"},
            },
        )
        certs = redcert.normalize_rows(raw)
        cert_validation = refresh_quality.validate_normalized_certs(
            "REDcert", certs, previous_count=len(old_rows),
        )
        # REDcert keeps its expired certificates in the register; our snapshot is replace-all, so
        # a register-side history trim would silently delete our history too. Critical => the
        # refresh keeps the previous snapshot.
        retention_validation = refresh_quality.validate_history_retention(
            "REDcert", old_rows, certs,
        )
        validation = refresh_quality.combine_validations(
            raw_validation, cert_validation, retention_validation,
        )
        if refresh_quality.validation_is_critical(validation):
            return _failure_report(
                decision="kept_previous",
                validation=validation,
                duration_s=_duration_s(started),
            )
        redcert.write_snapshot(certs, fetched_at=(run_id or None), source="daily-refresh")
        prior_coverage = prior_coverage_for_refresh(
            "REDcert", refresh_dir=state_paths.refresh_dir(), archive_sqlite=None,
        )
        # REDcert marks a re-issued certificate's old number withdrawn; the builder pairs it
        # with its successor and otherwise holds the withdrawal for a few nights. The hold is
        # deduped against the append-only log, so hand the builder what is already settled.
        settled = refresh_quality.settled_certificate_ids(
            load_change_events_jsonl(state_paths.refresh_dir() / "change_events.jsonl"),
            "REDcert",
        )
        return {
            "status": "ok",
            "decision": "accepted",
            "duration_s": _duration_s(started),
            "certs": len(certs),
            "validation": validation,
            "fingerprints": [refresh_quality.fingerprint_record(
                run_id=run_id, scheme="REDcert", kind="certs", path=path, count=len(certs),
            )],
            "change_events": _build_change_events_with_soft_alerts(
                run_id or _now_iso(), "REDcert", old_rows, certs, detected_at=run_id or None,
                status_authority=certs, authority_source="REDcert status field",
                prior_coverage=prior_coverage,
                settled_certificate_ids=settled,
            ),
            "compliance_history": [
                {
                    "run_id": run_id,
                    "scheme": "REDcert",
                    "kind": kind or "unknown",
                    "count": int(count),
                    "source": "REDcert status field",
                }
                for kind, count in sorted(Counter(str(c.get("status") or "").lower() for c in certs).items())
            ],
        }
    except Exception as exc:
        return _failure_report(decision="kept_previous", exc=exc, duration_s=_duration_s(started))


def _refresh_lapsing_register(
    *, scheme: str, certs: list[dict], old_rows: list[dict], path: Path,
    write_snapshot, raw_validation: dict | None, run_id: str, started: float,
) -> dict:
    """Shared tail of the 2BSvs and RSB refreshes -- exactly the REDcert sequence.

    Both registers publish their lapsed certificates next to the valid ones, and our snapshot
    write is replace-all, so a register-side history trim would silently delete our history too
    (critical => the refresh keeps the previous snapshot). The certificate number survives a
    renewal in both registers, so there is no REDcert-style replacement pairing here.
    """
    cert_validation = refresh_quality.validate_normalized_certs(
        scheme, certs, previous_count=len(old_rows),
    )
    retention_validation = refresh_quality.validate_history_retention(scheme, old_rows, certs)
    validation = refresh_quality.combine_validations(
        *([raw_validation] if raw_validation else []), cert_validation, retention_validation,
    )
    if refresh_quality.validation_is_critical(validation):
        return _failure_report(
            decision="kept_previous", validation=validation, duration_s=_duration_s(started),
        )
    write_snapshot(certs, fetched_at=(run_id or None), source="daily-refresh")
    prior_coverage = prior_coverage_for_refresh(
        scheme, refresh_dir=state_paths.refresh_dir(), archive_sqlite=None,
    )
    settled = refresh_quality.settled_certificate_ids(
        load_change_events_jsonl(state_paths.refresh_dir() / "change_events.jsonl"), scheme,
    )
    source_label = f"{scheme} register list"
    return {
        "status": "ok",
        "decision": "accepted",
        "duration_s": _duration_s(started),
        "certs": len(certs),
        "validation": validation,
        "fingerprints": [refresh_quality.fingerprint_record(
            run_id=run_id, scheme=scheme, kind="certs", path=path, count=len(certs),
        )],
        "change_events": _build_change_events_with_soft_alerts(
            run_id or _now_iso(), scheme, old_rows, certs, detected_at=run_id or None,
            status_authority=certs, authority_source=source_label,
            prior_coverage=prior_coverage,
            settled_certificate_ids=settled,
        ),
        "compliance_history": [
            {
                "run_id": run_id,
                "scheme": scheme,
                "kind": kind or "unknown",
                "count": int(count),
                "source": source_label,
            }
            for kind, count in sorted(Counter(str(c.get("status") or "").lower() for c in certs).items())
        ],
    }


def _refresh_2bsvs(*, run_id: str = "") -> dict:
    started = time.monotonic()
    import twobsvs  # noqa: E402
    import twobsvs_fetch  # noqa: E402
    try:
        path = ROOT / "data" / "2bsvs" / "certs.json.gz"
        old_rows = twobsvs.load_2bsvs_certs()
        # fetch_lists raises RegisterFetchError on a short/changed valid list, which lands in the
        # except below as a failed scheme: an empty or reshaped register never publishes.
        certs = twobsvs_fetch.build_certs(twobsvs_fetch.fetch_lists())
        return _refresh_lapsing_register(
            scheme="2BSvs", certs=certs, old_rows=old_rows, path=path,
            write_snapshot=twobsvs.write_snapshot, raw_validation=None,
            run_id=run_id, started=started,
        )
    except Exception as exc:
        return _failure_report(decision="kept_previous", exc=exc, duration_s=_duration_s(started))


def _refresh_rsb(*, run_id: str = "") -> dict:
    started = time.monotonic()
    import rsb  # noqa: E402
    import rsb_fetch  # noqa: E402
    try:
        path = ROOT / "data" / "rsb" / "certs.json.gz"
        old_rows = rsb.load_rsb_certs()
        http_client = rsb_fetch.UrllibHttp()
        # fetch_tables raises RsbFetchError on a missing/reshaped Airtable view (same rule).
        valid, lapsed = rsb_fetch.fetch_tables(http_client)
        certs = rsb.normalize_rows(valid + lapsed)
        # The holder-country cache is refresh-written (it rides the release); a PDF that fails
        # stays out of the cache and is retried tomorrow, so it is saved even if the snapshot
        # is later rejected.
        cache, _failures = rsb_fetch.enrich_addresses(
            certs, http_client, cache=rsb_fetch.load_address_cache(),
        )
        rsb_fetch.save_address_cache(cache)
        certs = rsb.apply_address_cache(certs, cache)
        return _refresh_lapsing_register(
            scheme="RSB", certs=certs, old_rows=old_rows, path=path,
            write_snapshot=rsb.write_snapshot, raw_validation=None,
            run_id=run_id, started=started,
        )
    except Exception as exc:
        return _failure_report(decision="kept_previous", exc=exc, duration_s=_duration_s(started))


def _refresh_redcert_pdfs(*, run_id: str = "", max_fetches: int = DEFAULT_REDCERT_PDF_MAX_FETCHES,
                          should_continue=None) -> dict:
    """Lazy-cache bulk fill: prune non-active entries, then fetch+store PDFs for active REDcert
    certs not already resolved, capped per run. Runs after refresh_databases() so it sees this
    run's freshest cert list."""
    started = time.monotonic()
    try:
        # Split the legacy single-file store into append-only segments once, before anything writes
        # to it. Idempotent, so every later run is a no-op.
        migration = redcert_pdf_store.migrate_monolith_to_segments()
        certs = redcert.load_redcert_certs()
        in_force_ids = redcert_pdf_store.select_in_force_certificate_ids(certs)
        prune_report = redcert_pdf_store.prune_not_active(in_force_ids)
        index = redcert_pdf_store.load_index()
        pending = [
            cid for cid in in_force_ids
            if index.get(cid, {}).get("status") not in
            (redcert_pdf_store.STATUS_HOSTED, redcert_pdf_store.STATUS_NO_DOCUMENT)
        ]
        batch = pending[:max_fetches]
        # The register drops requests in bursts (2026-09-29: answers stopped after ~300 of ~630).
        # A dropped answer is retried (20 s, then 40 s) within the process-wide retry budget,
        # which this fill shares with the PDF delta refresh.
        retry_budget = redcert_pdf.process_retry_budget()
        report = redcert_pdf_store.refresh_in_force(
            batch, fetch=lambda cid: redcert_pdf.fetch_pdf_detailed_with_retry(cid, budget=retry_budget),
            should_continue=should_continue)
        # Reclaim space from pruned certificates, one sealed segment per run at most.
        compaction = redcert_pdf_store.compact_segments()
        return {
            "status": "ok",
            "duration_s": _duration_s(started),
            "in_force_total": len(in_force_ids),
            "pending_before": len(pending),
            "batch_size": len(batch),
            "pruned": prune_report.get("pruned", 0),
            "kept": prune_report.get("kept", 0),
            "migration": migration,
            "compaction": compaction,
            "retry": retry_budget.snapshot(),
            **report,
        }
    except Exception as exc:
        return {
            "status": "error",
            "duration_s": _duration_s(started),
            "exception": {"class": type(exc).__name__, "message": str(exc)},
        }


def _refresh_ins(*, run_id: str = "", source: str = "daily-refresh",
                 data_root: Path | None = None) -> dict:
    started = time.monotonic()
    import ins_fetch  # noqa: E402
    import ins  # noqa: E402
    try:
        path = Path(data_root or (ROOT / "data")) / "ins" / "certs.json.gz"
        old_rows = _read_gzip_json_list(path)
        blobs = ins_fetch.fetch_blobs()
        blob_validation = refresh_quality.validate_ins_blobs(blobs)
        certs = ins.normalize_operatori(
            blobs.get("operatori"), blobs.get("tipologie"), blobs.get("prodotti"),
        )
        cert_validation = refresh_quality.validate_normalized_certs(
            "INS", certs, previous_count=len(old_rows),
        )
        validation = refresh_quality.combine_validations(blob_validation, cert_validation)
        if refresh_quality.validation_is_critical(validation):
            return _failure_report(
                decision="kept_previous",
                validation=validation,
                duration_s=_duration_s(started),
            )
        _write_gzip_json(path, certs)
        import bundle_meta  # noqa: E402

        bundle_meta.write_meta(
            path,
            rows=len(certs),
            source=source,
            fetched_at=(run_id or None),
        )
        prior_coverage = prior_coverage_for_refresh(
            "INS", refresh_dir=state_paths.refresh_dir(), archive_sqlite=None,
        )
        return {
            "status": "ok",
            "decision": "accepted",
            "duration_s": _duration_s(started),
            "certs": len(certs),
            "validation": validation,
            "fingerprints": [refresh_quality.fingerprint_record(
                run_id=run_id, scheme="INS", kind="certs", path=path, count=len(certs),
                records=certs,
            )],
            "change_events": _build_change_events_with_soft_alerts(
                run_id or _now_iso(), "INS", old_rows, certs, detected_at=run_id or None,
                status_authority=certs, authority_source="INS status field",
                prior_coverage=prior_coverage,
            ),
            "compliance_history": [
                {
                    "run_id": run_id,
                    "scheme": "INS",
                    "kind": kind or "unknown",
                    "count": int(count),
                    "source": "INS status field",
                }
                for kind, count in sorted(Counter(str(c.get("status") or "").lower() for c in certs).items())
            ],
        }
    except Exception as exc:
        return _failure_report(decision="kept_previous", exc=exc, duration_s=_duration_s(started))


def _pdf_delta_queue_depth(report: dict) -> dict[str, int]:
    """Documents still unfetched/unparsed after this run's nightly cap, per scheme -- summed
    across the certificate PDF delta queue and the (ISCC-only) audit report delta queue, which
    select independently and can both carry a backlog for the same scheme.

    This is distinct from sinetti_services/pdf_backlog_snapshots.py's broader remaining-by-reason
    snapshot (parser/review/acquisition backlog across ALL known documents): this number answers
    "is the nightly cap keeping up with what the delta queue found this cycle", not "how much of
    the whole corpus still needs work"."""
    depth: dict[str, int] = {}
    for key in ("pdf_delta", "audit_delta"):
        payload = report.get(key)
        delta = payload.get("delta") if isinstance(payload, dict) else None
        if not isinstance(delta, dict):
            continue
        for scheme, count in (delta.get("remaining_by_scheme") or {}).items():
            depth[scheme] = depth.get(scheme, 0) + int(count)
    return depth


def _record_accepted_backlog_snapshots(
    refresh_results: dict,
    *,
    refresh_dir: str | Path,
    run_id: str,
    as_of: str,
    details_db_path: str | Path | None = None,
    # scheme-registry-ok: deliberate subset - only ISCC + REDcert carry a PDF backlog to snapshot.
    schemes: tuple[str, ...] = ("iscc", "redcert"),
) -> dict[str, dict | None]:
    """Record PDF backlog snapshots for schemes whose register refresh was accepted."""
    from sinetti_services import data_trust_control_service
    from sinetti_services.data_trust_control_service import DEFAULT_DETAILS_ARTIFACT, IN_FORCE_COUNT_SCHEMES
    from sinetti_services.pdf_backlog_snapshots import record_refresh_snapshot

    scheme_labels = _REVIEW_SCHEME_NAMES
    db_path = Path(details_db_path) if details_db_path is not None else DEFAULT_DETAILS_ARTIFACT
    out: dict[str, dict | None] = {}
    in_force: dict[str, set[str]] | None = None
    try:
        # SIN-0971: ISCC and REDcert wait only for in-force certificates, judged on the app
        # artifact this run just rebuilt. Advisory: a failed read records today's unfiltered count.
        in_force = data_trust_control_service.in_force_ids_by_scheme(_candidate_app_artifact())
    except Exception:  # noqa: BLE001 - refresh must not fail on the in-force read
        in_force = None
    for key, result in refresh_results.items():
        if key not in schemes or not isinstance(result, dict):
            continue
        if result.get("decision") != "accepted":
            continue
        scheme = scheme_labels.get(str(key).lower(), str(key))
        try:
            out[key] = record_refresh_snapshot(
                refresh_dir=refresh_dir,
                scheme=scheme,
                run_id=run_id,
                as_of=as_of,
                details_db_path=db_path,
                in_force_ids=(
                    in_force.get(scheme) if in_force is not None and scheme in IN_FORCE_COUNT_SCHEMES else None
                ),
            )
        except Exception as exc:  # noqa: BLE001 — refresh must not fail on snapshot errors
            out[key] = {
                "status": "error",
                "exception": {"class": type(exc).__name__, "message": str(exc)[:240]},
            }
    return out


def _record_accepted_audit_backlog_snapshot(
    refresh_results: dict,
    *,
    refresh_dir: str | Path,
    run_id: str,
    as_of: str,
    details_db_path: str | Path | None = None,
) -> dict | None:
    """Record an ISCC audit-report snapshot after an accepted ISCC refresh."""
    from sinetti_services.audit_backlog_snapshots import record_refresh_audit_snapshot
    from sinetti_services.data_trust_control_service import DEFAULT_DETAILS_ARTIFACT

    iscc = refresh_results.get("iscc") if isinstance(refresh_results, dict) else None
    if not isinstance(iscc, dict) or iscc.get("decision") != "accepted":
        return None
    db_path = Path(details_db_path) if details_db_path is not None else DEFAULT_DETAILS_ARTIFACT
    try:
        return record_refresh_audit_snapshot(
            refresh_dir=refresh_dir,
            run_id=run_id,
            as_of=as_of,
            details_db_path=db_path,
        )
    except Exception as exc:  # noqa: BLE001 — refresh must not fail on snapshot errors
        return {
            "status": "error",
            "exception": {"class": type(exc).__name__, "message": str(exc)[:240]},
        }


def refresh_databases(schemes: list[str], *, run_id: str = "") -> dict:
    """Refresh selected scheme bundles and return per-scheme status objects."""
    out: dict[str, dict] = {}
    for scheme in schemes:
        try:
            if scheme == "iscc":
                payload = _refresh_iscc(run_id=run_id)
            elif scheme == "redcert":
                payload = _refresh_redcert(run_id=run_id)
            elif scheme == "ins":
                payload = _refresh_ins(run_id=run_id)
            elif scheme == "2bsvs":
                payload = _refresh_2bsvs(run_id=run_id)
            elif scheme == "rsb":
                payload = _refresh_rsb(run_id=run_id)
            else:
                out[scheme] = {"status": "skipped-unknown-scheme"}
                continue
            out[scheme] = payload if payload.get("status") else {"status": "ok", **payload}
        except Exception as exc:
            out[scheme] = {
                "status": "error",
                "decision": "kept_previous",
                "error": str(exc),
                "exception": {"class": type(exc).__name__, "message": str(exc)},
            }
    return out


def has_errors(report: dict) -> bool:
    return any(isinstance(v, dict) and v.get("status") == "error" for v in report.values())


def core_can_publish(refresh: dict, *, allow_partial: bool, baseline_failed: bool) -> bool:
    if baseline_failed:
        return False
    if not has_errors(refresh):
        return True
    if not allow_partial:
        return False
    return any(
        isinstance(item, dict)
        and item.get("status") == "ok"
        and item.get("decision") == "accepted"
        for item in refresh.values()
    )


def _env_job_budget_s() -> int:
    try:
        return max(0, int(os.environ.get("SINETTI_JOB_BUDGET_S") or 0))
    except (TypeError, ValueError):
        return 0


def _geocode_block_changed(block: dict, added_key: str) -> bool:
    status = str(block.get('status') or '')
    if not status or status.startswith('skipped'):
        return False
    if block.get('work_unknown'):
        return True
    added = int(block.get(added_key) or 0)
    if status in ('failed', 'error', 'google-error', 'stopped'):
        cached = int(block.get('negative_cached') or block.get('failed') or 0)
        cached += sum(int(item.get('negative_cached') or 0) for item in block.get('schemes', [])
                      if isinstance(item, dict))
        return added > 0 or cached > 0
    return added > 0 or int(block.get('attempted') or 0) > 0


def _tail_changed(report: dict) -> bool:
    overlap = report.get("pdf_stage_overlap") or {}
    if overlap.get("status") == "ok" and overlap.get("changed") and overlap.get("promoted"):
        return True
    geocode = report.get("geocode") or {}
    site = report.get("site_geocode") or {}
    pdf = report.get("pdf_delta") or {}
    audit = report.get("audit_delta") or {}
    redcert_host = report.get("redcert_pdf_hosting") or {}
    if _geocode_block_changed(geocode, "added"):
        return True
    address = report.get("address_geocode") or {}
    if _geocode_block_changed(address, "added"):
        return True
    if _geocode_block_changed(site, "geocoded"):
        return True
    post = report.get("post_pdf_geocode") or {}
    if (_geocode_block_changed(post.get("address") or {}, "added")
            or _geocode_block_changed(post.get("site") or {}, "geocoded")):
        return True
    if pdf.get("status") == "ok":
        delta = pdf.get("delta") if isinstance(pdf.get("delta"), dict) else {}
        if int(delta.get("selected") or pdf.get("parsed") or 0) > 0:
            return True
    if audit.get("status") == "ok" and int(
        audit.get("parsed") or 0
    ) + int(audit.get("failed") or 0) > 0:
        return True
    if redcert_host.get("status") == "ok" and int(redcert_host.get("fetched") or 0) > 0:
        return True
    return False


def _flush_receipt(report: dict, args, *, phase: str) -> None:
    heartbeat = Path(args.history_dir) / "job_heartbeat.json"
    flush_refresh_receipt(
        args.report_path, report, heartbeat_path=heartbeat, phase=phase,
    )


def _rebuild_app_and_review(report: dict, *, validated_app: dict | None = None) -> bool:
    """Rebuild the bundled app DB and run Trust Review. Returns True if rebuild failed."""
    try:
        report["app_db"] = validated_app if validated_app is not None else rebuild_bundled_app_database(root=ROOT, replace=True)
    except Exception as exc:
        report["app_db"] = {
            "status": "error",
            "exception": {
                "class": type(exc).__name__,
                "message": str(exc),
                "traceback": traceback.format_exc(),
            },
        }
        report["trust_review"] = {"status": "skipped", "reason": "no fixed accepted candidate"}
        return True
    retained = bool((report.get("config") or {}).get("require_retained_source_review"))
    # Trust Review expects LOADED schemes: a registered-but-empty register (NOT_YET_LOADED)
    # would make status evidence read "unavailable" and the retained review look partial.
    accepted_schemes = (
        _schemes.LOADED_CERTIFICATION_SCHEMES if retained
        else _accepted_review_schemes(report.get("refresh", {}))
    )
    if report["app_db"].get("status") == "ok" and accepted_schemes:
        report["trust_review"] = _post_refresh_review(
            candidate_root=ROOT,
            schemes=accepted_schemes,
        )
        if retained:
            report["trust_review_scope"] = {
                "source": "retained-fixed-candidate",
                "schemes": list(accepted_schemes),
                "candidate_revision": report["trust_review"].get("candidate_revision"),
            }
    else:
        report["trust_review"] = {"status": "skipped", "reason": "no fixed accepted candidate"}
    if report['app_db'].get('status') == 'ok':
        from sinetti_services.holder_coordinate_coverage import record_refresh_coverage
        report['coordinate_coverage'] = record_refresh_coverage(ROOT / 'data/sinetti_app.sqlite.gz', ROOT / 'data')
    return report["app_db"].get("status") != "ok"


def _advance_published_parent(report: dict, upload: dict) -> None:
    """Record the commit a successful publish created as the parent for the next one.

    The REDcert blob store rides its own commit behind the release, so when the side channel
    published too, ITS commit is the new head. A publish that reports no sha leaves the parent
    alone: the next publish then refuses on a stale parent, which is the safe direction.
    A rehearsal creates no Hub commit, so its placeholder sha cannot become the next parent.
    """
    if upload.get("rehearsal") or upload.get("status") != "ok":
        return
    blob_store = upload.get("blob_store") if isinstance(upload.get("blob_store"), dict) else {}
    published = (
        blob_store.get("commit_sha")
        if blob_store.get("status") == "ok"
        else None
    ) or upload.get("commit_sha")
    if published:
        report["baseline"] = {**(report.get("baseline") or {}), "hub_revision": str(published)}


def _publish_candidate(report: dict, args, *, hf_token: str) -> tuple[dict, bool]:
    enabled = refresh_publication.production_publication_enabled()
    if enabled or (report.get("config") or {}).get("require_retained_source_review"):
        review = report.get("trust_review") or {}
        if (review.get("status") not in {"completed", "completed_with_findings"}
                or not review.get("candidate_revision")
                or (enabled and (not review.get("publication_artifact_ids") or not args.data_repo_id))):
            return ({"status": "error-retained-review-required",
                     "error": "retained candidate requires a completed Trust Review with identity"}, True)
    snapshot = None
    failed = False
    upload: dict = {"status": "pending-upload"}
    try:
        snapshot, snapshot_root, changed = (
            _snapshot_reviewed_candidate(report["trust_review"], ROOT)
            if args.data_repo_id
            else _snapshot_reviewed_legacy_space(report["trust_review"], ROOT)
        )
        if changed is not None:
            failed = True
            upload = changed
        else:
            upload = upload_refresh_outputs(
                data_repo_id=args.data_repo_id,
                space_repo_id=args.hf_repo_id,
                token=hf_token,
                folder_path=snapshot_root or args.hf_folder,
                expected_parent=report["baseline"].get("hub_revision"),
                source_root=snapshot_root or ROOT,
            )
        if args.data_repo_id:
            if upload.get("status") != "ok":
                failed = True
            else:
                # A release is one commit on top of the last one we published. The core publish
                # (and the REDcert side channel riding behind it) moves the dataset HEAD, so the
                # tail publish has to parent on what we just wrote -- the job's own baseline is
                # stale by then and the Hub would refuse the commit. In rehearsal, nothing was
                # written, so keep the real baseline rather than the simulated commit marker.
                _advance_published_parent(report, upload)
                if enabled:
                    previous = getattr(args, "_publication_candidate", None)
                    args._publication_candidate = {
                        "snapshot": snapshot, "data_dir": snapshot_root / "data",
                        "review": copy.deepcopy(report["trust_review"]), "publish": copy.deepcopy(upload),
                    }
                    snapshot = None  # Keep these exact uploaded bytes through verification.
                    if getattr(args, "_core_publication_candidate", None) is None:
                        # The run's first publish is its core; keep it for the core tier (SIN-0904).
                        args._core_publication_candidate = args._publication_candidate
                    elif previous and previous is not args._core_publication_candidate:
                        previous["snapshot"].cleanup()
                    report["restart"] = {"status": "skipped-production-publication"}
                else:
                    report["restart"] = restart_space_after_publish(
                        space_repo_id=args.hf_repo_id,
                        token=hf_token,
                    )
        elif _publication_failure(upload):
            failed = True
    except Exception as exc:
        failed = True
        upload = {
            "status": "error",
            "exception": {"class": type(exc).__name__, "message": str(exc)},
        }
    finally:
        if snapshot is not None:
            snapshot.cleanup()
    return upload, failed


def _coordinate_runner_options(runner, baselines, *, kind, scheme='', address=False):
    if baselines is None or 'save_coords' not in inspect.signature(runner).parameters:
        return {}
    data_dir = ROOT / 'data'
    options = geocode_workspace.runner_options(data_dir, baselines, kind=kind, scheme=scheme)
    options['failures_path'] = (geocode_points.site_failures_nightly_path(data_dir) if kind == 'site' else
                                geocode_points.address_failures_nightly_path(data_dir, scheme) if address else
                                geocode_points.holder_failures_nightly_path(data_dir, scheme))
    return options


def run_geocode_step(run: Callable[[], dict], *, count_key: str = "added") -> dict:
    """Run one tail geocode pass; its exception becomes this step's error receipt.

    HF job 6ad268f3c662 (16 Oct 2026): one Google HTTP 400 escaped the holder pass right after
    the core publish, so the PDF delta, tail publish and auto-activation never ran. A pass that
    raised has already saved every coordinate it found but cannot count them, so ``work_unknown``
    lets the tail publish them; ``error`` holds auto-activation like ``stopped`` does.
    """
    try:
        return run()
    except Exception as exc:  # noqa: BLE001 - no geocode pass may take down the PDF stage
        # Some http.client errors quote the request URL, which carries the Google key.
        message = re.sub(r"key=[^&\s'\"]+", "key=REDACTED", str(exc))[:240]
        return {
            "status": "error",
            "attempted": 0,
            count_key: 0,
            "work_unknown": True,
            "exception": {"class": type(exc).__name__, "message": message},
        }


def _geocode_step_failed(report: dict) -> bool:
    return any((report.get(step) or {}).get("status") == "error"
               for step in ("geocode", "address_geocode", "site_geocode", "post_pdf_geocode"))


def geocode_with_budget(
    *,
    schemes: list[str],
    ledger_path: str | Path,
    daily_limit: int,
    monthly_limit: int,
    key: str,
    now: datetime | None = None,
    run_geocode: Callable[..., dict] = google_geocoder.run_geocode,
    sleep_s: float = 0.06,
    coordinate_baselines: dict | None = None,
) -> dict:
    if not key:
        return {
            "status": "skipped-no-google-key",
            "attempted": 0,
            "added": 0,
            "schemes": [],
        }

    ledger = geocode_budget.load_usage(ledger_path)
    month_calls_before = geocode_budget.month_calls(ledger, now=now)
    remaining_before = geocode_budget.remaining_calls(
        ledger,
        daily_limit=daily_limit,
        monthly_limit=monthly_limit,
        now=now,
    )
    if remaining_before <= 0:
        return {
            "status": "skipped-budget-exhausted",
            "remaining_before": 0,
            "remaining_after": 0,
            "attempted": 0,
            "added": 0,
            "schemes": [],
            "monthly_limit": int(monthly_limit),
            "month_calls_before": month_calls_before,
            "month_calls": geocode_budget.month_calls(ledger, now=now),
            "monthly_remaining_before": max(0, int(monthly_limit) - month_calls_before),
            "monthly_remaining_after": max(0, int(monthly_limit) - geocode_budget.month_calls(ledger, now=now)),
        }

    scheme_reports: list[dict] = []
    attempted_total = 0
    added_total = 0
    stop_error = ""
    for scheme in schemes:
        if scheme in _ADDRESS_BACKFILL_PENDING:
            continue
        ledger = geocode_budget.load_usage(ledger_path)
        remaining = geocode_budget.remaining_calls(
            ledger,
            daily_limit=daily_limit,
            monthly_limit=monthly_limit,
            now=now,
        )
        if remaining <= 0:
            break
        try:
            reservation_aware = "reserve_request" in inspect.signature(run_geocode).parameters
        except (TypeError, ValueError):
            reservation_aware = False
        if reservation_aware:
            result = run_geocode(
                scheme,
                remaining,
                key=key,
                sleep_s=sleep_s,
                **_coordinate_runner_options(run_geocode, coordinate_baselines, kind="holder", scheme=scheme),
                reserve_request=lambda: geocode_budget.reserve_call(
                    ledger_path,
                    daily_limit=daily_limit,
                    monthly_limit=monthly_limit,
                    now=now,
                ),
            )
        else:
            result = run_geocode(scheme, remaining, key=key, sleep_s=sleep_s, **_coordinate_runner_options(run_geocode, coordinate_baselines, kind="holder", scheme=scheme))
        attempted = max(0, int(result.get("attempted", 0)))
        added = max(0, int(result.get("added", 0)))
        scheme_reports.append(result)
        attempted_total += attempted
        added_total += added
        if attempted:
            metadata = {"scheme": scheme, "added": added, "status": result.get("status", "")}
            if reservation_aware:
                geocode_budget.append_run_summary(
                    ledger_path,
                    attempted,
                    now=now,
                    metadata=metadata,
                )
            else:
                geocode_budget.record_calls_atomic(
                    ledger_path,
                    attempted,
                    now=now,
                    metadata=metadata,
                )
        if result.get("status") == "stopped":
            stop_error = str(result.get("stop_error") or "Google refused the request")
            break

    ledger = geocode_budget.load_usage(ledger_path)
    month_calls = geocode_budget.month_calls(ledger, now=now)
    return {
        # "stopped" = Google refused (billing off, key denied, quota): from 2026-09-20 to 09-28
        # every night stopped after one request while this read "ok", so nobody noticed.
        "status": "stopped" if stop_error else "ok",
        "stop_error": stop_error,
        "remaining_before": remaining_before,
        "remaining_after": geocode_budget.remaining_calls(
            ledger,
            daily_limit=daily_limit,
            monthly_limit=monthly_limit,
            now=now,
        ),
        "attempted": attempted_total,
        "added": added_total,
        "schemes": scheme_reports,
        "monthly_limit": int(monthly_limit),
        "month_calls_before": month_calls_before,
        "month_calls": month_calls,
        "monthly_remaining_before": max(0, int(monthly_limit) - month_calls_before),
        "monthly_remaining_after": max(0, int(monthly_limit) - month_calls),
        "estimated_google_cost_usd": geocode_budget.estimate_google_geocoding_cost(month_calls),
    }


def geocode_addresses_with_budget(
    *,
    schemes: list[str],
    ledger_path: str | Path,
    daily_limit: int,
    monthly_limit: int,
    key: str,
    now: datetime | None = None,
    run_backfill: Callable[..., dict] | None = None,
    sleep_s: float = 0.06,
    coordinate_baselines: dict | None = None,
    rows: list[dict] | None = None,
    address_keys: set[tuple] | None = None,
) -> dict:
    """Per-site-address pass for multi-site holders, after the one-point-per-holder pass.

    The holder pass places each holder once; a holder certified at several addresses needs one
    point per address or every site collapses onto that single dot (Enilive, 2026-09-28: Rome,
    Gela and Porto Marghera drawn as one). Until now only a hand-run backfill filled these, so
    every site added after it stayed unmapped. Same Google budget ledger as the holder pass.
    """
    runner = run_backfill or address_geocoder.run_backfill
    if not key:
        return {"status": "skipped-no-google-key", "attempted": 0, "added": 0, "schemes": []}
    scheme_reports: list[dict] = []
    attempted_total = 0
    added_total = 0
    stop_error = ""
    status = "ok"
    for scheme in schemes:
        if scheme not in _schemes.CERTIFICATION_SLUGS or scheme in _ADDRESS_BACKFILL_PENDING:
            continue
        remaining = geocode_budget.remaining_calls(
            geocode_budget.load_usage(ledger_path),
            daily_limit=daily_limit,
            monthly_limit=monthly_limit,
            now=now,
        )
        if remaining <= 0:
            status = "budget-exhausted"
            break
        result = runner(
            scheme,
            remaining,
            key=key,
            sleep_s=sleep_s,
            **_coordinate_runner_options(runner, coordinate_baselines, kind="holder", scheme=scheme, address=True),
            **({'all_addresses': True} if 'all_addresses' in inspect.signature(runner).parameters else {}),
            **({'rows': [row for row in rows if str(row.get('scheme') or '').lower() == scheme]}
               if rows is not None else {}),
            **({'target_addresses': {(holder, address, country) for slug, holder, country, address in address_keys
                                    if slug == scheme}} if address_keys is not None else {}),
            reserve_request=lambda: geocode_budget.reserve_call(
                ledger_path,
                daily_limit=daily_limit,
                monthly_limit=monthly_limit,
                now=now,
            ),
        )
        attempted = max(0, int(result.get("attempted", 0)))
        added = max(0, int(result.get("added", 0)))
        scheme_reports.append(result)
        attempted_total += attempted
        added_total += added
        if attempted:
            geocode_budget.append_run_summary(
                ledger_path,
                attempted,
                now=now,
                metadata={"scheme": scheme, "pass": "address", "added": added,
                          "status": result.get("status", "")},
            )
        if result.get("status") == "stopped":
            status = "stopped"
            stop_error = str(result.get("stop_error") or "Google refused the request")
            break
        if result.get("status") == "budget-exhausted":
            status = "budget-exhausted"
            break
    return {
        "status": status,
        "stop_error": stop_error,
        "attempted": attempted_total,
        "added": added_total,
        "schemes": scheme_reports,
    }


SITE_LEDGER_NAME = "geocode_site_usage.json"


def site_ledger_path(ledger_path: str | Path) -> Path:
    """The site drip's own Google ledger, beside the holder ledger (same state volume)."""
    return Path(ledger_path).with_name(SITE_LEDGER_NAME)


def _record_map_coverage_snapshot(report: dict, *, refresh_dir: str | Path) -> dict:
    """In-force holders on the map per scheme, counted on the candidate this run rebuilt (SIN-0981)."""
    if (report.get("app_db") or {}).get("status") != "ok":
        return {"status": "skipped", "reason": "no rebuilt candidate"}
    from sinetti_services.map_coverage import record_refresh_snapshot

    return record_refresh_snapshot(
        refresh_dir=refresh_dir, run_id=report["started_at"], as_of=report["started_at"], root=ROOT,
    )


def _candidate_app_artifact() -> Path:
    """The serving artifact this run just rebuilt (and core-published) in the job workspace."""
    from scripts.rebuild_sinetti_app_artifact import bundled_paths
    return bundled_paths(ROOT)["artifact"]


def _geocode_candidate_inputs(artifact: Path, *, before: dict | None = None) -> dict:
    """Snapshot identities, or select only newly published PDF locations from the candidate.

    The certificate PDF runner already rebuilds this artifact; overlap promotes its rebuilt
    artifact too. Reopen it after those stages rather than reading yesterday's active release.
    Keep only identities in the initial snapshot, and only new rows in the second scan.
    """
    import producer_map

    result = {'address_ids': set(), 'site_ids': set(), 'address_rows': [], 'all_address_rows': [], 'site_rows': []}
    if not artifact.exists():
        return result
    with sinetti_data.open_serving_sqlite(artifact) as sqlite_path:
        conn = sinetti_data.connect_serving_sqlite(sqlite_path)
        conn.row_factory = sqlite3.Row
        try:
            tables = {row[0] for row in conn.execute("SELECT name FROM sqlite_master")}
            if 'certificates' in tables:
                columns = ', '.join(address_geocoder._SERVING_COLUMNS)
                for raw in conn.execute(f'SELECT {columns} FROM certificates'):
                    row = dict(raw)
                    identity = (str(row.get('scheme') or '').lower(), row.get('holder_norm'),
                                row.get('country'), producer_map.address_key(producer_map.cert_site_address(row)))
                    result['address_ids'].add(identity)
                    if before is not None:
                        result['all_address_rows'].append(row)
                    if before is not None and identity not in before['address_ids']:
                        result['address_rows'].append(row)
            if 'v_app_sites' in tables:
                present = {row[1] for row in conn.execute('PRAGMA table_info(v_app_sites)')}
                columns = ', '.join(c for c in site_geocoder._SITE_QUEUE_COLUMNS if c in present)
                for raw in conn.execute(f'SELECT {columns} FROM v_app_sites'):
                    row = dict(raw)
                    physical_id = site_geocoder.sg.physical_site_id(
                        source_site_id=str(row.get('source_site_id') or ''), address=str(row.get('address') or ''),
                        postcode=str(row.get('postcode') or ''), city=str(row.get('city') or ''),
                        country=str(row.get('site_country') or ''))
                    identity = (physical_id, *(str(row.get(key) or '').strip().casefold()
                                 for key in ('address', 'postcode', 'city', 'site_country')))
                    result['site_ids'].add(identity)
                    if before is not None and identity not in before['site_ids']:
                        result['site_rows'].append(row)
        finally:
            conn.close()
    return result


def run_post_pdf_geocode_step(args, *, before: dict, can_continue: bool,
                             should_continue: Callable[[], bool], coordinate_baselines=None,
                             skip_reason='skipped-no-time') -> dict:
    """Spend only unspent existing caps on locations first found by this run's PDF tail."""
    import rehearsal

    def skipped(status):
        return {'status': status, 'address': {'status': status, 'attempted': 0, 'added': 0},
                'site': {'status': status, 'attempted': 0, 'geocoded': 0}}

    if args.dry_run:
        return skipped('skipped-dry-run')
    if args.skip_geocode:
        return skipped('skipped')
    if rehearsal.is_rehearsal():
        return skipped('skipped-rehearsal')
    if not can_continue or not should_continue():
        return skipped(skip_reason)
    try:
        if before.get('status') == 'error':
            raise ValueError('pre-PDF location snapshot failed')
        fresh = _geocode_candidate_inputs(_candidate_app_artifact(), before=before)
    except Exception as exc:
        error = run_geocode_step(lambda: (_ for _ in ()).throw(exc))
        return {**skipped('skipped-input-error'), **error}
    result = skipped('skipped-no-delta')
    result['status'] = 'ok'
    if fresh['address_rows']:
        if not should_continue():
            return skipped('skipped-no-time')
        result['address'] = run_geocode_step(lambda: geocode_addresses_with_budget(
            schemes=_csv(args.geocode_schemes, DEFAULT_GEOCODE_SCHEMES),
            ledger_path=args.ledger_path, daily_limit=args.daily_geocode_limit,
            monthly_limit=args.monthly_geocode_limit, key=google_geocoder._key(),
            coordinate_baselines=coordinate_baselines, rows=fresh['all_address_rows'],
            address_keys=fresh['address_ids'] - before['address_ids']))
    if args.skip_site_geocode:
        result['site']['status'] = 'skipped'
    elif fresh['site_rows']:
        if not should_continue():
            result['site']['status'] = 'skipped-no-time'
        else:
            result['site'] = run_geocode_step(lambda: geocode_sites_with_budget(
                ledger_path=args.ledger_path, site_daily_limit=args.site_geocode_daily_limit,
                key=google_geocoder._key(), coordinate_baselines=coordinate_baselines,
                sites_rows=fresh['site_rows']), count_key='geocoded')
    if any(result[key]['status'] == 'error' for key in ('address', 'site')):
        result['status'] = 'error'
    return result


def geocode_sites_with_budget(
    *,
    ledger_path: str | Path,
    key: str,
    site_daily_limit: int = DEFAULT_SITE_GEOCODE_DAILY_LIMIT,
    site_artifact: str | Path | None = None,
    now: datetime | None = None,
    max_n: int | None = None,
    run_geocode_sites: Callable[..., dict] | None = None,
    sleep_s: float = 0.06,
    coordinate_baselines: dict | None = None,
    sites_rows: Iterable[dict] | None = None,
) -> dict:
    """Nightly drip of physical-site geocodes: site-accurate sites only, under their own cap.

    SIN-0809 / owner ruling SIN-0825 A (2026-10-17): the step had never been given its site list,
    so it queued 0 sites every night from 22 Jul 2026. It now streams the sites of the candidate
    it just rebuilt, geocodes only site-accurate ones (annex / operating sites / facilities, never
    a head-office pin), and spends at most ``site_daily_limit`` Google requests per UTC day ON TOP
    of the holder budget: its calls go to their own ledger (``site_ledger_path``), never the
    holder one. Code default 0 = off; the job registration sets the owner's cap (200). Placed
    sites and negative-cached failures are skipped by the runner; positions land in the nightly
    site file the release carries (``refresh/geocode/site_coords.json`` + geocode_points).
    """
    runner = run_geocode_sites or site_geocoder.run_geocode_sites
    cap = max(0, int(site_daily_limit or 0))
    if cap <= 0:
        return {"status": "skipped-disabled", "site_daily_limit": 0, "attempted": 0, "geocoded": 0}
    if not key:
        return {"status": "skipped-no-google-key", "site_daily_limit": cap, "attempted": 0, "geocoded": 0}

    ledger = site_ledger_path(ledger_path)
    # One night is one UTC day; the month ceiling only bounds the ledger, it never binds first.
    monthly = cap * 31
    remaining = geocode_budget.remaining_calls(
        geocode_budget.load_usage(ledger), daily_limit=cap, monthly_limit=monthly, now=now,
    )
    limit = remaining if max_n is None else min(int(max_n), remaining)
    if limit <= 0:
        return {"status": "skipped-budget-exhausted", "site_daily_limit": cap,
                "remaining_before": remaining, "attempted": 0, "geocoded": 0}

    def _run(rows) -> dict:
        return runner(
            max_n=limit,
            key=key,
            dry_run=False,
            sleep_s=sleep_s,
            **_coordinate_runner_options(runner, coordinate_baselines, kind="site"),
            sites_rows=rows,
            site_accurate_only=True,
            budget_ledger_path=ledger,
            daily_limit=cap,
            monthly_limit=monthly,
        )

    if sites_rows is None and site_artifact is not None:
        with site_geocoder.stream_site_rows(site_artifact) as rows:
            result = _run(rows)
    else:
        result = _run(sites_rows)
    site_month = geocode_budget.month_calls(geocode_budget.load_usage(ledger), now=now)
    holder_month = geocode_budget.month_calls(geocode_budget.load_usage(ledger_path), now=now)
    return {
        "status": result.get("status") or "ok",
        "site_daily_limit": cap,
        "remaining_before": remaining,
        "queued": int(result.get("queued") or 0),
        "todo": int(result.get("todo") or 0),
        "attempted": int(result.get("attempted") or 0),
        "geocoded": int(result.get("geocoded") or 0),
        "failed": int(result.get("failed") or 0),
        "site_month_calls": site_month,
        # Google's 10,000 free monthly requests are shared by holders and sites.
        "estimated_google_cost_usd_combined": geocode_budget.estimate_google_geocoding_cost(
            site_month + holder_month),
        "detail": result,
    }


def upload_to_hf_space(
    *,
    repo_id: str,
    token: str,
    folder_path: str | Path = ROOT,
    dry_run: bool = False,
) -> dict:
    if not repo_id:
        return {"status": "skipped-no-hf-repo"}
    if not token:
        return {"status": "skipped-no-hf-token"}
    from jobs_publication_policy import independent_runtime
    if independent_runtime():
        return {"status":"held-publication-authority","operation_id":None}
    if dry_run:
        return {"status": "dry-run", "repo_id": repo_id, "folder_path": str(folder_path)}
    if rehearsal.is_rehearsal():
        return rehearsal.skipped("Space folder upload", repo_id=repo_id,
                                 folder_path=str(folder_path))
    from huggingface_hub import HfApi

    HfApi(token=token).upload_folder(
        repo_id=repo_id,
        repo_type="space",
        folder_path=str(folder_path),
        ignore_patterns=DEFAULT_HF_IGNORE_PATTERNS,
        commit_message="chore: refresh Sinetti bundled datasets",
    )
    return {"status": "ok", "repo_id": repo_id, "folder_path": str(folder_path)}


def upload_refresh_outputs(*, data_repo_id: str, space_repo_id: str, token: str,
                           folder_path, expected_parent: str | None,
                           api_factory=None, source_root=ROOT) -> dict:
    """Dataset repo when configured (root-cure path); legacy Space-repo upload otherwise."""
    if data_repo_id:
        result = data_plane.publish(
            Path(source_root), repo_id=data_repo_id, token=token,
            producer="daily_refresh", expected_parent=expected_parent,
            api_factory=api_factory,
        )
        # The REDcert blob store rides its own commit. It is reported but never fatal: the release
        # is already published and correct. The job pull hydrates the previous segments first, so
        # this publish appends; a blob that is still absent degrades to the live register hop.
        blob_store = {"status": "skipped-release-not-published"}
        if result.get("status") == "ok":
            from jobs_publication_policy import independent_runtime
            blob_store = data_plane.publish_side_channel(
                Path(source_root), repo_id=data_repo_id, token=token,
                producer="daily_refresh", api_factory=api_factory,
                **({"expected_parent":None if result.get("rehearsal") else result.get("commit_sha")}
                   if independent_runtime() else {}),
            )
        return {"target": "dataset", "blob_store": blob_store, **result}
    result = upload_to_hf_space(repo_id=space_repo_id, token=token, folder_path=folder_path)
    return {"target": "space", **result}


def restart_space_after_publish(*, space_repo_id: str, token: str, api_factory=None) -> dict:
    """Poke the Space so the app reboots and pulls the fresh data. A failure here is a warning:
    the data is safely published; the app catches up on its next natural restart."""
    if not space_repo_id or not token:
        return {"status": "skipped"}
    if rehearsal.is_rehearsal():
        return rehearsal.skipped("Space restart", repo_id=space_repo_id)
    if api_factory is None:
        from huggingface_hub import HfApi as api_factory  # noqa: N813
    try:
        from jobs_publication_policy import guard_effect, PublicationHeld
        guard_effect("space.restart", target=space_repo_id, parent=None, payload={},
                     effect=lambda _: api_factory(token=token).restart_space(space_repo_id))
    except PublicationHeld as exc:
        return {"status":"held-publication-authority","operation_id":exc.operation_id}
    except Exception as exc:  # noqa: BLE001
        return {"status": "error", "repo_id": space_repo_id, "error": str(exc)[:240]}
    return {"status": "ok", "repo_id": space_repo_id}


def _space_base_url(repo_id: str) -> str:
    slug = repo_id.strip().lower().replace("/", "-")
    return f"https://{slug}.hf.space"


def _publication_tail_reserve_s() -> float:
    if not refresh_publication.production_publication_enabled():
        return TAIL_RESERVE_S
    minimum = TAIL_RESERVE_S + refresh_publication.publication_budget_s()
    try:
        configured = float(os.environ.get("SINETTI_REFRESH_PUBLISH_RESERVE_S", minimum))
    except ValueError:
        return minimum
    return max(minimum, configured) if configured < float("inf") else minimum


def _redcert_hosting_reserve_s() -> float:
    """Time the REDcert PDF fill leaves for the tail rebuild, its publish and the switch-on.

    The fill runs after the PDF tail has already used the tail reserve's start window, and on
    29 Oct it took 43 min unbounded: the switch-on then had 25 of its 34 min (SIN-1076)."""
    reserve = TAIL_PUBLISH_RESERVE_S
    if refresh_publication.production_publication_enabled():
        reserve += refresh_publication.publication_budget_s()
    return reserve


def _publication_failure(value) -> bool:
    """Legacy job success can hide a failed tail; inspect its actual receipts."""
    if isinstance(value, dict):
        status = str(value.get("status") or "").lower()
        failed = value.get("failed", 0)
        if (status.startswith(("error", "failed", "partial", "google-error", "interrupted", "stopped", "held", "unknown"))
                or value.get("exception") or value.get("error") or value.get("errors")
                or (isinstance(failed, (int, float)) and failed > 0)):
            return True
        return any(_publication_failure(item) for item in value.values())
    if isinstance(value, list):
        return any(_publication_failure(item) for item in value)
    return False


def _exact_nonnegative_count(value: object) -> bool:
    return type(value) is int and value >= 0


def _valid_empty_failure_envelope(value: object) -> bool:
    from cert_pdf_publication import EVIDENCE_VERSION

    return (
        isinstance(value, dict)
        and frozenset(value) == {"version", "failures"}
        and type(value.get("version")) is int
        and value["version"] == EVIDENCE_VERSION
        and isinstance(value.get("failures"), list)
        and not value["failures"]
    )


def _valid_certificate_report_shape(value: object) -> bool:
    if not isinstance(value, dict) or not _exact_nonnegative_count(value.get("failed")):
        return False
    if value["failed"] != 0 or "publication_failure_evidence" not in value:
        return True
    return _valid_empty_failure_envelope(value["publication_failure_evidence"])


def _valid_no_delta_report(value: object) -> bool:
    """Validate a producer's explicit proof that it selected no parse work."""
    if (not isinstance(value, dict) or value.get("status") != "skipped-no-delta"
            or not _exact_nonnegative_count(value.get("queued")) or value["queued"] != 0):
        return False
    for key in ("selected", "attempted", "failed"):
        if key in value and (not _exact_nonnegative_count(value[key]) or value[key] != 0):
            return False
    if ("publication_failure_evidence" in value
            and not _valid_empty_failure_envelope(value["publication_failure_evidence"])):
        return False
    return True


def _certificate_report_without_eligible_failures(value: dict) -> dict:
    """Mask only a structurally valid, eligible positive certificate failure count."""
    from cert_pdf_publication import validate_failure_evidence

    return {**value, "failed": 0} if validate_failure_evidence(value) else value


def _optional_count(value: dict, key: str) -> tuple[bool, int | None]:
    if key not in value:
        return True, None
    count = value[key]
    return _exact_nonnegative_count(count), count if _exact_nonnegative_count(count) else None


def _work_count(value: dict) -> tuple[bool, int | None]:
    counts = []
    for key in ("selected", "attempted", "queued"):
        valid, count = _optional_count(value, key)
        if not valid:
            return False, None
        if count is not None:
            counts.append(count)
    return True, max(counts) if counts else None


def _failure_within_work_bounds(value: dict, failed: int) -> bool:
    return all(failed <= value[key] for key in ("selected", "attempted", "queued") if key in value)


def _audit_report_shape_valid(value: object, *, required: bool) -> bool:
    if not isinstance(value, dict):
        return not required
    if "failed" not in value:
        return not required
    return _exact_nonnegative_count(value["failed"])


def _known_status(value: dict, allowed: frozenset[str]) -> bool:
    return "status" not in value or (
        isinstance(value["status"], str) and value["status"].lower() in allowed
    )


def _operational_receipt_valid(value: object) -> bool:
    """Reject any recorded worker failure, regardless of parse failures.

    A time deferral (``stopped_no_time``) is work that was never started, so earlier evidence is
    untouched; any exact non-negative count passes (SIN-0904). A malformed count still fails.
    """
    if not isinstance(value, dict):
        return False
    if "stopped_no_time" in value and not _exact_nonnegative_count(value["stopped_no_time"]):
        return False
    if "timing" in value:
        timing = value["timing"]
        if not isinstance(timing, dict):
            return False
        for key in ("timeouts", "crashes", "recycles"):
            if key in timing and (
                    not _exact_nonnegative_count(timing[key]) or timing[key] != 0):
                return False
    return True


def _admission_receipt_valid(value: object) -> bool:
    deferred = value.get("stopped_no_time") if isinstance(value, dict) else None
    return isinstance(value, dict) and (deferred is None or _exact_nonnegative_count(deferred)) and all(
        item is None or (type(item) is int and item == 0)
        for item in (value.get(key) for key in ("timeouts", "crashes", "recycles"))
    )


def _publication_phase_failure(phase: str, value: object) -> bool:
    if not isinstance(value, dict):
        return _publication_failure(value)
    candidate = copy.deepcopy(value)
    status = str(candidate.get("status") or "").lower()
    phase_statuses = frozenset({
        "ok", "error", "dry-run", "skipped", "skipped-dry-run",
        "skipped-refresh-error", "skipped-no-time",
    })
    if phase in {"pdf_delta", "audit_delta", "pdf_stage_overlap"}:
        if not _known_status(candidate, phase_statuses):
            return True
        if not _operational_receipt_valid(candidate):
            return True
        if status in {"dry-run", "skipped", "skipped-dry-run", "skipped-refresh-error"}:
            return _publication_failure(candidate)
    if phase == "pdf_delta":
        delta = candidate.get("delta")
        if "delta" in candidate and not isinstance(delta, dict):
            return True
        work_valid, selected = _work_count(delta) if isinstance(delta, dict) else (True, None)
        if not work_valid or (isinstance(delta, dict) and not _operational_receipt_valid(delta)):
            return True
        raw_parse = candidate.get("raw_parse")
        if status == "ok" and not isinstance(raw_parse, dict):
            return True
        if isinstance(raw_parse, dict) and not _known_status(
                raw_parse, frozenset({"ok", "skipped-no-delta"})):
            return True
        if isinstance(raw_parse, dict) and not _operational_receipt_valid(raw_parse):
            return True
        if isinstance(raw_parse, dict):
            if raw_parse.get("status") == "skipped-no-delta":
                if selected not in {None, 0} or not _valid_no_delta_report(raw_parse):
                    return True
            elif not _valid_certificate_report_shape(raw_parse):
                return True
            else:
                if isinstance(delta, dict) and not _failure_within_work_bounds(delta, raw_parse["failed"]):
                    return True
                candidate["raw_parse"] = _certificate_report_without_eligible_failures(raw_parse)
    elif phase == "audit_delta":
        if status != "ok":
            return _publication_failure(candidate)
        work_valid, work = _work_count(candidate)
        if not work_valid:
            return True
        delta = candidate.get("delta")
        if "delta" in candidate and (
                not isinstance(delta, dict) or not _operational_receipt_valid(delta)):
            return True
        if not _audit_report_shape_valid(candidate, required=True):
            return True
        raw_parse = candidate.get("raw_parse")
        if not isinstance(raw_parse, dict):
            return True
        if not _known_status(raw_parse, frozenset({"ok", "skipped-no-delta"})):
            return True
        if not _operational_receipt_valid(raw_parse):
            return True
        if raw_parse.get("status") == "skipped-no-delta":
            if work not in {None, 0} or not _valid_no_delta_report(raw_parse):
                return True
        else:
            if not _audit_report_shape_valid(raw_parse, required=True):
                return True
            if raw_parse["failed"] != candidate["failed"]:
                return True
            if not _failure_within_work_bounds(candidate, raw_parse["failed"]):
                return True
    elif phase == "pdf_stage_overlap":
        admission_present = "admission" in candidate
        admission = candidate.get("admission", {})
        if not _admission_receipt_valid(admission):
            return True
        pdf_delta = candidate.get("pdf_delta")
        audit_delta = candidate.get("audit_delta")
        if not isinstance(pdf_delta, dict) or not isinstance(audit_delta, dict):
            return True
        if not _operational_receipt_valid(pdf_delta) or not _operational_receipt_valid(audit_delta):
            return True
        for receipt in (pdf_delta, audit_delta):
            delta = receipt.get("delta")
            if "delta" in receipt and (
                    not isinstance(delta, dict) or not _operational_receipt_valid(delta)):
                return True
        if (not _known_status(pdf_delta, frozenset({"ok", "skipped"}))
                or not _known_status(audit_delta, frozenset({"ok", "skipped"}))):
            return True
        cert_selected_valid, cert_selected = _work_count(pdf_delta)
        audit_selected_valid, audit_selected = _work_count(audit_delta)
        if not cert_selected_valid or not audit_selected_valid:
            return True
        raw_parse = pdf_delta.get("raw_parse")
        pool = candidate.get("pool")
        if "pool" in candidate and not isinstance(pool, dict):
            return True
        if isinstance(pool, dict) and not _operational_receipt_valid(pool):
            return True
        if isinstance(pool, dict) and "admission" in pool:
            pool_admission = pool["admission"]
            if (not _admission_receipt_valid(pool_admission)
                    or (admission_present and pool_admission != admission)):
                return True
        pool_report = pool.get("certificate_report") if isinstance(pool, dict) else None
        pool_audit = pool.get("audit_report") if isinstance(pool, dict) else None
        nested_statuses = frozenset({"ok", "skipped", "skipped-no-delta"})
        if (isinstance(raw_parse, dict) and not _known_status(raw_parse, nested_statuses)):
            return True
        if (isinstance(pool_report, dict) and not _known_status(pool_report, nested_statuses)):
            return True
        if (isinstance(pool_audit, dict) and not _known_status(pool_audit, nested_statuses)):
            return True
        if any(isinstance(receipt, dict) and not _operational_receipt_valid(receipt)
               for receipt in (raw_parse, pool_report, pool_audit)):
            return True
        if cert_selected and (not isinstance(raw_parse, dict) or not isinstance(pool_report, dict)):
            return True
        if isinstance(raw_parse, dict):
            if raw_parse.get("status") == "skipped-no-delta":
                if cert_selected not in {None, 0} or not _valid_no_delta_report(raw_parse):
                    return True
            elif not _valid_certificate_report_shape(raw_parse):
                return True
            elif not _failure_within_work_bounds(pdf_delta, raw_parse["failed"]):
                return True
        if isinstance(pool_report, dict):
            if pool_report.get("status") == "skipped-no-delta":
                if cert_selected not in {None, 0} or not _valid_no_delta_report(pool_report):
                    return True
            elif not _valid_certificate_report_shape(pool_report):
                return True
            elif not _failure_within_work_bounds(pdf_delta, pool_report["failed"]):
                return True
        if not _audit_report_shape_valid(audit_delta, required=bool(audit_selected)):
            return True
        audit_raw = audit_delta.get("raw_parse")
        if isinstance(audit_raw, dict) and not _known_status(audit_raw, nested_statuses):
            return True
        if isinstance(audit_raw, dict) and not _operational_receipt_valid(audit_raw):
            return True
        if audit_selected and not isinstance(audit_raw, dict):
            return True
        if isinstance(audit_raw, dict):
            if audit_raw.get("status") == "skipped-no-delta":
                if audit_selected not in {None, 0} or not _valid_no_delta_report(audit_raw):
                    return True
            elif not _audit_report_shape_valid(audit_raw, required=True):
                return True
            elif audit_raw["failed"] != audit_delta["failed"]:
                return True
            elif not _failure_within_work_bounds(audit_delta, audit_raw["failed"]):
                return True
        if audit_selected and not isinstance(pool, dict):
            return True
        if isinstance(pool, dict) and not _audit_report_shape_valid(pool_audit, required=True):
            return True
        if isinstance(pool_audit, dict) and pool_audit["failed"] != audit_delta["failed"]:
            return True
        from cert_pdf_publication import validate_failure_evidence
        if isinstance(raw_parse, dict) and isinstance(pool_report, dict):
            raw_eligible = validate_failure_evidence(raw_parse)
            pool_eligible = validate_failure_evidence(pool_report)
            if raw_eligible != pool_eligible:
                return True
            if raw_eligible:
                if (raw_parse.get("failed") != pool_report.get("failed")
                        or raw_parse.get("publication_failure_evidence")
                        != pool_report.get("publication_failure_evidence")):
                    return True
                candidate["pdf_delta"]["raw_parse"] = _certificate_report_without_eligible_failures(raw_parse)
                candidate["pool"]["certificate_report"] = _certificate_report_without_eligible_failures(pool_report)
    return _publication_failure(candidate)


def _publication_hold(report: dict, reason: str) -> dict:
    """Record an enabled caller hold through the publication incident ledger."""
    tail = report.get("tail_publish") or {}
    publish = tail if tail.get("status") == "ok" else report.get("core_publish") or {}
    revision = publish.get("commit_sha")
    receipt = {"status": "held", "reason": reason, "candidate_revision": revision}
    try:
        from sinetti_services.prod_auto_activation import notify

        shared = state_paths.state_root().resolve()
        candidate = ROOT.resolve()
        if (not os.environ.get("SINETTI_STATE_DIR", "").strip()
                or shared == state_paths.BUNDLED_DATA_DIR.resolve()
                or shared == candidate or shared in candidate.parents or candidate in shared.parents):
            raise ValueError("publication incident requires external shared state outside the candidate")
        receipt["notification"] = notify(
            status="failed", reason=reason, candidate_revision=revision,
            store_path=shared / "trust_review.sqlite",
        )
    except Exception as exc:
        receipt["notification"] = {"status": "error", "reason": str(exc)[:240]}
    return receipt


_FULL_PUBLICATION_PHASES = (
    "baseline", "refresh", "app_db", "trust_review", "core_publish", "tail_publish",
    "pdf_delta", "audit_delta", "pdf_stage_overlap", "geocode", "address_geocode", "site_geocode",
    "post_pdf_geocode",
    "redcert_pdf_hosting", "acquisition_retry_queue", "coordinate_workspace",
    "candidate_frozen_preservation",
)
# The core commit is published before any PDF, geocode or hosting work and carries its own review.
# The report's app_db and trust_review may already describe the tail rebuild (SIN-0904).
_CORE_PUBLICATION_PHASES = ("baseline", "refresh", "candidate_frozen_preservation")


def _full_publication_problem(report: dict, args) -> str | None:
    if report.get("status") != "ok" or any(
        _publication_phase_failure(key, report.get(key)) for key in _FULL_PUBLICATION_PHASES
    ):
        return "refresh or tail did not complete successfully"
    tail = report.get("tail_publish") or {}
    if tail.get("status") not in {"ok", "skipped"}:
        return "tail publication did not complete"
    published = tail if tail.get("status") == "ok" else report.get("core_publish") or {}
    candidate = getattr(args, "_publication_candidate", None)
    if (not candidate or published.get("status") != "ok"
            or candidate["publish"].get("commit_sha") != published.get("commit_sha")):
        return "no fixed reviewed upload from this run"
    return None


def _core_publication_problem(report: dict, args) -> str | None:
    if any(_publication_phase_failure(key, report.get(key)) for key in _CORE_PUBLICATION_PHASES):
        return "refresh did not complete successfully"
    # The core tier covers PDF-side holds only. A tail the independent publication controller
    # held is a controller refusal for this run, not unfinished PDF work (SIN-0980).
    if (report.get("tail_publish") or {}).get("status") == "held-publication-authority":
        return "publication authority held the tail"
    core = report.get("core_publish") or {}
    candidate = getattr(args, "_core_publication_candidate", None)
    if (not candidate or core.get("status") != "ok"
            or candidate["publish"].get("commit_sha") != core.get("commit_sha")):
        return "no fixed reviewed core upload from this run"
    return None


def run_prod_auto_activation_step(*, report: dict, args) -> dict:
    """Publish only this run's exact reviewed upload, after its durable refresh receipt.

    Full tier: every phase passed, so the run's final commit activates. Core tier (SIN-0904):
    when PDF-side work holds the full tier, the run's core commit (registers, reviewed before any
    PDF work) still activates. Every production-publish check runs on whichever commit is chosen.
    """
    try:
        if rehearsal.is_rehearsal():
            return {"status": "skipped-rehearsal"}
        if not refresh_publication.production_publication_enabled():
            return {"status": "skipped-disabled"}
        if report.get("dry_run") or getattr(args, "dry_run", False):
            return {"status": "skipped-dry-run"}
        if getattr(args, "skip_refresh", False):
            return _publication_hold(report, "no successful source refresh")
        full_problem = _full_publication_problem(report, args)
        if full_problem is None:
            tier, candidate, by, extra = "full", args._publication_candidate, "nightly-data-refresh", {}
        elif _core_publication_problem(report, args) is None:
            tier, candidate, by = "core", args._core_publication_candidate, "nightly-data-refresh-core"
            extra = {"full_hold_reason": full_problem}
        else:
            return _publication_hold(report, full_problem)
        budget = getattr(args, "_publication_job_budget_s", 0)
        remaining = remaining_budget_s(
            started_monotonic=args._publication_started_monotonic, budget_s=budget,
            now_monotonic=time.monotonic(),
        ) if budget > 0 else 0
        if remaining < refresh_publication.publication_budget_s():
            return {**_publication_hold(report, "insufficient bounded publication recovery time"), **extra}
        result = refresh_publication.publish_to_production(
            publish=candidate["publish"], review=candidate["review"],
            candidate_data_dir=candidate["data_dir"], state_root=state_paths.state_root(),
            space_repo_id=args.hf_repo_id, token=os.environ.get("HF_TOKEN", ""),
            base_url=os.environ.get("SINETTI_PROD_AUTO_ACTIVATE_BASE_URL") or _space_base_url(args.hf_repo_id or ""),
            by=by, remaining_s=remaining,
        )
        return {**result, "tier": tier, **extra} if isinstance(result, dict) else result
    except Exception as exc:
        return {"status": f"error: {type(exc).__name__}: {str(exc)[:200]}"}


def run_customer_package_step(*, report: dict, args) -> dict:
    """Sign and deliver the customer data package from this run's published full release (SIN-0918).

    Off unless ``SINETTI_CUSTOMER_DATA_PACKAGE=1``: the customer-data key and the delivery location
    are owner steps. Runs only on the full tier's exact published upload; a refusal or error is a
    nightly failure. Remote delivery is refused inside the lane (SIN-0935 rollout).
    """
    from sinetti_operator import customer_data_package
    try:
        if str(os.environ.get(customer_data_package.ENABLE_ENV, "")).strip() != "1":
            return {"status": "skipped-disabled"}
        if _full_publication_problem(report, args) is not None:
            return {"status": "skipped-no-full-release"}
        candidate = args._publication_candidate
        return customer_data_package.nightly_step(
            data_dir=candidate["data_dir"], source_release=str(candidate["publish"].get("commit_sha") or ""),
            work_root=args.job_workspace or None,
        )
    except Exception as exc:
        return {"status": f"error: {type(exc).__name__}: {str(exc)[:200]}"}


def write_report(path: str | Path, report: dict) -> None:
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(json.dumps(report, indent=2, sort_keys=True), encoding="utf-8")


def build_arg_parser() -> argparse.ArgumentParser:
    ap = argparse.ArgumentParser(description="Daily Sinetti refresh with capped geocoding.")
    ap.add_argument("--dry-run", action="store_true", help="write a plan/report without network calls")
    ap.add_argument("--skip-refresh", action="store_true", help="skip CS database refresh")
    ap.add_argument("--require-retained-source-review", action="store_true",
                    help="review retained fixed candidate (every certification scheme) when skipping refresh")
    ap.add_argument(
        "--use-bundled-business-overrides",
        action="store_true",
        help="pin retained-source review to business overrides bundled with this code snapshot",
    )
    ap.add_argument("--skip-geocode", action="store_true", help="skip Google geocoding")
    ap.add_argument(
        "--skip-site-geocode",
        action="store_true",
        help="skip physical-site geocode drip after holder geocoding",
    )
    ap.add_argument("--allow-partial", action="store_true",
                    help="continue to geocode/upload even if one refresh source fails")
    ap.add_argument(
        "--job-workspace",
        action="store_true",
        help="confirm this checkout is an isolated candidate-building job workspace",
    )
    ap.add_argument("--rehearsal-candidate-dir", type=Path)
    ap.add_argument("--rehearsal-candidate-manifest", type=Path)
    ap.add_argument("--rehearsal-candidate-expected-revision")
    ap.add_argument("--rehearsal-candidate-manifest-sha256")
    ap.add_argument("--refresh-schemes", default=os.environ.get("REFRESH_SCHEMES"),
                    help="comma-separated schemes to refresh (default: every certification scheme in schemes.py)")
    ap.add_argument("--geocode-schemes", default=os.environ.get("GEOCODE_SCHEMES"),
                    help="comma-separated schemes to geocode (default: every certification scheme in schemes.py)")
    ap.add_argument("--daily-geocode-limit", type=int,
                    default=_env_int("GEOCODE_DAILY_LIMIT", DEFAULT_DAILY_GEOCODE_LIMIT))
    ap.add_argument("--monthly-geocode-limit", type=int,
                    default=_env_int("GEOCODE_MONTHLY_LIMIT", DEFAULT_MONTHLY_GEOCODE_LIMIT))
    ap.add_argument("--site-geocode-daily-limit", type=int,
                    default=_env_int("SITE_GEOCODE_DAILY_LIMIT", DEFAULT_SITE_GEOCODE_DAILY_LIMIT),
                    help="Google requests per night for site-accurate certified sites, on top of "
                         "the holder budget; 0 disables (default)")
    ap.add_argument("--pdf-delta-max-pdfs", type=int,
                    default=_env_int("PDF_DELTA_MAX_PDFS", DEFAULT_PDF_DELTA_MAX_PDFS),
                    help="max certificate PDFs to parse in this refresh; 0 disables PDF parsing")
    ap.add_argument("--pdf-delta-prefer-new-documents", action="store_true",
                    default=_env_bool("PDF_DELTA_PREFER_NEW_DOCUMENTS", False),
                    help="when parsing PDFs, select new documents before retrying previous failures")
    ap.add_argument("--pdf-delta-workers", type=int,
                    default=max(1, _env_int("PDF_DELTA_WORKERS", 1)),
                    help="worker processes for certificate PDF parsing; 1 = one at a time")
    ap.add_argument("--pdf-cache-dir", type=Path,
                    help="optional PDF cache path for the delta parser")
    ap.add_argument("--skip-pdf-delta", action="store_true", help="skip certificate PDF delta parsing")
    ap.add_argument("--pdf-stage-overlap", action="store_true", default=_env_bool("PDF_STAGE_OVERLAP", False))
    for kind, default in (("total", 8), ("certificate", 4), ("audit", 4), ("iscc", 4), ("redcert", 1)):
        ap.add_argument(f"--pdf-stage-{kind}-workers", type=int,
                        default=os.environ.get(f"PDF_STAGE_{kind.upper()}_WORKERS", str(default)))
    ap.add_argument("--skip-acquisition-queue", action="store_true",
                    default=_env_bool("SKIP_ACQUISITION_RETRY_QUEUE", False),
                    help="ignore an admin-queued acquisition-retry batch this run")
    ap.add_argument("--audit-delta-max-pdfs", type=int,
                    default=_env_int("AUDIT_DELTA_MAX_PDFS", DEFAULT_AUDIT_DELTA_MAX_PDFS),
                    help="max ISCC audit reports to parse; 0 disables")
    ap.add_argument("--audit-delta-prefer-new-documents", action="store_true",
                    default=_env_bool("AUDIT_DELTA_PREFER_NEW_DOCUMENTS", False),
                    help="when parsing audit reports, select new documents before retrying failures")
    ap.add_argument("--audit-delta-workers", type=int,
                    default=max(1, _env_int("AUDIT_DELTA_WORKERS", 1)),
                    help="worker processes for ISCC audit report parsing; 1 = one at a time")
    ap.add_argument("--skip-audit-delta", action="store_true", help="skip ISCC audit report delta parsing")
    ap.add_argument("--redcert-pdf-max-fetches", type=int,
                    default=_env_int("REDCERT_PDF_MAX_FETCHES", DEFAULT_REDCERT_PDF_MAX_FETCHES),
                    help="max REDcert certificate PDFs to fetch+host in this refresh; 0 disables")
    ap.add_argument("--skip-redcert-pdf-hosting", action="store_true",
                    help="skip REDcert certificate PDF fetch+host lazy-cache fill")
    ap.add_argument("--skip-app-db-rebuild", action="store_true",
                    help="skip rebuilding sinetti_app.sqlite.gz before upload")
    ap.add_argument("--ledger-path", default=os.environ.get("GEOCODE_LEDGER_PATH", str(DEFAULT_LEDGER_PATH)))
    ap.add_argument("--report-path", default=os.environ.get("REFRESH_REPORT_PATH", str(DEFAULT_REPORT_PATH)))
    ap.add_argument("--history-dir", default=os.environ.get("REFRESH_HISTORY_DIR", str(DEFAULT_HISTORY_DIR)))
    ap.add_argument("--hf-repo-id", default=os.environ.get("HF_SPACE_REPO_ID", ""))
    ap.add_argument("--hf-folder", default=os.environ.get("HF_UPLOAD_FOLDER", str(ROOT)))
    ap.add_argument("--data-repo-id", default=os.environ.get("SINETTI_DATA_REPO_ID", ""),
                    help="private dataset repo for refresh artifacts; empty = legacy Space upload")
    return ap


def main(argv: list[str] | None = None) -> int:
    from sinetti_operator.registration import register_live_providers
    register_live_providers()  # jobs never reach full_registration (SIN-0999)
    parser = build_arg_parser()
    args = parser.parse_args(argv)
    if args.pdf_stage_overlap:
        try:
            _pdf_stage_limits(args).validate()
            if float(os.environ.get("AUDIT_PARSE_IDLE_TIMEOUT_S", "180")) <= 0:
                raise ValueError("audit item timeout must be positive")
        except ValueError as exc:
            parser.error(str(exc))
    candidate_values = (args.rehearsal_candidate_dir, args.rehearsal_candidate_manifest,
                        args.rehearsal_candidate_expected_revision, args.rehearsal_candidate_manifest_sha256)
    if any(candidate_values):
        if not all(candidate_values):
            parser.error("candidate rehearsal requires directory, manifest, and expected revision")
        if not rehearsal.is_rehearsal() or not args.job_workspace:
            parser.error("candidate rehearsal is allowed only in an isolated rehearsal job workspace")
        if tuple(_csv(args.refresh_schemes, DEFAULT_REFRESH_SCHEMES)) != rehearsal_candidate_baseline.EIGHT_SCHEMES:
            parser.error("candidate rehearsal requires the exact eight refreshed schemes")
        if args.dry_run or args.skip_refresh or args.skip_app_db_rebuild:
            parser.error("candidate rehearsal requires register refresh and serving rebuild")
    if args.require_retained_source_review and not args.skip_refresh:
        parser.error("--require-retained-source-review requires --skip-refresh")
    if args.use_bundled_business_overrides and not (
        args.skip_refresh and args.require_retained_source_review
    ):
        parser.error(
            "--use-bundled-business-overrides requires --skip-refresh "
            "--require-retained-source-review"
        )
    business_overrides = None
    if args.use_bundled_business_overrides:
        try:
            business_overrides = _validate_bundled_business_overrides(ROOT)
        except ValueError as exc:
            parser.error(str(exc))
    args.business_overrides = business_overrides
    args._publication_candidate = None
    args._core_publication_candidate = None
    memtrace_stop = start_memory_trace()
    try:
        if business_overrides is None:
            return _main_run(args)
        with _bind_bundled_business_overrides(business_overrides):
            return _main_run(args)
    finally:
        candidate = getattr(args, "_publication_candidate", None)
        core = getattr(args, "_core_publication_candidate", None)
        if candidate:
            candidate["snapshot"].cleanup()
        if core and core is not candidate:
            core["snapshot"].cleanup()
        memtrace_stop.set()


def _require_candidate_eight_accepted(refresh: dict) -> None:
    expected = rehearsal_candidate_baseline.EIGHT_SCHEMES
    if set(refresh) != set(expected):
        raise ValueError("candidate rehearsal must report exactly eight schemes")
    failed = [scheme for scheme in expected if not isinstance(refresh.get(scheme), dict)
              or refresh[scheme].get("status") != "ok" or refresh[scheme].get("decision") != "accepted"]
    if failed:
        raise ValueError("candidate rehearsal requires accepted refreshes: " + ", ".join(failed))


def _rehearsal_refresh_schemes(schemes: list[str], *, candidate_mode: bool) -> list[str]:
    """Keep ordinary routing intact; only the sealed rehearsal delegates local legs."""
    return ([scheme for scheme in schemes if scheme not in rehearsal_candidate_registers.LOCAL_SCHEMES]
            if candidate_mode else list(schemes))


def _pdf_coverage_preflight_required(args) -> bool:
    return (not args.dry_run and not rehearsal.is_rehearsal()
        and os.environ.get('SINETTI_PDF_COVERAGE_DRY_RUN') != '1')


def run_pdf_coverage_preflight_step(args) -> dict | None:
    """Independent authenticated heartbeat check before any refresh; opt-in only."""
    if os.environ.get('SINETTI_PDF_COVERAGE_MONITOR_ENABLED') != '1':
        return None
    from scripts.monitor_trust_jobs import _coverage_guard, _coverage_runtime_config, _coverage_notifier, _coverage_unknown
    if guarded := _coverage_guard(dry_run=args.dry_run):
        return guarded
    try:
        import uuid
        from sinetti_services.pdf_coverage_runtime import RuntimePorts, run_hosted_preflight
        config = _coverage_runtime_config()
        return run_hosted_preflight(config, run_id='hosted-' + uuid.uuid4().hex,
            ports=RuntimePorts(notify=_coverage_notifier(config.root, 'hosted')))
    except Exception as exc:
        return {**_coverage_unknown('hosted_entrypoint_unavailable'), 'error_type': type(exc).__name__}


def _main_run(args) -> int:
    # Split out of main() only so start_memory_trace() can wrap it in a try/finally --
    # main()'s own source is asserted on verbatim by test_daily_refresh_auto_activation_hook.py
    # and test_daily_refresh_email_hook.py, so those wiring checks must look at THIS function.
    candidate_values = (args.rehearsal_candidate_dir, args.rehearsal_candidate_manifest,
                        args.rehearsal_candidate_expected_revision, args.rehearsal_candidate_manifest_sha256)
    started = datetime.now(timezone.utc)
    started_mono = time.monotonic()
    budget_s = _env_job_budget_s()
    args._publication_started_monotonic = started_mono
    args._publication_job_budget_s = budget_s
    tail_reserve_s = _publication_tail_reserve_s()

    def can_start_next_tail_document() -> bool:
        remaining = remaining_budget_s(
            started_monotonic=started_mono,
            budget_s=budget_s,
            now_monotonic=time.monotonic(),
        )
        return can_start_tail(remaining_s=remaining, reserve_s=tail_reserve_s)

    report: dict = {
        "started_at": _now_iso(started),
        "dry_run": args.dry_run,
        "config": {
            "space_revision": os.environ.get("SINETTI_SPACE_REVISION") or "",
            # SIN-1007: the Space commit the app ran when this pinned job started.
            "app_revision": os.environ.get("SINETTI_APP_REVISION") or "",
            # SIN-1007 step 3: set when the job ran a jobs code release instead of the Space.
            "jobs_code_revision": os.environ.get("SINETTI_JOBS_CODE_REVISION") or "",
            "jobs_code_release": os.environ.get("SINETTI_JOBS_CODE_RELEASE") or "",
            "jobs_code_source": os.environ.get("SINETTI_JOBS_CODE_SOURCE") or "",
            "refresh_schemes": _csv(args.refresh_schemes, DEFAULT_REFRESH_SCHEMES),
            "frozen_rehearsal_schemes": list(rehearsal_candidate_baseline.FROZEN_SCHEMES) if any(candidate_values) else [],
            "geocode_schemes": _csv(args.geocode_schemes, DEFAULT_GEOCODE_SCHEMES),
            "daily_geocode_limit": args.daily_geocode_limit,
            "monthly_geocode_limit": args.monthly_geocode_limit,
            "site_geocode_daily_limit": args.site_geocode_daily_limit,
            "pdf_delta_max_pdfs": args.pdf_delta_max_pdfs,
            "pdf_delta_prefer_new_documents": args.pdf_delta_prefer_new_documents,
            "require_retained_source_review": args.require_retained_source_review,
            "pdf_cache_dir": str(args.pdf_cache_dir) if args.pdf_cache_dir else None,
            "audit_delta_max_pdfs": args.audit_delta_max_pdfs,
            "audit_delta_prefer_new_documents": args.audit_delta_prefer_new_documents,
            "pdf_stage_overlap": args.pdf_stage_overlap,
            "pdf_stage_requested_limits": dict(_pdf_stage_limits(args).__dict__) if args.pdf_stage_overlap else None,
            "pdf_stage_effective_limits": dict(_pdf_stage_limits(args).__dict__) if args.pdf_stage_overlap else None,
            "pdf_delta_workers": args.pdf_delta_workers,
            "audit_delta_workers": args.audit_delta_workers,
            "redcert_pdf_max_fetches": args.redcert_pdf_max_fetches,
            "ledger_path": str(args.ledger_path),
            "history_dir": str(args.history_dir),
        },
    }
    if args.business_overrides is not None:
        report["config"]["business_overrides"] = args.business_overrides
    preflight = run_pdf_coverage_preflight_step(args)
    if preflight is not None:
        report['pdf_coverage_preflight'] = preflight
    upload_failed = False
    app_db_failed = False
    hf_token = os.environ.get("HF_TOKEN", "")

    def _on_interrupt() -> None:
        if report.get("status") not in {"ok", "error", "partial"}:
            report["status"] = "interrupted"
        _flush_receipt(report, args, phase=str(report.get("phase") or "interrupted"))

    install_interrupt_flush(_on_interrupt)
    divergent_legacy_root = bool(
        not args.data_repo_id
        and Path(args.hf_folder).resolve() != ROOT.resolve()
    )

    coordinate_baselines = None if args.dry_run or not args.job_workspace else geocode_workspace.snapshot_baselines(
        ROOT / 'data', geocode_points.HOLDER_SCHEMES,
    )

    if args.dry_run:
        report["baseline"] = {"status": "skipped-dry-run"}
    elif divergent_legacy_root:
        report["baseline"] = {
            "status": "error-divergent-upload-root",
            "error": "legacy upload root must match the rebuilt and reviewed candidate root",
        }
    elif args.job_workspace and not args.data_repo_id:
        report["baseline"] = {
            "status": "error-missing-data-repo",
            "error": "isolated refresh jobs require SINETTI_DATA_REPO_ID",
        }
    elif args.data_repo_id and not args.job_workspace:
        report["baseline"] = {
            "status": "error-nonisolated-workspace",
            "error": "dataset refresh refused outside an explicitly isolated job workspace",
        }
    elif args.data_repo_id:
        report["baseline"] = data_plane.boot_sync(
            repo_id=args.data_repo_id,
            token=hf_token,
            data_dir=ROOT / "data",
            materialize=True,
        )
    else:
        report["baseline"] = {"status": "disabled"}
    baseline_failed = bool(
        (args.data_repo_id or args.job_workspace or divergent_legacy_root)
        and not args.dry_run
        and (
            report["baseline"].get("status") != "ok"
            or not report["baseline"].get("hub_revision")
        )
    )

    if any(candidate_values) and not baseline_failed:
        try:
            report["candidate_frozen_pre_overlay"] = rehearsal_candidate_registers.frozen_artifact_proof(ROOT / "data")
            report["candidate_baseline"] = rehearsal_candidate_baseline.seed_candidate_baseline(
                candidate_dir=args.rehearsal_candidate_dir,
                manifest_path=args.rehearsal_candidate_manifest,
                destination_data=ROOT / "data",
                expected_revision=args.rehearsal_candidate_expected_revision,
                expected_manifest_sha256=args.rehearsal_candidate_manifest_sha256,
                synced_revision=report["baseline"]["hub_revision"],
                rehearsal_enabled=rehearsal.is_rehearsal(), job_workspace=args.job_workspace,
            )
            report["candidate_frozen_start"] = rehearsal_candidate_registers.frozen_artifact_proof(ROOT / "data")
            rehearsal_candidate_registers.require_frozen_overlay_preservation(
                report["candidate_frozen_pre_overlay"], report["candidate_frozen_start"],
            )
            report["candidate_recovery_start"] = rehearsal_recovery_contract.verify_contract(
                report["candidate_baseline"]["recovery_contract"], data_dir=ROOT / "data",
            )
        except (ValueError, OSError, sqlite3.Error) as exc:
            report["candidate_baseline"] = {"status": "error", "error": str(exc)}
            baseline_failed = True

    if args.dry_run or baseline_failed or not args.job_workspace:
        report['coordinate_workspace'] = {
            'status': 'skipped-dry-run' if args.dry_run else
            'skipped-baseline-error' if baseline_failed else 'skipped-standalone-checkout',
        }
    else:
        report['coordinate_workspace'] = geocode_workspace.prepare_workspace(ROOT / 'data', coordinate_baselines)
        if report['coordinate_workspace'].get('status') != 'ok':
            baseline_failed = True

    if baseline_failed:
        report["refresh"] = {"status": "skipped-baseline-error"}
    elif args.dry_run or args.skip_refresh:
        report["refresh"] = {"status": "skipped-dry-run" if args.dry_run else "skipped"}
    else:
        refresh_schemes = _rehearsal_refresh_schemes(
            report["config"]["refresh_schemes"], candidate_mode=bool(any(candidate_values)),
        )
        report["refresh"] = refresh_databases(refresh_schemes, run_id=report["started_at"])
        if any(candidate_values):
            local_refresh = rehearsal_candidate_registers.run_rehearsal_local_registers(
                ROOT / "data", report["started_at"],
            )
            report["candidate_local_registers"] = local_refresh
            report["refresh"].update(local_refresh["refresh"])
            try:
                _require_candidate_eight_accepted(report["refresh"])
            except ValueError as exc:
                report["candidate_eight_scheme_gate"] = {"status": "error", "error": str(exc)}
                baseline_failed = True
            else:
                report["candidate_eight_scheme_gate"] = {"status": "ok"}
            if local_refresh["status"] != "ok":
                baseline_failed = True

    report["status"] = "running"
    report["phase"] = "registers"
    _flush_receipt(report, args, phase="registers")

    core_ok = (not args.dry_run) and core_can_publish(
        report.get("refresh") or {},
        allow_partial=args.allow_partial,
        baseline_failed=baseline_failed,
    )

    # Operator CLI only (`scripts/build_holder_location_fallback.py`). The nightly must
    # not load the ~1 GB postal index to rewrite a git-tracked file the data plane
    # never publishes.
    report["holder_location_fallback"] = {
        "status": "skipped-operator",
        "reason": "rebuild is an explicit CLI step; not part of daily_refresh",
    }

    if args.dry_run:
        report["app_db"] = {"status": "skipped-dry-run"}
        report["trust_review"] = {"status": "skipped", "reason": "no fixed accepted candidate"}
        report["upload"] = {"status": "skipped-dry-run"}
        report["core_publish"] = {"status": "skipped-dry-run"}
    elif args.skip_app_db_rebuild:
        report["app_db"] = {"status": "skipped"}
        report["trust_review"] = {"status": "skipped", "reason": "no fixed accepted candidate"}
        report["upload"] = {"status": "skipped"}
        report["core_publish"] = {"status": "skipped"}
    elif not core_ok:
        skip_status = "skipped-baseline-error" if baseline_failed else "skipped-refresh-error"
        report["app_db"] = {"status": skip_status}
        report["trust_review"] = {"status": "skipped", "reason": "no fixed accepted candidate"}
        report["upload"] = {"status": skip_status}
        report["core_publish"] = {"status": skip_status}
    else:
        app_db_failed = _rebuild_app_and_review(report)
        if app_db_failed:
            report["upload"] = {"status": "skipped-app-db-error"}
            report["core_publish"] = {"status": "skipped-app-db-error"}
        else:
            report["upload"] = {"status": "pending-upload"}
            upload, upload_failed = _publish_candidate(report, args, hf_token=hf_token)
            report["upload"] = upload
            report["core_publish"] = dict(upload)
            report["phase"] = "core"
            _flush_receipt(report, args, phase="core")

    remaining = remaining_budget_s(
        started_monotonic=started_mono, budget_s=budget_s, now_monotonic=time.monotonic(),
    )
    run_tail = (
        (report.get("core_publish") or {}).get("status") == "ok"
        and can_start_tail(remaining_s=remaining, reserve_s=tail_reserve_s)
    )
    can_continue = run_tail
    if args.dry_run or args.skip_geocode or not can_continue:
        if args.dry_run:
            status = "skipped-dry-run"
        elif args.skip_geocode:
            status = "skipped"
        else:
            status = "skipped-refresh-error" if not core_ok else "skipped-no-time"
        report["geocode"] = {"status": status, "attempted": 0, "added": 0}
        report["address_geocode"] = {"status": status, "attempted": 0, "added": 0}
    else:
        report["geocode"] = run_geocode_step(lambda: geocode_with_budget(
            schemes=report["config"]["geocode_schemes"],
            ledger_path=args.ledger_path,
            daily_limit=args.daily_geocode_limit,
            monthly_limit=args.monthly_geocode_limit,
            key=google_geocoder._key(),
            coordinate_baselines=coordinate_baselines,
        ))
        report["address_geocode"] = run_geocode_step(lambda: geocode_addresses_with_budget(
            schemes=report["config"]["geocode_schemes"],
            ledger_path=args.ledger_path,
            daily_limit=args.daily_geocode_limit,
            monthly_limit=args.monthly_geocode_limit,
            key=google_geocoder._key(),
            coordinate_baselines=coordinate_baselines,
        ))
        _flush_receipt(report, args, phase="geocode")

    if args.dry_run or args.skip_geocode or args.skip_site_geocode or not can_continue:
        if args.dry_run:
            site_status = "skipped-dry-run"
        elif args.skip_geocode or args.skip_site_geocode:
            site_status = "skipped"
        else:
            site_status = "skipped-refresh-error" if not core_ok else "skipped-no-time"
        report["site_geocode"] = {"status": site_status, "attempted": 0, "geocoded": 0}
    else:
        report["site_geocode"] = run_geocode_step(lambda: geocode_sites_with_budget(
            ledger_path=args.ledger_path,
            site_daily_limit=args.site_geocode_daily_limit,
            key=google_geocoder._key(),
            coordinate_baselines=coordinate_baselines,
            site_artifact=_candidate_app_artifact(),
        ), count_key="geocoded")

    before_pdf_locations = {}
    if can_continue and not args.dry_run and not args.skip_geocode:
        before_pdf_locations = run_geocode_step(lambda: _geocode_candidate_inputs(_candidate_app_artifact()))

    source_audit = run_pdf_source_audit_step(data_root=ROOT / 'data', can_continue=can_continue,
        dry_run=args.dry_run, should_continue=can_start_next_tail_document,
        remaining_s=max(0.0, remaining_budget_s(started_monotonic=started_mono,
            budget_s=budget_s, now_monotonic=time.monotonic()) - tail_reserve_s))
    report['source_version_audit'] = source_audit['report']
    audit_kwargs = {'source_audit': source_audit} if source_audit.get('durable') else {}
    report["acquisition_retry_queue"] = run_acquisition_queue_step(
        can_continue=can_continue, dry_run=args.dry_run, skip=args.skip_acquisition_queue,
    )

    if args.pdf_stage_overlap:
        overlap = run_pdf_stage_overlap_step(args, can_continue=can_continue, dry_run=args.dry_run,
            should_continue=can_start_next_tail_document, acquisition=report["acquisition_retry_queue"],
            skip_reason="skipped-no-time" if core_ok else "skipped-refresh-error", **audit_kwargs)
        report["pdf_stage_overlap"] = overlap
        report["pdf_delta"] = overlap.get("pdf_delta") or {"status": overlap["status"]}
        report["audit_delta"] = overlap.get("audit_delta") or {"status": overlap["status"]}
        if _pdf_stage_overlap_failed(report):
            report["acquisition_retry_queue"] = {**report["acquisition_retry_queue"], "downstream": "blocked-overlap-error"}
        else:
            report["acquisition_retry_queue"] = finish_acquisition_queue_step(
                report["acquisition_retry_queue"], pdf_delta=report["pdf_delta"])
    else:
        if args.dry_run:
            report["pdf_delta"] = {"status": "skipped-dry-run", "max_pdfs": args.pdf_delta_max_pdfs}
        elif args.skip_pdf_delta or args.pdf_delta_max_pdfs <= 0:
            report["pdf_delta"] = {"status": "skipped", "max_pdfs": args.pdf_delta_max_pdfs}
        elif not can_continue:
            report["pdf_delta"] = {
                "status": "skipped-refresh-error" if not core_ok else "skipped-no-time",
                "max_pdfs": args.pdf_delta_max_pdfs,
            }
        else:
            try:
                pdf_kwargs = {
                    "max_pdfs": args.pdf_delta_max_pdfs,
                    "prefer_new_documents": args.pdf_delta_prefer_new_documents,
                    "should_continue": can_start_next_tail_document,
                }
                pdf_kwargs.update(audit_kwargs)
                if args.pdf_cache_dir is not None:
                    pdf_kwargs["cache_dir"] = args.pdf_cache_dir
                if args.pdf_delta_workers > 1:
                    pdf_kwargs["workers"] = args.pdf_delta_workers
                report["pdf_delta"] = run_pdf_delta_refresh(**pdf_kwargs)
            except Exception as exc:
                report["pdf_delta"] = {
                    "status": "error",
                    "max_pdfs": args.pdf_delta_max_pdfs,
                    "exception": {"class": type(exc).__name__, "message": str(exc)},
                }

        report["acquisition_retry_queue"] = finish_acquisition_queue_step(
            report["acquisition_retry_queue"], pdf_delta=report["pdf_delta"],
        )

        report["audit_delta"] = run_audit_delta_step(
            args,
            can_continue=can_continue,
            dry_run=args.dry_run,
            should_continue=can_start_next_tail_document,
            **audit_kwargs,
        )
    report['post_pdf_geocode'] = run_post_pdf_geocode_step(
        args, before=before_pdf_locations,
        can_continue=can_continue and not _pdf_stage_overlap_failed(report),
        should_continue=can_start_next_tail_document, coordinate_baselines=coordinate_baselines,
        skip_reason=('skipped-refresh-error' if not core_ok or _pdf_stage_overlap_failed(report)
                     else 'skipped-no-time'))
    _flush_receipt(report, args, phase='post-pdf-geocode')
    report["queue_depth"] = _pdf_delta_queue_depth(report)

    if not baseline_failed and not args.dry_run and not args.skip_refresh:
        report["pdf_backlog_snapshots"] = _record_accepted_backlog_snapshots(
            report["refresh"],
            refresh_dir=args.history_dir,
            run_id=report["started_at"],
            as_of=report["started_at"],
        )
        report["audit_backlog_snapshot"] = _record_accepted_audit_backlog_snapshot(
            report["refresh"],
            refresh_dir=args.history_dir,
            run_id=report["started_at"],
            as_of=report["started_at"],
        )
        report["map_coverage"] = _record_map_coverage_snapshot(report, refresh_dir=args.history_dir)

    if args.dry_run:
        report["redcert_pdf_hosting"] = {"status": "skipped-dry-run",
                                          "max_fetches": args.redcert_pdf_max_fetches}
    elif args.skip_redcert_pdf_hosting or args.redcert_pdf_max_fetches <= 0:
        report["redcert_pdf_hosting"] = {"status": "skipped", "max_fetches": args.redcert_pdf_max_fetches}
    elif not can_continue:
        report["redcert_pdf_hosting"] = {"status": "skipped-refresh-error",
                                          "max_fetches": args.redcert_pdf_max_fetches}
    else:
        report["redcert_pdf_hosting"] = _refresh_redcert_pdfs(
            run_id=report["started_at"], max_fetches=args.redcert_pdf_max_fetches,
            should_continue=lambda: remaining_budget_s(
                started_monotonic=started_mono, budget_s=budget_s, now_monotonic=time.monotonic(),
            ) > _redcert_hosting_reserve_s(),
        )
        _flush_receipt(report, args, phase="tail")

    # Per-document tail callbacks stop when the publish reserve begins. The second
    # publish consumes that reserve, so it must not apply the start-work predicate again.
    if (
        core_ok and not app_db_failed and not upload_failed
        and _tail_changed(report)
        and not _pdf_stage_overlap_failed(report)
        and not args.dry_run
        and not args.skip_app_db_rebuild
    ):
        overlap = report.get("pdf_stage_overlap") or {}
        validated_app = (overlap.get("app_db") if overlap.get("promoted")
            and _pdf_stage_inputs_unchanged(overlap.get("app_input_proof", {})) else None)
        tail_rebuild_failed = (_rebuild_app_and_review(report, validated_app=validated_app)
            if validated_app is not None else _rebuild_app_and_review(report))
        if tail_rebuild_failed:
            report["tail_publish"] = {"status": "skipped-app-db-error"}
        else:
            tail_upload, tail_upload_failed = _publish_candidate(report, args, hf_token=hf_token)
            report["tail_publish"] = tail_upload
            if not tail_upload_failed:
                report["upload"] = tail_upload
            _flush_receipt(report, args, phase="tail-publish")
    elif "tail_publish" not in report:
        report["tail_publish"] = {"status": "skipped"}

    if _pdf_stage_overlap_failed(report):
        report["tail_publish"] = {"status": "skipped-pdf-stage-overlap-error"}

    if any(candidate_values) and isinstance(report.get("candidate_baseline"), dict) and report["candidate_baseline"].get("status") == "seeded":
        try:
            report["candidate_recovery_end"] = rehearsal_recovery_contract.verify_contract(
                report["candidate_baseline"]["recovery_contract"], data_dir=ROOT / "data",
            )
            frozen_end = rehearsal_candidate_registers.frozen_artifact_proof(ROOT / "data")
            report["candidate_frozen_end"] = frozen_end
            if frozen_end != report.get("candidate_frozen_start"):
                report["candidate_frozen_preservation"] = {"status": "error", "reason": "frozen raw or serving slices changed"}
                baseline_failed = True
            else:
                report["candidate_frozen_preservation"] = {"status": "ok"}
        except (OSError, ValueError, sqlite3.Error) as exc:
            report["candidate_frozen_preservation"] = {"status": "error", "error": str(exc)}
            baseline_failed = True
    elif any(candidate_values):
        report["candidate_frozen_preservation"] = {"status": "error", "error": "candidate baseline was not seeded"}
        baseline_failed = True

    finished = datetime.now(timezone.utc)
    candidate_final_failed = (
        (report.get("candidate_frozen_preservation") or {}).get("status") == "error"
        or ("candidate_recovery_end" not in report and bool(any(candidate_values)))
    )
    core_published = (report.get("core_publish") or {}).get("status") == "ok"
    refresh_failed = has_errors(report.get("refresh", {})) and not args.allow_partial
    job_failed = candidate_final_failed or ((not core_published) and (
        baseline_failed or upload_failed or app_db_failed or refresh_failed or _pdf_stage_overlap_failed(report)
    ))
    from jobs_publication_policy import independent_runtime
    if independent_runtime() and any(_publication_failure(report.get(name))
                                     for name in ("core_publish", "tail_publish", "upload")):
        job_failed = True
    if args.require_retained_source_review and (
        (report.get("tail_publish") or {}).get("status") not in {"ok", "skipped"}
    ):
        # A completed core publish does not make unreviewed PDF work durable.
        job_failed = True
    if job_failed:
        report["status"] = "error"
    elif _pdf_stage_overlap_failed(report) or ((report.get("pdf_delta") or {}).get("status") == "error" and core_published):
        report["status"] = "partial"
    elif (has_errors(report.get("refresh", {})) or (report.get("pdf_delta") or {}).get("status") == "error"
          or _geocode_step_failed(report)):
        report["status"] = "partial"
    else:
        report["status"] = "ok"
    monitor_failed = bool(preflight is not None and preflight['verdict'] != 'pass'
        and _pdf_coverage_preflight_required(args))
    if monitor_failed and report['status'] == 'ok':
        report['status'] = 'partial'
    report["finished_at"] = _now_iso(finished)
    report["duration_s"] = round((finished - started).total_seconds(), 2)
    write_report(args.report_path, report)
    refresh_quality.append_history_files(args.history_dir, report)

    # Production auto-activation tail (spec: docs/superpowers/specs/2026-08-11-production-
    # auto-activation-design.md). Runs after the refresh legs' receipt is already flushed above,
    # so a job timeout mid-tail never erases refresh evidence. Off by default
    # (SINETTI_PROD_AUTO_ACTIVATE unset); shared publication gates the whole step.
    # Belt-and-braces containment mirrors the daily-email call site below: the step already
    # wraps itself, but the call site must survive even if that containment is ever bypassed.
    try:
        report["auto_activation"] = run_prod_auto_activation_step(report=report, args=args)
    except Exception as exc:
        report["auto_activation"] = {"status": f"error: {type(exc).__name__}: {str(exc)[:200]}"}
    write_report(args.report_path, report)

    # Customer data package (SIN-0918), from the full tier's exact published upload. Off by default.
    try:
        report["customer_package"] = run_customer_package_step(report=report, args=args)
    except Exception as exc:
        report["customer_package"] = {"status": f"error: {type(exc).__name__}: {str(exc)[:200]}"}
    from sinetti_operator.customer_data_package import nightly_failed as customer_package_failed
    customer_failed = customer_package_failed(report["customer_package"])
    write_report(args.report_path, report)

    # Co-located regulatory producer + Task 6 delivery. This is a no-op under windows_task
    # authority and is isolated from certificate refresh like the general email tail below.
    try:
        from scripts.run_regulatory_monitor import run_daily_refresh_regulatory_step
        report["regulatory_monitor"] = run_daily_refresh_regulatory_step()
    except Exception as exc:
        report["regulatory_monitor"] = {
            "ok": False, "status": f"error: {type(exc).__name__}: {str(exc)[:200]}",
            "notification": {"status": "not-attempted"},
        }
    write_report(args.report_path, report)

    # Daily emails last, from the final report, guaranteed non-fatal (service wraps everything).
    try:
        from sinetti_services.daily_email_job import run_daily_email_step
        report["daily_email"] = run_daily_email_step(
            report, history_dir=args.history_dir,
            regulatory_result=report["regulatory_monitor"].get("notification"),
        )
    except Exception as exc:
        report["daily_email"] = {
            "status": f"error: {type(exc).__name__}: {str(exc)[:200]}",
            "admin": "not-attempted",
            "digest": "not-attempted",
            "red_flag_count": 0,
        }
    write_report(args.report_path, report)
    print(json.dumps(report, indent=2, sort_keys=True))
    return int(job_failed or monitor_failed or customer_failed)


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