Spaces:
Running
Running
#!/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())
#!/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