Blame for scripts/run_iscc_cloud_ocr.py

#!/usr/bin/env python
"""Cloud OCR job: read scanned ISCC certificates the nightly cannot, publish one candidate release.



Owner name: the morning OCR run. Since SIN-0645 it also has a REDcert lane (redcert_cloud_ocr):

scanned REDcert certificates, read from the dataset's hosted REDcert PDF store, merged into the same

candidate. Its default cap is `redcert_cloud_ocr.DEFAULT_MAX_CERTS`; an explicit zero disables it.



Runs as its own Hugging Face Scheduled Job each morning after the nightly data refresh has

published (registration: scripts/register_hf_cloud_ocr_job.py). Runs the DEPLOYED Space code,

like every scheduled job, inside an isolated job workspace:



  1. hydrate the newest dataset release into ./data (data_plane.boot_sync, materialize=True);

  2. queue scanned ISCC certificates never OCR'd before (iscc_cloud_ocr.select_queue), capped;

  3. download each PDF and OCR it with the pinned Tesseract (the same reader the PC uses);

  4. offer only results that pass the approved quality rules to the existing merge tool;

  5. rebuild structured + app data when anything was promoted, record every attempt in the

     ledger that rides the release, and publish ONE candidate release on top of the baseline.



Production publication is explicitly opted in with SINETTI_PROD_AUTO_ACTIVATE; otherwise the

publish remains a candidate. Complete Trust Review and exact revision verification gate it.

`SINETTI_REHEARSAL=1` withholds the publish (data_plane.publish) and nothing else is outward.

`--parity-ids` re-reads certificates the PC already OCR'd and reports differences; it never

publishes. Decision brief: docs/superpowers/briefs/2026-09-29-cloud-ocr-job-brief.md.

"""
from __future__ import annotations

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

# Never .resolve(): in a Hugging Face snapshot every file is a symlink into the blob store, and a
# resolved root lands there instead of in the checkout (same rule as scripts/daily_refresh.py).
ROOT = Path(__file__).parent.parent
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

import cert_pdf_store  # noqa: E402
import data_plane  # noqa: E402
import iscc_cloud_ocr as ocr_job  # noqa: E402
import redcert_cloud_ocr as redcert_lane  # noqa: E402
import rehearsal  # noqa: E402
import state_paths  # noqa: E402
from sinetti_services.refresh_job_lifecycle import (  # noqa: E402
    can_start_tail,
    remaining_budget_s,
)

PRODUCER = "iscc_cloud_ocr"


def iscc_lane_deadline(started_monotonic: float, now_monotonic: float,

                       budget_s: int, reserve_s: int,

                       iscc_selected: int, redcert_selected: int) -> float | None:
    """Absolute ISCC cutoff that reserves REDcert's proportional share of the OCR window.



    No bounded budget, an empty lane, or REDcert disabled keeps the historical single-lane

    behavior. The global publication reserve remains the hard cutoff for both lanes.

    """
    if budget_s <= 0 or iscc_selected <= 0 or redcert_selected <= 0:
        return None
    elapsed = max(0.0, now_monotonic - started_monotonic)
    work_window = max(0.0, budget_s - elapsed - max(0, reserve_s))
    share = iscc_selected / (iscc_selected + redcert_selected)
    return now_monotonic + work_window * share


def can_start_ocr(*, now_monotonic: float, started_monotonic: float, budget_s: int,

                  reserve_s: int, lane_deadline: float | None = None) -> bool:
    """Whether another OCR document may start under the lane and global cutoffs."""
    if lane_deadline is not None and now_monotonic >= lane_deadline:
        return False
    if budget_s <= 0:
        return True
    remaining = remaining_budget_s(started_monotonic=started_monotonic, budget_s=budget_s,
                                   now_monotonic=now_monotonic)
    return can_start_tail(remaining_s=remaining, reserve_s=reserve_s)


def review_candidate(*, data_dir: Path, state_root: Path) -> dict:
    """Catalogue and review the entire fixed certificate release; fail closed on lost evidence."""
    from scripts.build_cert_format_catalog import main as build_catalog
    import sinetti_data
    from schemes import CERTIFICATION_SCHEMES, LOADED_CERTIFICATION_SCHEMES
    from sinetti_services import refresh_publication, trust_review_service

    try:
        if build_catalog([
            "--artifact", str(data_dir / "cert_pdf_details.sqlite.gz"),
            "--out", str(data_dir / "cert_format_catalog.json"),
            "--ins-index", str(data_dir / "ins" / "cert_pdfs.json"), "--skip-thumbs",
        ]) != 0:
            raise ValueError("certificate-format catalog build failed")
        identity = refresh_publication.capture_review_identity(data_dir)
        groups = sinetti_data.load_app_certificate_groups(data_dir / "sinetti_app.sqlite.gz")
        review_schemes = tuple(scheme for scheme in CERTIFICATION_SCHEMES
                               if scheme in LOADED_CERTIFICATION_SCHEMES or groups.get(scheme))
        review = trust_review_service.safe_post_refresh_review(
            candidate_artifact=data_dir / "sinetti_app.sqlite.gz",
            details_artifact=data_dir / "cert_pdf_details.sqlite.gz",
            schemes=review_schemes,
            store_path=state_root / "trust_review.sqlite",
        )
        if refresh_publication.capture_review_identity(data_dir) != identity:
            raise ValueError("candidate changed during review")
        review["publication_artifact_ids"] = identity["artifact_ids"]
        return review
    except Exception as exc:  # noqa: BLE001 - a candidate may be saved, never falsely reviewed
        return {"status": "failed", "error": f"{type(exc).__name__}: {str(exc)[:300]}"}


def activate_publication(**kwargs) -> dict:
    from sinetti_services.refresh_publication import publish_to_production

    return publish_to_production(**kwargs)


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


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def engine_report(engine) -> dict:
    """Version + model checksum of the Tesseract the engine will call, and whether it is pinned."""
    executable = engine._executable()  # noqa: SLF001 -- same resolution the OCR itself uses
    if not executable:
        return {"executable": "", "version": "", "model_sha256": "",
                "problem": ocr_job.engine_pin_problem(version="", model_sha256="")}

    def _run(*args: str) -> str:
        done = subprocess.run([executable, *args], capture_output=True, text=True, timeout=60,
                              check=False)
        return (done.stdout or "") + (done.stderr or "")

    version = ocr_job.tesseract_series(_run("--version"))
    model_dir = ocr_job.tessdata_dir(_run("--list-langs"))
    model = Path(model_dir) / "eng.traineddata" if model_dir else None
    model_sha = ocr_job.sha256_file(model) if model is not None and model.is_file() else ""
    return {
        "executable": executable,
        "version": version,
        "model_path": str(model or ""),
        "model_sha256": model_sha,
        "problem": ocr_job.engine_pin_problem(version=version, model_sha256=model_sha),
    }


def download_pdfs(queue: list[dict], pdf_dir: Path, *, fetch=None) -> tuple[list[dict], list[dict]]:
    """(entries with a local PDF, failed download attempts). A failure is reported, never fatal."""
    if fetch is None:
        from cert_pdf_download import fetch_url_bytes as fetch
    pdf_dir.mkdir(parents=True, exist_ok=True)
    ready, failed = [], []
    for entry in queue:
        target = pdf_dir / f"{entry['certificate_id']}.pdf"
        try:
            data = fetch(entry["pdf_url"], timeout=90)
            if not data.startswith(b"%PDF"):
                raise ValueError("response is not a PDF")
            target.write_bytes(data)
            ready.append({**entry, "local_pdf": target.name})
        except Exception as exc:  # noqa: BLE001 -- one broken link must not stop the run
            failed.append({
                "certificate_id": entry["certificate_id"],
                "disposition": ocr_job.DOWNLOAD_FAILED,
                "ocr_status": "",
                "pdf_sha256": entry.get("pdf_sha256", ""),
                "error": f"{type(exc).__name__}: {str(exc)[:200]}",
            })
    return ready, failed


def run_ocr(entries: list[dict], scratch: Path, engine, *, should_continue) -> tuple[dict, Path]:
    from cert_pdf_ocr import run_iscc_ocr_backlog

    manifest = scratch / "backlog.jsonl"
    manifest.write_text(
        "".join(json.dumps(ocr_job.backlog_manifest_row(entry, f"pdfs/{entry['local_pdf']}")) + "\n"
                for entry in entries),
        encoding="utf-8",
    )
    ocr_db = scratch / "ocr.sqlite"
    report = run_iscc_ocr_backlog(
        manifest_path=manifest,
        pdf_root=scratch / "pdfs",
        root=scratch,
        output_db=ocr_db,
        artifact_path=scratch / "ocr.sqlite.gz",
        ocr_engine=engine,
        resume=False,
        should_continue=should_continue,
    )
    return report, ocr_db


def merge_and_rebuild(*, ocr_db: Path | None, accepted: set[str], scratch: Path, data_dir: Path,

                      rebuild=None, validate=None, force_rebuild: bool = False,

                      redcert: dict | None = None) -> dict:
    """Merge accepted results into the release's details DB; rebuild what reads it when changed.



    `redcert` ({"ocr_db", "accepted"}) is the REDcert lane's share, merged after the ISCC results

    with its own reason code and closing rule. `force_rebuild` (rehearsal only) rebuilds even when

    nothing was promoted, so a rehearsal on a day without recoveries still proves and times the

    rebuild in the cloud image."""
    from cert_pdf_ocr_promote import merge_ocr_recoveries

    details_gz = data_dir / "cert_pdf_details.sqlite.gz"
    work_db = data_dir / "cert_pdf_details.sqlite"
    redcert_accepted = set((redcert or {}).get("accepted") or ())
    result: dict = {"status": "skipped-nothing-accepted", "promoted_ids": [],
                    "redcert_promoted_ids": []}
    promoted: list[str] = []
    redcert_promoted: list[str] = []
    if accepted or redcert_accepted:
        source = scratch / "details-base.sqlite"
        cert_pdf_store.gunzip_artifact(details_gz, source)
        result["status"] = "ok"
        if accepted:
            accepted_db = scratch / "accepted.sqlite"
            ocr_job.write_accepted_db(ocr_db, accepted_db, accepted)
            result["merge"] = merge_ocr_recoveries(
                source, accepted_db, work_db,
                reason_codes=frozenset({ocr_job.OCR_REASON_CODE}), replace=True,
                closed_empty_codes=ocr_job.CLOSED_EMPTY_CODES,
            )
            promoted = sorted(ocr_job.promoted_ids(work_db, accepted))
            source = scratch / "details-after-iscc.sqlite"
            work_db.replace(source)
        if redcert_accepted:
            redcert_db = scratch / "redcert-accepted.sqlite"
            ocr_job.write_accepted_db(redcert["ocr_db"], redcert_db, redcert_accepted)
            result["redcert_merge"] = merge_ocr_recoveries(
                source, redcert_db, work_db,
                reason_codes=frozenset({redcert_lane.OCR_REASON_CODE}), replace=True,
                closed_empty_codes=redcert_lane.CLOSED_EMPTY_CODES,
            )
            redcert_promoted = sorted(redcert_lane.promoted_ids(work_db, redcert_accepted))
        else:
            source.replace(work_db)
        result["promoted_ids"] = promoted
        result["redcert_promoted_ids"] = redcert_promoted
        if not promoted and not redcert_promoted:
            result["status"] = "skipped-nothing-promoted"
    if not promoted and not redcert_promoted:
        work_db.unlink(missing_ok=True)  # the release keeps its own, unchanged, artifact
        if not force_rebuild:
            return result
    else:
        cert_pdf_store.gzip_artifact(work_db, details_gz)
    if rebuild is None:
        from scripts.run_cert_pdf_delta_refresh import rebuild_downstream_from_raw_artifact

        rebuild = rebuild_downstream_from_raw_artifact
    if validate is None:
        import sinetti_data_release

        validate = sinetti_data_release.validate_data_release
    rebuild_started = time.monotonic()
    downstream = rebuild(
        raw_work_db=work_db,
        raw_artifact=details_gz,
        structured_work_db=data_dir / "cert_pdf_structured.sqlite",
        structured_artifact=data_dir / "cert_pdf_structured.sqlite.gz",
        app_root=data_dir.parent,
    )
    result["rebuild"] = {
        "minutes": round((time.monotonic() - rebuild_started) / 60, 1),
        "forced": bool(force_rebuild and not promoted and not redcert_promoted),
        "status": downstream.get("status"),
        "structured": downstream.get("structured"),
        "app_db": {k: (downstream.get("app_db") or {}).get(k) for k in ("status", "error")},
    }
    if (downstream.get("app_db") or {}).get("status") != "ok":
        result["status"] = "error-rebuild"
        return result
    validation = validate(
        data_dir / "sinetti_app.sqlite.gz",
        manifest_path=data_dir / "sinetti_data_release.json",
        inspect_database=True,
    )
    result["validation"] = {"ok": bool(validation.get("ok")), "reason": validation.get("reason")}
    if not validation.get("ok"):
        result["status"] = "error-validation"
    return result


def build_arg_parser() -> argparse.ArgumentParser:
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("--job-workspace", action="store_true",
                    help="required: this checkout is a disposable job workspace whose ./data may "
                         "be replaced by the newest dataset release")
    ap.add_argument("--data-repo-id", default=os.environ.get("SINETTI_DATA_REPO_ID", ""))
    ap.add_argument("--max-certs", type=int,
                    default=_env_int("CLOUD_OCR_MAX_CERTS", ocr_job.DEFAULT_MAX_CERTS))
    ap.add_argument("--redcert-max-certs", type=int,
                    default=_env_int(redcert_lane.MAX_CERTS_ENV, redcert_lane.DEFAULT_MAX_CERTS),
                    help="REDcert scans per run (SIN-0645); 0 keeps the REDcert lane off")
    ap.add_argument("--publish-reserve-s", type=int,
                    default=_env_int("CLOUD_OCR_PUBLISH_RESERVE_S", ocr_job.DEFAULT_PUBLISH_RESERVE_S))
    ap.add_argument("--dpi", type=int, default=300)
    ap.add_argument("--tesseract-cmd", default="tesseract")
    ap.add_argument("--parity-ids", default="",
                    help="comma-separated certificates the PC already OCR'd: re-read them and "
                         "report differences; never publishes")
    ap.add_argument("--retry-ids", default="",
                    help="REHEARSAL ONLY: queue these scanned certificates even though they were "
                         "tried before (proves the OCR path on a day with no new scans)")
    ap.add_argument("--rehearse-rebuild", action="store_true",
                    help="REHEARSAL ONLY: run the structured/app rebuild even when nothing was "
                         "promoted, to prove and time it in the cloud image")
    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)
    args = build_arg_parser().parse_args(argv)
    if rehearsal.is_rehearsal():
        # A mounted production bucket may be present even during rehearsal. Keep every review
        # and final receipt off it, including early failures and independent state-path aliases.
        with tempfile.TemporaryDirectory(prefix="sinetti-ocr-rehearsal-state-") as temporary:
            return _run_job(args, state_root=Path(temporary))
    return _run_job(args, state_root=state_paths.external_state_root() or ROOT / "data")


def _run_job(args: argparse.Namespace, *, state_root: Path) -> int:
    from cert_pdf_ocr import TesseractCliOcrEngine

    started_mono = time.monotonic()
    budget_s = _env_int("SINETTI_JOB_BUDGET_S", 0)
    from sinetti_services import refresh_publication

    activation_enabled = refresh_publication.production_publication_enabled()
    effective_reserve = args.publish_reserve_s
    if activation_enabled:
        effective_reserve = max(
            effective_reserve,
            ocr_job.DEFAULT_PUBLISH_RESERVE_S + refresh_publication.publication_budget_s(),
        )
    run_id = _now_iso()
    report: dict = {
        "producer": PRODUCER, "run_id": run_id, "started_at": run_id,
        "rehearsal": rehearsal.is_rehearsal(),
        "config": {"max_certs": args.max_certs, "redcert_max_certs": args.redcert_max_certs,
                   "publish_reserve_s": effective_reserve,
                   "prod_auto_activate": activation_enabled,
                   "budget_s": budget_s, "dpi": args.dpi,
                   "parity_ids": [c for c in args.parity_ids.split(",") if c.strip()]},
        # SIN-1007: which code this run executed -- a pinned Space commit, or a jobs code release
        # (step 3) -- next to the commit the app ran when it started.
        "code": {key: os.environ.get(name) or "" for key, name in (
            ("space_revision", "SINETTI_SPACE_REVISION"), ("app_revision", "SINETTI_APP_REVISION"),
            ("jobs_code_revision", "SINETTI_JOBS_CODE_REVISION"),
            ("jobs_code_release", "SINETTI_JOBS_CODE_RELEASE"),
            ("jobs_code_source", "SINETTI_JOBS_CODE_SOURCE"))},
    }

    def finish(status: str, code: int) -> int:
        report["status"] = status
        report["finished_at"] = _now_iso()
        report["duration_s"] = round(time.monotonic() - started_mono, 1)
        # This final receipt is outside the candidate snapshot: it can record its real commit.
        ocr_job.write_json(state_root / "refresh" / "last_cloud_ocr_publication.json", report)
        print(json.dumps(report, indent=2, sort_keys=True, default=str), flush=True)
        return code

    retry_ids = [c.strip() for c in args.retry_ids.split(",") if c.strip()]
    if (retry_ids or args.rehearse_rebuild) and not rehearsal.is_rehearsal():
        # Re-reading tried certificates is an owner decision (brief: out of scope), and a forced
        # rebuild publishes nothing new; both exist only to prove the path in a rehearsal.
        report["config"]["refused"] = "--retry-ids / --rehearse-rebuild need SINETTI_REHEARSAL=1"
        return finish("error-rehearsal-only-flag", 2)

    engine = TesseractCliOcrEngine(tesseract_cmd=args.tesseract_cmd, dpi=args.dpi)
    report["engine"] = engine_report(engine)
    print(f"engine: {json.dumps(report['engine'])}", flush=True)
    if report["engine"]["problem"]:
        return finish("error-engine-not-pinned", 1)

    if not args.job_workspace or not args.data_repo_id:
        report["baseline"] = {"status": "error-not-a-job-workspace",
                              "error": "needs --job-workspace and SINETTI_DATA_REPO_ID"}
        return finish("error", 1)
    token = os.environ.get("HF_TOKEN", "") or os.environ.get("HF_JOB_TOKEN", "")
    data_dir = ROOT / "data"
    report["baseline"] = data_plane.boot_sync(
        repo_id=args.data_repo_id, token=token, data_dir=data_dir, materialize=True,
    )
    baseline_revision = report["baseline"].get("hub_revision")
    if report["baseline"].get("status") != "ok" or not baseline_revision:
        return finish("error-baseline", 1)
    details_gz = data_dir / "cert_pdf_details.sqlite.gz"

    with tempfile.TemporaryDirectory(prefix="sinetti-cloud-ocr-") as tmp:
        scratch = Path(tmp)
        parity_ids = report["config"]["parity_ids"]
        if parity_ids:
            baseline_db = scratch / "details-release.sqlite"
            cert_pdf_store.gunzip_artifact(details_gz, baseline_db)
            queue = ocr_job.parity_queue(baseline_db, parity_ids)
            report["queue"] = {"parity": True, "selected": len(queue)}
        else:
            ledger = ocr_job.load_ledger(ROOT / ocr_job.LEDGER_REL)
            seed = ocr_job.load_seed(ROOT / ocr_job.SEED_REL)
            retry_set = set(retry_ids)
            attempted = {cid: pin for cid, pin in ocr_job.attempted_index(ledger, seed).items()
                         if cid not in retry_set}
            queue, report["queue"] = ocr_job.select_queue(
                details_gz, attempted,
                max_certs=len(retry_ids) + 10**6 if retry_ids else args.max_certs,
            )
            if retry_ids:
                wanted = set(retry_ids)
                queue = [entry for entry in queue if entry["certificate_id"] in wanted]
                queue = queue[:args.max_certs]
                report["queue"]["retry_ids"] = len(queue)
            report["queue"]["seed_ids"] = len(seed)
            redcert_ledger = ocr_job.load_ledger(ROOT / redcert_lane.LEDGER_REL)
            redcert_attempted = {
                cid: pin for cid, pin in ocr_job.attempted_index(
                    redcert_ledger, set(),
                    legacy_reader_version=redcert_lane.LEGACY_READER_VERSION).items()
                if cid not in retry_set}
            redcert_queue, redcert_stats = redcert_lane.select_queue(
                details_gz, redcert_attempted,
                max_certs=len(retry_ids) + 10**6 if retry_ids else args.redcert_max_certs,
            )
            if retry_ids:
                redcert_queue = [entry for entry in redcert_queue
                                 if entry["certificate_id"] in retry_set][:args.redcert_max_certs]
                redcert_stats["retry_ids"] = len(redcert_queue)
            report["redcert"] = {"queue": redcert_stats}
            print(f"redcert queue: {json.dumps(redcert_stats)}", flush=True)
        print(f"queue: {json.dumps(report['queue'])}", flush=True)
        if parity_ids:
            redcert_queue = []
        if queue:
            # The reader checks each queued PDF against the published register (SIN-0443). A
            # partial context cannot establish source identity, so refuse before any lane starts.
            queued_ids = sorted({str(entry["certificate_id"]) for entry in queue})
            report["queue"]["register_context_required"] = len(queued_ids)
            report["queue"]["register_context_error"] = ""
            report["queue"]["register_context_missing_ids"] = []
            report["queue"]["register_context_incomplete_fields"] = {}
            try:
                context = ocr_job.load_register_context(
                    data_dir / "sinetti_app.sqlite.gz", set(queued_ids))
            except Exception as exc:  # noqa: BLE001 - report loader failures before processing
                report["queue"]["register_context"] = 0
                report["queue"]["register_context_error"] = f"{type(exc).__name__}: {str(exc)[:200]}"
                return finish("error-register-context", 1)
            missing_ids = [cid for cid in queued_ids if cid not in context]
            incomplete_fields = {
                cid: [field for field in ocr_job.REGISTER_CONTEXT_FIELDS
                      if not str((context.get(cid) or {}).get(field) or "").strip()]
                for cid in queued_ids if cid in context
            }
            incomplete_fields = {cid: fields for cid, fields in incomplete_fields.items() if fields}
            report["queue"]["register_context_missing_ids"] = missing_ids
            report["queue"]["register_context_incomplete_fields"] = incomplete_fields
            report["queue"]["register_context"] = len(context)
            # A certificate the published register does not describe completely (old backlog
            # entries that left the register, a blank holder or period) cannot be checked against
            # its PDF, so it is never read -- but one such entry must not stop the whole run:
            # v2026.10.6 aborted every morning on five such certificates. Skip and report them.
            skipped_ids = sorted(set(missing_ids) | set(incomplete_fields))
            report["queue"]["register_context_skipped_ids"] = skipped_ids
            if skipped_ids:
                skipped = set(skipped_ids)
                queue = [entry for entry in queue if str(entry["certificate_id"]) not in skipped]
                print(f"register context: skipped {len(skipped_ids)} certificate(s) without complete "
                      f"register context: {', '.join(skipped_ids[:10])}", flush=True)
            queue = ocr_job.attach_register_context(queue, context)
        if redcert_queue:
            redcert_queue, context_report = redcert_lane.attach_reviewed_register_context(
                redcert_queue, data_dir / "sinetti_app.sqlite.gz")
            report["redcert"]["queue"].update(context_report)
            if context_report["register_context_error"]:
                return finish("error-redcert-register-context", 1)
        if not queue and not redcert_queue:
            return finish("ok-nothing-to-do", 0)

        allocation_started = time.monotonic()
        iscc_deadline = iscc_lane_deadline(
            started_mono, allocation_started, budget_s, effective_reserve,
            len(queue), len(redcert_queue))
        if iscc_deadline is not None:
            available = max(0.0, budget_s - (allocation_started - started_mono)
                            - effective_reserve)
            report["config"]["ocr_budget_allocation"] = {
                "iscc_selected": len(queue), "redcert_selected": len(redcert_queue),
                "work_window_s": round(available, 1),
                "iscc_share_s": round(iscc_deadline - allocation_started, 1),
            }

        def should_continue() -> bool:
            return can_start_ocr(
                now_monotonic=time.monotonic(), started_monotonic=started_mono,
                budget_s=budget_s, reserve_s=effective_reserve)

        def iscc_should_continue() -> bool:
            return can_start_ocr(
                now_monotonic=time.monotonic(), started_monotonic=started_mono,
                budget_s=budget_s, reserve_s=effective_reserve, lane_deadline=iscc_deadline)

        ocr_db: Path | None = None
        results: list[dict] = []
        download_failed: list[dict] = []
        ocr_report: dict = {"status": "ok", "queued": 0}
        ocr_minutes = 0.0
        if queue:
            ready, download_failed = download_pdfs(queue, scratch / "pdfs")
            report["download"] = {"ok": len(ready), "failed": len(download_failed),
                                  "failures": download_failed[:10]}
            ocr_started = time.monotonic()
            ocr_report, ocr_db = run_ocr(
                ready, scratch, engine, should_continue=iscc_should_continue)
            results = ocr_job.classify_results(ocr_db) if ocr_db.exists() else []
            ocr_minutes = (time.monotonic() - ocr_started) / 60
        report["ocr"] = {
            **{k: ocr_report.get(k) for k in ("status", "queued", "parsed", "no_materials",
                                              "failed", "stopped_no_time", "reason")},
            "minutes": round(ocr_minutes, 1),
            "minutes_per_certificate": round(ocr_minutes / len(results), 2) if results else None,
            "dispositions": {
                name: sum(1 for r in results if r["disposition"] == name)
                for name in (ocr_job.ACCEPT, ocr_job.HOLD, ocr_job.UNREADABLE)
            },
        }
        if ocr_report.get("status") != "ok":
            return finish("error-ocr", 1)

        if parity_ids:
            report["parity"] = ocr_job.parity_compare(baseline_db, ocr_db, parity_ids)
            return finish("ok-parity", 0)

        # REDcert lane (SIN-0645): same engine and time budget, after the ISCC lane.
        redcert_results: list[dict] = []
        redcert_failed: list[dict] = []
        redcert_ocr_db = scratch / "redcert-ocr.sqlite"
        if redcert_queue:
            redcert_pdfs = scratch / "redcert-pdfs"
            redcert_ready, redcert_failed, redcert_stale = redcert_lane.fetch_hosted_pdfs(
                redcert_queue, redcert_pdfs)
            report["redcert"]["download"] = {
                "ok": len(redcert_ready), "failed": len(redcert_failed),
                "stale": len(redcert_stale), "failures": redcert_failed[:10],
                "stale_files": redcert_stale[:10]}
            redcert_started = time.monotonic()
            redcert_report = redcert_lane.run_ocr(
                redcert_ready, redcert_pdfs, redcert_ocr_db, engine=engine,
                should_continue=should_continue)
            if redcert_ocr_db.exists():
                redcert_results = redcert_lane.classify_results(redcert_ocr_db)
            report["redcert"]["ocr"] = {
                **redcert_report,
                "minutes": round((time.monotonic() - redcert_started) / 60, 1),
                "dispositions": {
                    name: sum(1 for r in redcert_results if r["disposition"] == name)
                    for name in (redcert_lane.ACCEPT, redcert_lane.HOLD, redcert_lane.UNREADABLE)
                },
            }
            if redcert_report.get("status") != "ok":
                return finish("error-ocr", 1)
        redcert_accepted = {r["certificate_id"] for r in redcert_results
                            if r["disposition"] == redcert_lane.ACCEPT}

        accepted = {r["certificate_id"] for r in results if r["disposition"] == ocr_job.ACCEPT}
        report["merge"] = merge_and_rebuild(
            ocr_db=ocr_db, accepted=accepted, scratch=scratch, data_dir=data_dir,
            force_rebuild=args.rehearse_rebuild,
            redcert={"ocr_db": redcert_ocr_db, "accepted": redcert_accepted},
        )
        if report["merge"]["status"].startswith("error"):
            return finish("error-merge", 1)
        promoted = set(report["merge"]["promoted_ids"])
        reader = ocr_job.OCR_READER_VERSION
        attempts = [
            {**r, "promoted": r["certificate_id"] in promoted, "reader": reader} for r in results
        ] + download_failed
        redcert_promoted = set(report["merge"]["redcert_promoted_ids"])
        redcert_attempts = [
            {**r, "promoted": r["certificate_id"] in redcert_promoted,
             "reader": redcert_lane.OCR_READER_VERSION} for r in redcert_results
        ] + redcert_failed
        report["attempts"] = len(attempts) + len(redcert_attempts)
        if not report["attempts"]:
            return finish("ok-nothing-attempted", 0)
        if attempts:
            ledger = ocr_job.record_attempts(ledger, attempts, run_id=run_id)
            ocr_job.write_json(ROOT / ocr_job.LEDGER_REL, ledger)
            report["ledger_entries"] = len(ledger["entries"])
        if redcert_attempts:
            redcert_ledger = ocr_job.record_attempts(redcert_ledger, redcert_attempts, run_id=run_id)
            ocr_job.write_json(ROOT / redcert_lane.LEDGER_REL, redcert_ledger)
            report["redcert"]["ledger_entries"] = len(redcert_ledger["entries"])
        # The candidate's internal receipt predates upload and cannot contain its own commit.
        ocr_job.write_json(ROOT / ocr_job.RECEIPT_REL, {**report, "status": "candidate-built"})

    with tempfile.TemporaryDirectory(prefix="sinetti-ocr-candidate-") as tmp:
        candidate_root = Path(tmp)
        candidate_data_dir = candidate_root / "data"
        # Physical copies freeze snapshot symlinks before checks, upload and activation.
        shutil.copytree(data_dir, candidate_data_dir, copy_function=shutil.copyfile)
        report["trust_review"] = review_candidate(data_dir=candidate_data_dir, state_root=state_root)
        try:
            artifact_ids = report["trust_review"].get("publication_artifact_ids")
            if artifact_ids and refresh_publication.capture_review_identity(candidate_data_dir)["artifact_ids"] != artifact_ids:
                raise ValueError("candidate changed after review")
            report["publish"] = data_plane.publish(
                candidate_root, repo_id=args.data_repo_id, token=token, producer=PRODUCER,
                expected_parent=baseline_revision,
            )
        except Exception as exc:  # noqa: BLE001 - stale parent or mutation saves no release
            report["publish"] = {"status": "error",
                                 "error": f"{type(exc).__name__}: {str(exc)[:300]}"}
        if report["publish"].get("status") != "ok":
            return finish("error-publish", 1)
        # Flush independent refresh success before a restart/rollback can fail or time out.
        ocr_job.write_json(state_root / "refresh" / "last_cloud_ocr_publication.json", report)
        if activation_enabled and not report["publish"].get("rehearsal"):
            space = os.environ.get("HF_SPACE_REPO_ID", "")
            base_url = os.environ.get("SINETTI_PROD_AUTO_ACTIVATE_BASE_URL") or (
                f"https://{space.replace('/', '-').lower()}.hf.space" if space else ""
            )
            try:
                report["auto_activation"] = activate_publication(
                    publish=report["publish"], review=report["trust_review"],
                    candidate_data_dir=candidate_data_dir, state_root=state_root,
                    space_repo_id=space, token=token, base_url=base_url, by=PRODUCER,
                    remaining_s=remaining_budget_s(started_monotonic=started_mono, budget_s=budget_s,
                                                   now_monotonic=time.monotonic()),
                )
            except Exception as exc:  # noqa: BLE001 - preserve successful refresh and surface failure
                report["auto_activation"] = {"status": "error",
                                             "reason": f"{type(exc).__name__}: {str(exc)[:300]}"}
        else:
            report["auto_activation"] = {"status": "skipped", "reason": "rehearsal-or-disabled"}
    return finish("ok", 0)


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