Blame for scripts/refresh_quality.py

"""Validation and durable history helpers for the daily refresh pipeline.



Pure helpers only: no network calls and no Streamlit imports.

"""
from __future__ import annotations

import csv
import hashlib
import io
import json
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
from typing import Mapping

from sinetti_services import certificate_succession, renewal_match
from sinetti_services.event_classification import expired_before
from sinetti_services.returning_holder import classify_returning, format_off_scheme_evidence


_SEVERITY_RANK = {"ok": 0, "warning": 1, "critical": 2}


def _utc_iso(value=None) -> str:
    if value is None:
        value = datetime.now(timezone.utc)
    if isinstance(value, str):
        return value
    if value.tzinfo is None:
        value = value.replace(tzinfo=timezone.utc)
    return value.astimezone(timezone.utc).isoformat()


def _norm_header(value: str) -> str:
    return "".join(ch for ch in str(value or "").strip().lower() if ch.isalnum())


def _check(severity: str, code: str, message: str, **context) -> dict:
    out = {"severity": severity, "code": code, "message": message}
    if context:
        out["context"] = context
    return out


def _status(checks: list[dict]) -> str:
    worst = "ok"
    for check in checks:
        sev = check.get("severity", "ok")
        if _SEVERITY_RANK.get(sev, 0) > _SEVERITY_RANK[worst]:
            worst = sev
    return worst


def _ratio(ok: int, total: int) -> float:
    return round((ok / total), 4) if total else 0.0


def _parse_date(value: str) -> bool:
    raw = str(value or "").strip()
    if not raw:
        return False
    for fmt in ("%Y-%m-%d", "%d.%m.%Y", "%d/%m/%Y"):
        try:
            datetime.strptime(raw, fmt)
            return True
        except ValueError:
            pass
    return False


def _collapse_check(checks: list[dict], row_count: int, previous_count: int) -> None:
    if previous_count <= 0:
        return
    if row_count < max(1, int(previous_count * 0.5)):
        checks.append(_check(
            "critical", "row_count_collapse",
            "Row count collapsed versus the previous accepted snapshot.",
            previous_count=previous_count, row_count=row_count,
        ))
    elif row_count > int(previous_count * 2.0):
        checks.append(_check(
            "warning", "row_count_spike",
            "Row count is much larger than the previous accepted snapshot.",
            previous_count=previous_count, row_count=row_count,
        ))


_ISCC_HEADER_GROUPS = {
    "certificate_id": {"certificate"},
    "holder_name": {"certificateholder"},
    "country": {"country"},
    "scope": {"scope"},
    "feedstocks": {"rawmaterial"},
    "valid_from": {"validfrom"},
    "valid_until": {"validuntil"},
    "issuing_cb": {"issuingcb", "certificationbody"},
}


def validate_iscc_csv_text(csv_text: str, *, previous_count: int = 0) -> dict:
    text = str(csv_text or "").lstrip("\ufeff")
    reader = csv.DictReader(io.StringIO(text))
    headers = list(reader.fieldnames or [])
    norm_headers = {_norm_header(h) for h in headers}
    rows = list(reader)
    checks: list[dict] = []
    missing = [
        name for name, aliases in _ISCC_HEADER_GROUPS.items()
        if not (aliases & norm_headers)
    ]
    if missing:
        checks.append(_check(
            "critical", "missing_required_columns",
            "ISCC CSV is missing required columns.",
            missing_columns=missing,
        ))
    if not rows:
        checks.append(_check("critical", "empty_export", "ISCC CSV parsed zero data rows."))
    if text.lstrip().startswith("<"):
        checks.append(_check("critical", "html_response", "ISCC CSV looks like an HTML response."))
    _collapse_check(checks, len(rows), int(previous_count or 0))

    def value_for(row: dict, canonical: str) -> str:
        aliases = _ISCC_HEADER_GROUPS[canonical]
        for key, value in row.items():
            if _norm_header(key) in aliases:
                return str(value or "").strip()
        return ""

    if rows and not missing:
        id_rate = _ratio(sum(1 for r in rows if value_for(r, "certificate_id")), len(rows))
        holder_rate = _ratio(sum(1 for r in rows if value_for(r, "holder_name")), len(rows))
        date_rate = _ratio(sum(1 for r in rows if _parse_date(value_for(r, "valid_until"))), len(rows))
        if id_rate < 0.98:
            checks.append(_check("critical", "low_certificate_id_coverage",
                                 "Too many ISCC rows have no certificate id.", coverage=id_rate))
        if holder_rate < 0.95:
            checks.append(_check("critical", "low_holder_coverage",
                                 "Too many ISCC rows have no holder name.", coverage=holder_rate))
        if date_rate < 0.90:
            checks.append(_check("warning", "low_valid_until_parse_rate",
                                 "Many ISCC valid-until dates did not parse.", coverage=date_rate))
    else:
        date_rate = 0.0

    matched_headers = set().union(*_ISCC_HEADER_GROUPS.values())
    return {
        "status": _status(checks),
        "row_count": len(rows),
        "missing_columns": missing,
        "extra_columns": [h for h in headers if _norm_header(h) not in matched_headers],
        "date_parse_rate": date_rate,
        "checks": checks or [_check("ok", "contract_ok", "ISCC CSV contract passed.")],
    }


_IN_FORCE_STATUSES = frozenset({"active", "valid", "expiring"})


def validate_normalized_certs(scheme: str, certs: list[dict], *, previous_count: int = 0) -> dict:
    rows = list(certs or [])
    checks: list[dict] = []
    if not rows:
        checks.append(_check("critical", "empty_normalized_snapshot",
                             f"{scheme} normalized snapshot has zero rows."))
    _collapse_check(checks, len(rows), int(previous_count or 0))
    required = ["certificate_id", "holder_name", "holder_norm"]
    if str(scheme).lower() != "ins":
        required += ["country", "valid_until"]
    bad_fields = []
    # RSB publishes no holder country for lapsed certificates (its holder-country comes from the
    # certificate PDF, which the register only keeps for certificates still in force). Country
    # coverage for RSB is therefore judged over the in-force rows; every other rule, and every
    # other scheme, is judged over all rows. Brief 2026-09-29-rsb-register-brief.md.
    in_force_rows = [
        row for row in rows
        if str(row.get("status") or "").strip().casefold() in _IN_FORCE_STATUSES
    ]
    for field in required:
        judged = (
            in_force_rows
            if str(scheme).lower() == "rsb" and field == "country" and in_force_rows
            else rows
        )
        coverage = _ratio(sum(1 for row in judged if str(row.get(field, "")).strip()), len(judged))
        if coverage < 0.90:
            bad_fields.append(field)
            checks.append(_check(
                "critical", "low_field_coverage",
                f"{scheme} field coverage is below threshold.",
                field=field, coverage=coverage,
            ))
    statuses = sorted({str(row.get("status") or row.get("raw_status") or "").strip() for row in rows})
    return {
        "status": _status(checks),
        "row_count": len(rows),
        "bad_fields": bad_fields,
        "status_values": [s for s in statuses if s][:25],
        "checks": checks or [_check("ok", "contract_ok", f"{scheme} normalized contract passed.")],
    }


def validate_history_retention(

    scheme: str,

    old_rows: list[dict],

    new_rows: list[dict],

    *,

    today: str | None = None,

) -> dict:
    """Guard for registers that keep their expired certificates (REDcert publishes history back

    to 2010). Our snapshot write is replace-all, so if the register ever trims old rows the trim

    would silently propagate into our corpus on the next refresh. Expired rows leaving the export

    are anomalous for such a register: any disappearance is a warning, and a disappearance beyond

    max(25, 1% of the previously expired rows) is critical — the refresh keeps the previous

    snapshot. Not for valid-only feeds (ISCC), where expired rows leaving is normal lifecycle."""
    checks: list[dict] = []
    today = today or datetime.now(timezone.utc).date().isoformat()
    new_ids = {
        str(row.get("certificate_id") or "").strip().casefold()
        for row in (new_rows or [])
        if str(row.get("certificate_id") or "").strip()
    }
    expired_old = [
        row for row in (old_rows or [])
        if len(str(row.get("valid_until") or "").strip()) == 10
        and str(row.get("valid_until")).strip() < today
    ]
    missing = [
        row for row in expired_old
        if str(row.get("certificate_id") or "").strip().casefold() not in new_ids
    ]
    threshold = max(25, int(0.01 * len(expired_old)))
    if missing:
        severity = "critical" if len(missing) > threshold else "warning"
        checks.append(_check(
            severity, "history_trim",
            f"{scheme} export dropped previously expired certificates — possible register "
            "history trim.",
            missing_expired=len(missing),
            expired_previous=len(expired_old),
            threshold=threshold,
            examples=sorted(
                str(row.get("certificate_id") or "").strip() for row in missing
            )[:5],
        ))
    return {
        "status": _status(checks),
        "row_count": len(new_rows or []),
        "missing_expired": len(missing),
        "checks": checks or [_check(
            "ok", "history_retained",
            f"{scheme} export still carries all previously expired certificates.",
        )],
    }


def validate_tabular_headers(source: str, rows: list[dict], header_groups: dict[str, set[str]]) -> dict:
    rows = list(rows or [])
    checks: list[dict] = []
    if not rows:
        checks.append(_check("critical", "empty_table", f"{source} parsed zero rows."))
        headers = set()
    else:
        headers = {_norm_header(k) for row in rows[:10] for k in row}
    normalized_groups = {
        name: {_norm_header(alias) for alias in aliases}
        for name, aliases in (header_groups or {}).items()
    }
    missing = [
        name for name, aliases in normalized_groups.items()
        if not (aliases & headers)
    ]
    if missing:
        checks.append(_check(
            "critical", "missing_required_columns",
            f"{source} is missing required columns.",
            missing_columns=missing,
        ))
    return {
        "status": _status(checks),
        "row_count": len(rows),
        "missing_columns": missing,
        "checks": checks or [_check("ok", "contract_ok", f"{source} header contract passed.")],
    }


def validate_compliance_records(records: list[dict], *, previous_count: int = 0) -> dict:
    rows = list(records or [])
    checks: list[dict] = []
    _collapse_check(checks, len(rows), int(previous_count or 0))
    if not rows:
        checks.append(_check("warning", "empty_compliance_snapshot",
                             "Compliance snapshot has zero rows."))
    status_values = {str(r.get("status") or r.get("raw_status") or "").lower() for r in rows}
    unexpected = sorted(v for v in status_values if v and v not in {"suspended", "withdrawn"})
    if unexpected:
        checks.append(_check("warning", "unexpected_compliance_status",
                             "Compliance snapshot contains unexpected status values.",
                             values=unexpected[:10]))
    id_rate = _ratio(sum(1 for r in rows if str(r.get("certificate_id", "")).strip()), len(rows))
    if rows and id_rate < 0.90:
        checks.append(_check("critical", "low_certificate_id_coverage",
                             "Too many compliance rows have no certificate id.", coverage=id_rate))
    return {
        "status": _status(checks),
        "row_count": len(rows),
        "status_values": sorted(v for v in status_values if v),
        "checks": checks or [_check("ok", "contract_ok", "Compliance contract passed.")],
    }


def validate_blocklist_records(records: list[dict], *, previous_count: int = 0) -> dict:
    rows = list(records or [])
    checks: list[dict] = []
    _collapse_check(checks, len(rows), int(previous_count or 0))
    if not rows:
        checks.append(_check("warning", "empty_blocklist_snapshot",
                             "Blocklist snapshot has zero rows."))
    category_rate = _ratio(sum(1 for r in rows if str(r.get("category", "")).strip()), len(rows))
    name_rate = _ratio(sum(
        1 for r in rows
        if str(r.get("name") or r.get("holder_name") or r.get("company") or "").strip()
    ),
                       len(rows))
    if rows and category_rate < 0.95:
        checks.append(_check("critical", "low_category_coverage",
                             "Too many blocklist rows have no category.", coverage=category_rate))
    if rows and name_rate < 0.90:
        checks.append(_check("critical", "low_name_coverage",
                             "Too many blocklist rows have no name.", coverage=name_rate))
    return {
        "status": _status(checks),
        "row_count": len(rows),
        "checks": checks or [_check("ok", "contract_ok", "Blocklist contract passed.")],
    }


def validate_ins_blobs(blobs: dict) -> dict:
    checks: list[dict] = []
    counts = {k: len(blobs.get(k) or []) for k in ("operatori", "tipologie", "prodotti")}
    if counts["operatori"] <= 0:
        checks.append(_check("critical", "missing_operatori_blob",
                             "INS operator blob is missing or empty."))
    for key in ("tipologie", "prodotti"):
        if counts[key] <= 0:
            checks.append(_check("warning", f"missing_{key}_blob",
                                 f"INS {key} blob is missing or empty."))
    return {
        "status": _status(checks),
        "row_count": counts["operatori"],
        "blob_counts": counts,
        "checks": checks or [_check("ok", "contract_ok", "INS embedded blobs passed.")],
    }


def combine_validations(*reports: dict) -> dict:
    checks: list[dict] = []
    missing_columns: list[str] = []
    bad_fields: list[str] = []
    row_count = 0
    for report in reports:
        if not isinstance(report, dict):
            continue
        checks.extend(report.get("checks", []) or [])
        missing_columns.extend(report.get("missing_columns", []) or [])
        bad_fields.extend(report.get("bad_fields", []) or [])
        row_count = max(row_count, int(report.get("row_count") or 0))
    return {
        "status": _status(checks),
        "row_count": row_count,
        "missing_columns": sorted(set(missing_columns)),
        "bad_fields": sorted(set(bad_fields)),
        "checks": checks or [_check("ok", "contract_ok", "Combined contract passed.")],
    }


def validation_is_critical(report: dict) -> bool:
    return (report or {}).get("status") == "critical"


def append_jsonl(path: str | Path, record: dict) -> None:
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    with p.open("a", encoding="utf-8") as fh:
        fh.write(json.dumps(record, sort_keys=True, ensure_ascii=False) + "\n")


def read_jsonl(path: str | Path, *, limit: int | None = None) -> dict:
    p = Path(path)
    records: list[dict] = []
    skipped = 0
    if not p.exists():
        return {"records": [], "skipped": 0}
    for line in p.read_text(encoding="utf-8").splitlines():
        raw = line.strip()
        if not raw:
            continue
        try:
            value = json.loads(raw)
        except json.JSONDecodeError:
            skipped += 1
            continue
        if isinstance(value, dict):
            records.append(value)
        else:
            skipped += 1
    if limit is not None:
        records = records[-max(0, int(limit)):]
    return {"records": records, "skipped": skipped}


def sha256_file(path: str | Path) -> str:
    p = Path(path)
    h = hashlib.sha256()
    with p.open("rb") as fh:
        for chunk in iter(lambda: fh.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def content_sha256(records: list[dict]) -> str:
    """Order-independent content hash over a list of records. Unlike sha256_file(), this hashes

    the DATA, not a container — a gzip file's header embeds the write timestamp, so hashing the

    file makes byte-identical content produce a different hash on every run (see

    fingerprint_record's `records` param)."""
    canon = sorted(
        json.dumps(r, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
        for r in records
    )
    return hashlib.sha256("".join(canon).encode("utf-8")).hexdigest()


def fingerprint_record(*, run_id: str, scheme: str, kind: str, path: str | Path,

                       count: int, source: str = "daily_refresh",

                       accepted_at=None, records: list[dict] | None = None) -> dict:
    return {
        "run_id": run_id,
        "scheme": scheme,
        "kind": kind,
        "count": int(count or 0),
        "sha256": content_sha256(records) if records is not None else sha256_file(path),
        "source": source,
        "accepted_at": _utc_iso(accepted_at),
    }


# A certificate that leaves a risky status for one of these did not come back — it ended.
_TERMINAL_STATUSES = {"expired", "terminated", "withdrawn"}

_RISKY_STATUSES = {
    "suspended": ("certificate_suspended", "high"),
    # Authority cancellation across every register (REDcert "entzogen", INS "ANNULLATO",
    # LCFS/BC/CFR "withdrawn") now lands on the single harmonised status, so it emits the
    # single event type. "certificate_revoked_or_terminated" survives for terminated only:
    # the event log is append-only, so its stored name keeps its historical spelling.
    "withdrawn": ("certificate_withdrawn", "high"),
    "terminated": ("certificate_revoked_or_terminated", "low"),
}


def _cert_key(cert: dict) -> str:
    return str(cert.get("certificate_id") or "").strip()


def _status_value(cert: dict) -> str:
    return str(cert.get("status") or cert.get("raw_status") or "").strip().lower()


def _risk_category(event_type: str, *, status: str = "") -> str:
    if event_type == "certificate_revoked_or_terminated" and status == "terminated":
        return "status_change"
    if event_type in {"certificate_suspended", "certificate_withdrawn", "certificate_revoked_or_terminated"}:
        return "compliance_risk"
    if event_type in {"certificate_removed_from_valid_list", "certificate_reinstated"}:
        return "status_change"
    if event_type in {"holder_new", "holder_returning"}:
        return "market_intelligence"
    return "operational"


def _as_str_list(value) -> list[str]:
    if value is None:
        return []
    if isinstance(value, (list, tuple, set)):
        return [str(item).strip() for item in value if str(item).strip()]
    text = str(value).strip()
    return [text] if text else []


def _cert_context(cert: dict) -> dict:
    return {
        "country": str(cert.get("country") or "").strip(),
        "scope": _as_str_list(cert.get("scope")),
        "feedstocks": _as_str_list(cert.get("feedstocks")),
        "valid_from": str(cert.get("valid_from") or "").strip(),
        "valid_until": str(cert.get("valid_until") or "").strip(),
        "issuing_cb": str(cert.get("issuing_cb") or "").strip(),
        "status": _status_value(cert),
    }


def _event(run_id: str, scheme: str, event_type: str, severity: str, confidence: str,

           cert: dict, *, detected_at, evidence: str, old=None, new=None) -> dict:
    new = new or {}
    status = str((new.get("status") if isinstance(new, dict) else "") or "").strip().lower()
    if not status:
        status = _status_value(cert)
    event = {
        "run_id": run_id,
        "detected_at": _utc_iso(detected_at),
        "scheme": scheme,
        "event_type": event_type,
        "severity": severity,
        "risk_category": _risk_category(event_type, status=status),
        "confidence": confidence,
        "certificate_id": _cert_key(cert),
        "holder_norm": str(cert.get("holder_norm", "")),
        "holder_name": str(cert.get("holder_name", "")),
        "old": old or {},
        "new": new,
        "evidence": evidence,
    }
    event.update(_cert_context(cert))
    # ISCC publishes the suspension window on the compliance register (`cert_suspended_from` /
    # `cert_suspended_until`). Keep it on the event so an episode has real dates instead of only
    # the observation interval between two nightly runs. Absent keys stay absent: every other
    # event type would otherwise gain two permanently empty fields.
    for key in ("suspended_date", "suspended_until"):
        value = str(cert.get(key) or "").strip()
        if value:
            event[key] = value
    return event

# Event types that settle a withdrawn certificate's story: once one is on the append-only log
# for a certificate, the held-withdrawal branch below must never fire for it again.
_SETTLING_EVENT_TYPES = frozenset({
    "certificate_withdrawn", "certificate_replaced", "certificate_revoked_or_terminated",
})

# A held withdrawal older than this is left alone: it predates event logging (or the log lost
# it), and the row already reads withdrawn. Keeps a first run with an incomplete log from
# emitting a night of thousands of stale high-severity events.
_HELD_WITHDRAWAL_MAX_AGE_DAYS = 60


def settled_certificate_ids(events: list[dict], scheme: str) -> set[str]:
    """Certificate ids of `scheme` that already carry a withdrawn / replaced / terminated event."""
    out: set[str] = set()
    for event in events or []:
        if not isinstance(event, dict):
            continue
        if str(event.get("scheme") or "") != scheme:
            continue
        if str(event.get("event_type") or "") not in _SETTLING_EVENT_TYPES:
            continue
        cid = str(event.get("certificate_id") or "").strip()
        if cid:
            out.add(cid)
    return out


def _days_since(day_text: object, detected_iso: str) -> int | None:
    day = certificate_succession._parse_date(day_text)
    detected = certificate_succession._parse_date(detected_iso)
    if day is None or detected is None:
        return None
    return (detected - day).days


def _replaced_event(run_id: str, scheme: str, pair, *, detected_at, old_status: str) -> dict:
    old_id = _cert_key(pair.old)
    new_id = _cert_key(pair.new)
    changes = list(pair.changes)
    change_text = "; ".join(changes) if changes else "no visible scope or feedstock change"
    event = _event(
        run_id, scheme, "certificate_replaced", "low", "inferred", pair.old,
        detected_at=detected_at,
        evidence=(
            f"{scheme} re-issued this certificate under a new number: {new_id} starts "
            f"{pair.new.get('valid_from') or '?'} ({change_text}). The register marks the old "
            f"number withdrawn when it re-issues; this is not an enforcement action."
        ),
        old={"status": old_status, "certificate_id": old_id},
        new={
            "status": _status_value(pair.old),
            "replaced_by": new_id,
            "valid_from": str(pair.new.get("valid_from") or ""),
            "valid_until": str(pair.new.get("valid_until") or ""),
            "changes": changes,
        },
    )
    event["replaced_by"] = new_id
    event["replacement_changes"] = changes
    event["replacement_gap_days"] = pair.gap_days
    return event


def build_change_events(run_id: str, scheme: str, old_certs: list[dict], new_certs: list[dict],

                        *, detected_at=None, status_authority: list[dict] | None = None,

                        authority_source: str | None = None,

                        prior_coverage: Mapping[tuple[str, str], dict] | None = None,

                        settled_certificate_ids: set[str] | None = None) -> list[dict]:
    """Diff two snapshots of one scheme into change events.



    `settled_certificate_ids` -- certificate ids that already carry a withdrawn / replaced /

    terminated event on the log (see `settled_certificate_ids()`). When given, a REDcert

    withdrawal with no successor yet is HELD for `certificate_succession.REPLACEMENT_WINDOW_DAYS`

    and fires high only once that window passes with no replacement; when None, the flip fires

    immediately as before (callers that do not keep a log)."""
    old_by_id = {_cert_key(c): c for c in (old_certs or []) if _cert_key(c)}
    new_by_id = {_cert_key(c): c for c in (new_certs or []) if _cert_key(c)}
    authority_by_id = {_cert_key(c): c for c in (status_authority or []) if _cert_key(c)}
    source_label = (authority_source or "").strip() or f"{scheme} status authority"
    old_holders = {
        str(c.get("holder_norm", "")).strip()
        for c in (old_certs or [])
        if str(c.get("holder_norm", "")).strip()
    }
    events: list[dict] = []
    detected_iso = _utc_iso(detected_at)

    # ISCC and REDcert renumber on renewal, so a routine renewal reaches us as one certificate
    # disappearing and a differently-numbered one appearing. Pair those up BEFORE classifying
    # either side: unpaired, the successor reads as a brand-new certificate and the superseded
    # certificate reads as an exit — a false high-severity "removed while still valid" alarm
    # whenever the renewal was issued before the old certificate expired.
    renewals = renewal_match.match_renewals(
        [old for cid, old in old_by_id.items() if cid not in new_by_id],
        [new for cid, new in new_by_id.items() if cid not in old_by_id],
    )
    renewed_from = {_cert_key(pair.new): pair.old for pair in renewals}
    superseded = {_cert_key(pair.old) for pair in renewals}

    # REDcert marks the OLD number withdrawn when it re-issues a certificate (scope change,
    # correction, early renewal) -- 2,758 of its 3,744 withdrawn certificates have a same-holder
    # successor within 3 days. Read literally, that is a high-severity enforcement signal on a
    # holder that simply got a new number. Pair withdrawn rows with their successor in the NEW
    # snapshot first; a paired withdrawal becomes one low-severity `certificate_replaced` and its
    # successor's appearance is folded into it (brief 2026-09-17).
    replacement_by_old: dict[str, certificate_succession.Replacement] = {}
    if scheme in certificate_succession.REPLACEMENT_SCHEMES:
        for pair in certificate_succession.match_replacements(list(new_by_id.values()), scheme=scheme):
            replacement_by_old[_cert_key(pair.old)] = pair
    replaced_successors = {_cert_key(pair.new) for pair in replacement_by_old.values()}
    hold_withdrawals = (
        settled_certificate_ids is not None
        and scheme in certificate_succession.REPLACEMENT_SCHEMES
    )
    window_days = certificate_succession.REPLACEMENT_WINDOW_DAYS

    for cid, old in sorted(old_by_id.items()):
        if cid not in new_by_id:
            if cid in superseded:
                continue        # not an exit — the holder renewed under a new number
            if expired_before(old.get("valid_until"), detected_iso):
                events.append(_event(
                    run_id, scheme, "certificate_expired", "low", "explicit", old,
                    detected_at=detected_at,
                    evidence=f"{scheme} certificate reached its expiry date",
                    old={"status": _status_value(old)}, new={},
                ))
            else:
                auth = authority_by_id.get(cid)
                auth_status = _status_value(auth) if auth else ""
                if auth is not None and auth_status in _RISKY_STATUSES:
                    event_type, severity = _RISKY_STATUSES[auth_status]
                    # Prefer old row for holder identity; overlay status fields from authority.
                    cert_for_event = dict(old)
                    for key in ("status", "raw_status", "suspended_date", "suspended_until",
                                "valid_until", "issuing_cb"):
                        if auth.get(key):
                            cert_for_event[key] = auth.get(key)
                    # Valid-only feeds (ISCC CSV) often carry raw codes like "1", not a named
                    # lifecycle label — omit those from old.status so digests say "suspended"
                    # rather than "1 → suspended".
                    prior_status = _status_value(old)
                    if prior_status and prior_status not in _RISKY_STATUSES and prior_status not in {
                        "active", "valid", "expiring", "expired",
                    }:
                        prior_status = ""
                    events.append(_event(
                        run_id, scheme, event_type, severity, "explicit", cert_for_event,
                        detected_at=detected_at,
                        evidence=source_label,
                        old={"status": prior_status},
                        new={"status": auth_status},
                    ))
                else:
                    events.append(_event(
                        run_id, scheme, "certificate_removed_from_valid_list", "high", "inferred", old,
                        detected_at=detected_at,
                        evidence=(
                            f"{scheme} certificate left accepted snapshot while still valid "
                            f"(no terminated / suspended / withdrawn match)"
                        ),
                        old={"status": _status_value(old)}, new={},
                    ))
    for cid, new in sorted(new_by_id.items()):
        if cid not in old_by_id:
            if cid in replaced_successors:
                continue        # its appearance IS the replacement, reported on the old number
            predecessor = renewed_from.get(cid)
            if predecessor is not None:
                events.append(_event(
                    run_id, scheme, "certificate_renewed", "low", "inferred", new,
                    detected_at=detected_at,
                    evidence=(f"{scheme} certificate renewed under a new number "
                              f"(supersedes {_cert_key(predecessor)})"),
                    old={"certificate_id": _cert_key(predecessor),
                         "valid_until": predecessor.get("valid_until"),
                         "status": _status_value(predecessor)},
                    new={"valid_until": new.get("valid_until"), "status": _status_value(new)},
                ))
                continue
            holder_norm = str(new.get("holder_norm", "")).strip()
            is_new_holder = bool(holder_norm and holder_norm not in old_holders)
            if not is_new_holder:
                events.append(_event(
                    run_id, scheme, "certificate_new", "low", "explicit", new,
                    detected_at=detected_at, evidence=f"{scheme} accepted snapshot",
                    old={}, new={"status": _status_value(new)},
                ))
                continue

            prior = (prior_coverage or {}).get((scheme, holder_norm)) or {}
            decision = classify_returning(
                scheme=scheme,
                holder_norm=holder_norm,
                valid_from=str(new.get("valid_from") or ""),
                detected_at=detected_iso,
                last_coverage_until=prior.get("last_coverage_until"),
                prior_evidence=list(prior.get("prior_evidence") or []),
            )
            if decision is not None:
                evidence = format_off_scheme_evidence(
                    scheme, decision["gap_days"], decision["last_coverage_until"],
                )
                event = _event(
                    run_id, scheme, "holder_returning", decision["severity"], "inferred", new,
                    detected_at=detected_at,
                    evidence=evidence,
                    old={"last_coverage_until": decision["last_coverage_until"]},
                    new={"status": _status_value(new), "valid_from": str(new.get("valid_from") or "")},
                )
                event["gap_days"] = decision["gap_days"]
                event["prior_evidence"] = decision["prior_evidence"]
                events.append(event)
            else:
                events.append(_event(
                    run_id, scheme, "holder_new", "medium", "explicit", new,
                    detected_at=detected_at, evidence=f"{scheme} accepted snapshot",
                    old={}, new={"status": _status_value(new)},
                ))
            continue
        old = old_by_id[cid]
        old_status = _status_value(old)
        new_status = _status_value(new)
        pair = replacement_by_old.get(cid)
        if pair is not None and (old_status != new_status or _cert_key(pair.new) not in old_by_id):
            # Paired on the flip night, or paired late when the successor arrived a night or
            # two after the withdrawal. A pair whose successor was already in the previous
            # snapshot was reported when it appeared -- never again.
            events.append(_replaced_event(
                run_id, scheme, pair, detected_at=detected_at, old_status=old_status,
            ))
            continue
        if (
            hold_withdrawals
            and new_status == "withdrawn"
            and old_status == new_status
            and cid not in (settled_certificate_ids or set())
        ):
            # Held on the flip night; fire once the window has passed with no successor.
            age = _days_since(new.get("valid_until"), detected_iso)
            if age is not None and window_days <= age <= _HELD_WITHDRAWAL_MAX_AGE_DAYS:
                event_type, severity = _RISKY_STATUSES[new_status]
                events.append(_event(
                    run_id, scheme, event_type, severity, "explicit", new,
                    detected_at=detected_at,
                    evidence=(
                        f"{scheme} explicit status field; no replacement certificate appeared "
                        f"within {window_days} days of the cut-off {new.get('valid_until')}"
                    ),
                    old={"status": old_status}, new={"status": new_status},
                ))
            continue
        if new_status != old_status and new_status in _RISKY_STATUSES:
            if hold_withdrawals and new_status == "withdrawn":
                age = _days_since(new.get("valid_until"), detected_iso)
                if age is not None and age < window_days:
                    continue    # held: the successor may land in the next nightly
            event_type, severity = _RISKY_STATUSES[new_status]
            events.append(_event(
                run_id, scheme, event_type, severity, "explicit", new,
                detected_at=detected_at, evidence=f"{scheme} explicit status field",
                old={"status": old_status}, new={"status": new_status},
            ))
        elif old_status in _RISKY_STATUSES and new_status not in _RISKY_STATUSES:
            # Leaving a risky status is only a reinstatement when the certificate returns to a
            # LIVE status. suspended -> expired is the certificate reaching the end of its term
            # while suspended; labelling that "reinstated" told the reader the opposite of what
            # happened (31 of 31 stored reinstatements were REDcert suspended -> expired).
            reinstated = new_status not in _TERMINAL_STATUSES
            events.append(_event(
                run_id, scheme,
                "certificate_reinstated" if reinstated else "certificate_expired",
                "medium" if reinstated else "low",
                "explicit", new,
                detected_at=detected_at, evidence=f"{scheme} explicit status field",
                old={"status": old_status}, new={"status": new_status},
            ))
        elif old.get("valid_until") and new.get("valid_until") and old.get("valid_until") != new.get("valid_until"):
            events.append(_event(
                run_id, scheme, "certificate_renewed", "low", "explicit", new,
                detected_at=detected_at, evidence=f"{scheme} validity date changed",
                old={"valid_until": old.get("valid_until")}, new={"valid_until": new.get("valid_until")},
            ))
    return events


def build_compliance_reversal_events(

    run_id: str,

    scheme: str,

    old_authority: list[dict] | None,

    new_authority: list[dict] | None,

    valid_certs: list[dict] | None,

    *,

    detected_at=None,

    authority_source: str | None = None,

) -> list[dict]:
    """Reinstatements, read from the compliance register rather than from the valid-only feed.



    ISCC's valid CSV carries no lifecycle status (every row reads `1`), so the reinstatement

    branch in build_change_events — which needs a status field to change — can never fire for it.

    A lifted ISCC suspension therefore left no trace at all: the certificate simply reappeared in

    the valid list and the high-severity suspension stood forever. Diffing the compliance register

    against its previous accepted copy gives the missing signal directly.



    Two deliberate refusals:



    * An empty/missing `new_authority` yields nothing. It is indistinguishable from a failed

      compliance fetch, and treating it as "the register cleared everyone" would reinstate every

      suspended certificate in a single run.

    * A certificate cleared from the register but ABSENT from the valid list yields nothing. That

      shape is ambiguous — a withdrawal that has also aged out looks the same — and the exit is

      already covered by the disappearance branch.

    """
    if not new_authority:
        return []
    old_adverse = {
        _cert_key(row): row
        for row in (old_authority or [])
        if _cert_key(row) and _status_value(row) in _RISKY_STATUSES
    }
    if not old_adverse:
        return []
    still_adverse = {
        _cert_key(row)
        for row in new_authority
        if _cert_key(row) and _status_value(row) in _RISKY_STATUSES
    }
    valid_by_id = {_cert_key(row): row for row in (valid_certs or []) if _cert_key(row)}
    source_label = (authority_source or "").strip() or f"{scheme} status authority"

    events: list[dict] = []
    for cid, prior in sorted(old_adverse.items()):
        if cid in still_adverse or cid not in valid_by_id:
            continue
        prior_status = _status_value(prior)
        # Identity from the live valid row; the episode window only exists on the register row.
        cert_for_event = dict(valid_by_id[cid])
        for key in ("suspended_date", "suspended_until"):
            if prior.get(key):
                cert_for_event[key] = prior.get(key)
        events.append(_event(
            run_id, scheme, "certificate_reinstated", "medium", "explicit", cert_for_event,
            detected_at=detected_at,
            evidence=(f"{source_label} no longer lists this certificate as {prior_status}, "
                      f"and it is back in the accepted snapshot"),
            old={"status": prior_status}, new={"status": ""},
        ))
    return events


_ADVERSE_EVENT_STATUS = {
    event_type: status for status, (event_type, _severity) in _RISKY_STATUSES.items()
}

# Events that say where a certificate's lifecycle stands. Anything else on the log (renewals,
# scope or address changes, soft alerts) neither opens nor closes an adverse episode.
_LIFECYCLE_EVENT_TYPES = frozenset(_ADVERSE_EVENT_STATUS) | {
    "certificate_reinstated", "certificate_expired", "certificate_replaced",
    "certificate_removed_from_valid_list",
}

# Same horizon as a held withdrawal: past it, the log entry may predate reinstatement logging,
# and good news that late is noise rather than a correction.
_ORPHANED_EPISODE_MAX_AGE_DAYS = 60


def build_orphaned_suspension_reinstatements(

    run_id: str,

    scheme: str,

    history_events: list[dict] | None,

    new_authority: list[dict] | None,

    valid_certs: list[dict] | None,

    *,

    detected_at=None,

    authority_source: str | None = None,

    skip_certificate_ids: set[str] | None = None,

) -> list[dict]:
    """Reinstatements the register diff cannot see, read from the append-only change history.



    build_compliance_reversal_events diffs the register against its previous ACCEPTED copy. When

    the nightly fails to publish, that copy freezes while the change log keeps growing — so an

    episode that opens and closes inside the outage is reported as a suspension (every night) and

    never as lifted: the frozen baseline predates it. Shandong Sanju, EU-ISCC-Cert-DE129-35412331,

    suspended 2026-08-31 and lifted 2026-09-14 during the 2026-08-26 outage, left 14 suspension

    events and no reinstatement.



    The history does not freeze. A certificate whose LATEST lifecycle event is adverse, which the

    fresh register no longer lists as adverse, and which sits in the accepted valid list, has been

    reinstated. Self-limiting: the emitted event becomes the latest lifecycle event, so the next

    run is silent. Same two refusals as the register diff — an empty register is a failed fetch,

    and a certificate absent from the valid list is ambiguous.

    """
    if not new_authority:
        return []
    still_adverse = {
        _cert_key(row)
        for row in new_authority
        if _cert_key(row) and _status_value(row) in _RISKY_STATUSES
    }
    valid_by_id = {_cert_key(row): row for row in (valid_certs or []) if _cert_key(row)}
    skip = set(skip_certificate_ids or ())

    latest: dict[str, dict] = {}
    for event in history_events or []:
        if not isinstance(event, dict) or str(event.get("scheme") or "") != scheme:
            continue
        if str(event.get("event_type") or "") not in _LIFECYCLE_EVENT_TYPES:
            continue
        cid = _cert_key(event)
        if not cid:
            continue
        current = latest.get(cid)
        # >= so that, within one run, the later line of the append-only log wins.
        if current is None or str(event.get("detected_at") or "") >= str(current.get("detected_at") or ""):
            latest[cid] = event

    detected_iso = _utc_iso(detected_at)
    source_label = (authority_source or "").strip() or f"{scheme} status authority"
    events: list[dict] = []
    for cid, last in sorted(latest.items()):
        prior_status = _ADVERSE_EVENT_STATUS.get(str(last.get("event_type") or ""))
        if prior_status is None or cid in still_adverse or cid in skip or cid not in valid_by_id:
            continue
        last_seen = str(last.get("detected_at") or "")[:10]
        age = _days_since(last_seen, detected_iso)
        if age is None or age > _ORPHANED_EPISODE_MAX_AGE_DAYS:
            continue
        cert_for_event = dict(valid_by_id[cid])
        # Only the start survives from the log: its `suspended_until` was the SCHEDULED end while
        # the episode ran. The actual end comes from the certificate's own register record.
        cert_for_event.pop("suspended_until", None)
        if prior_status == "suspended" and last.get("suspended_date"):
            cert_for_event["suspended_date"] = last.get("suspended_date")
        events.append(_event(
            run_id, scheme, "certificate_reinstated", "medium", "explicit", cert_for_event,
            detected_at=detected_at,
            evidence=(f"{source_label} no longer lists this certificate as {prior_status} "
                      f"(last reported {prior_status} on {last_seen}), and it is in the accepted "
                      f"snapshot"),
            old={"status": prior_status}, new={"status": ""},
        ))
    return events


def build_register_diff_events(run_id: str, scheme: str, old_certs: list[dict], new_certs: list[dict],

                               diff: Mapping[str, object], *, detected_at=None) -> list[dict]:
    """Change events for registers with a plain added/removed-style diff (LCFS, EPA) --

    no certificate renumbering-on-renewal and no external status authority, unlike

    build_change_events's ISCC/REDcert reconciliation. `diff` is the diff_*_snapshot() output:

    outcome name -> set/list of certificate_id. Recognized outcomes reuse EXISTING event types

    so RED_FLAG_EVENT_TYPES and the digest labels in daily_digest.py need no changes:

    added -> certificate_new (info only), retired/withdrawn -> certificate_withdrawn (explicit

    status flip, red flag; LCFS says retired, Oregon/BC/CFR say withdrawn),

    removed/deregistered -> certificate_removed_from_valid_list (the register

    stopped listing this certificate at all, red flag).

    RenovaBio adds suspended -> certificate_suspended (red flag), terminated -> certificate_revoked_or_terminated (not a red flag), expired -> certificate_expired, renewed pairs -> certificate_renewed on the successor."""
    old_by_id = {_cert_key(c): c for c in (old_certs or []) if _cert_key(c)}
    new_by_id = {_cert_key(c): c for c in (new_certs or []) if _cert_key(c)}
    events: list[dict] = []
    for cid in sorted(diff.get("added") or ()):
        cert = new_by_id.get(cid)
        if cert is None:
            continue
        events.append(_event(
            run_id, scheme, "certificate_new", "low", "explicit", cert,
            detected_at=detected_at, evidence=f"{scheme} accepted snapshot",
            old={}, new={"status": _status_value(cert)},
        ))
    withdrawn_ids = set(diff.get("retired") or ()) | set(diff.get("withdrawn") or ())
    for cid in sorted(withdrawn_ids):
        cert = new_by_id.get(cid) or old_by_id.get(cid)
        if cert is None:
            continue
        events.append(_event(
            run_id, scheme, "certificate_withdrawn", "high", "explicit", cert,
            detected_at=detected_at, evidence=f"{scheme} status field flipped to withdrawn",
            old={"status": _status_value(old_by_id.get(cid, {}))}, new={"status": "withdrawn"},
        ))
    removed_ids = set(diff.get("removed") or ()) | set(diff.get("deregistered") or ())
    for cid in sorted(removed_ids):
        cert = old_by_id.get(cid)
        if cert is None:
            continue
        events.append(_event(
            run_id, scheme, "certificate_removed_from_valid_list", "high", "inferred", cert,
            detected_at=detected_at,
            evidence=f"{scheme} certificate disappeared from accepted snapshot",
            old={"status": _status_value(cert)}, new={},
        ))
    # Registers that publish an explicit lifecycle column per certificate (RenovaBio, spec
    # 2026-09-29). The other registers' diffs never carry these keys.
    for outcome, (event_type, severity) in (
        ("suspended", ("certificate_suspended", "high")),
        ("terminated", ("certificate_revoked_or_terminated", "low")),
        ("expired", ("certificate_expired", "low")),
    ):
        for cid in sorted(diff.get(outcome) or ()):
            cert = new_by_id.get(cid)
            if cert is None:
                continue
            events.append(_event(
                run_id, scheme, event_type, severity, "explicit", cert,
                detected_at=detected_at, evidence=f"{scheme} lists the certificate as {outcome}",
                old={"status": _status_value(old_by_id.get(cid, {}))}, new={"status": outcome},
            ))
    for old_id, new_id in diff.get("renewed") or ():
        new = new_by_id.get(new_id)
        if new is None:
            continue
        predecessor = old_by_id.get(old_id) or new_by_id.get(old_id) or {}
        events.append(_event(
            run_id, scheme, "certificate_renewed", "low", "explicit", new,
            detected_at=detected_at,
            evidence=f"{scheme} names {new_id} as the successor of {old_id}",
            old={"certificate_id": old_id, "valid_until": predecessor.get("valid_until"),
                 "status": _status_value(predecessor)},
            new={"valid_until": new.get("valid_until"), "status": _status_value(new)},
        ))
    return events


def _overall_status(report: dict) -> str:
    if report.get("status"):
        return str(report["status"])
    refresh = report.get("refresh", {})
    if isinstance(refresh, dict) and any(
        isinstance(v, dict) and v.get("status") == "error"
        for v in refresh.values()
    ):
        return "error"
    return "ok"


def _history_compact(report: dict) -> dict:
    refresh = report.get("refresh", {}) if isinstance(report.get("refresh"), dict) else {}
    decisions = {
        k: v.get("decision", "")
        for k, v in refresh.items()
        if isinstance(v, dict) and v.get("decision")
    }
    return {
        "run_id": report.get("started_at", ""),
        "started_at": report.get("started_at", ""),
        "finished_at": report.get("finished_at", ""),
        "duration_s": report.get("duration_s"),
        "status": _overall_status(report),
        "decisions": decisions,
        "geocode_status": (report.get("geocode") or {}).get("status", ""),
        "app_db_status": (report.get("app_db") or {}).get("status", ""),
        "upload_status": (report.get("upload") or {}).get("status", ""),
        # Unfetched/unparsed PDF delta-queue backlog per scheme, carried verbatim from the
        # caller-computed report["queue_depth"] (daily_refresh.py merges pdf_delta + audit_delta
        # remaining_by_scheme). Absent for callers with no PDF delta queue (INS/US-local
        # refreshes) -- {} there, never a KeyError downstream.
        "queue_depth": dict(report.get("queue_depth") or {}),
    }


def append_outcome_history(history_dir: str | Path, report: dict) -> None:
    """Append the finished run's compact history line, without repeating fingerprints.



    A dataset publish records a non-success line before the upload so a crash mid-publish

    is not a success. This second line is the outcome the monitor is allowed to trust.

    """
    append_jsonl(Path(history_dir) / "refresh_history.jsonl", _history_compact(report))


def append_history_files(history_dir: str | Path, report: dict) -> None:
    root = Path(history_dir)
    refresh = report.get("refresh", {}) if isinstance(report.get("refresh"), dict) else {}
    append_jsonl(root / "refresh_history.jsonl", _history_compact(report))

    events: list[dict] = []
    for payload in refresh.values():
        if not isinstance(payload, dict):
            continue
        for record in payload.get("fingerprints", []) or []:
            append_jsonl(root / "snapshot_history.jsonl", record)
        for record in payload.get("compliance_history", []) or []:
            append_jsonl(root / "compliance_history.jsonl", record)
        events.extend(payload.get("change_events", []) or [])
    for event in events:
        append_jsonl(root / "change_events.jsonl", event)

    by_type = Counter(str(e.get("event_type", "")) for e in events if e.get("event_type"))
    by_severity = Counter(str(e.get("severity", "")) for e in events if e.get("severity"))
    summary = {
        "run_id": report.get("started_at", ""),
        "events_total": len(events),
        "events_by_type": dict(sorted(by_type.items())),
        "events_by_severity": dict(sorted(by_severity.items())),
    }
    (root / "latest_change_summary.json").write_text(
        json.dumps(summary, indent=2, sort_keys=True), encoding="utf-8"
    )
    _mirror_events_to_shared_log(root, events)


def _mirror_events_to_shared_log(history_dir: Path, events: list) -> None:
    """Flags and the daily mail read one log. A staging dir keeps its own copy;

    the same event dicts are appended to state_paths.refresh_dir()."""
    import state_paths

    shared = Path(state_paths.refresh_dir())
    if history_dir.resolve() == shared.resolve():
        return
    for event in events:
        if isinstance(event, dict) and event.get("event_type"):
            append_jsonl(shared / "change_events.jsonl", event)