skills/regulatory/scripts/integrity.py

"""Immutable source-lineage checks for the Regulatory skill. Stdlib only."""

from __future__ import annotations

import hashlib
import json
import re
from datetime import date
from pathlib import Path

SCHEMA = "regulatory-integrity/v1"
CONFIRMED = "confirmed"
HISTORICAL_SELECTED = "historical_selected"
UNRESOLVED = "unresolved"
BLOCKED = "blocked"
RELIANCE_STATES = {CONFIRMED, HISTORICAL_SELECTED}
STOP_SEVERITIES = {"superseded", "incomplete"}
EDITORIAL_MARKER = re.compile(r"[►◄▼▲]\s*[A-Z]?\d*")
WHITESPACE = re.compile(r"\s+")
IDENTITY_FIELDS = ("jurisdiction", "profile", "instrument_id")


class IntegrityError(ValueError):
    """The saved evidence chain is incomplete or no longer matches."""


def sha256_bytes(content: bytes) -> str:
    return hashlib.sha256(content).hexdigest()


def sha256_file(path: Path) -> str:
    if not path.is_file():
        raise IntegrityError(f"Required saved file is missing: {path.name}")
    return sha256_bytes(path.read_bytes())


NOT_A_SAVED_FILE = (
    "{label} is not a saved file: {path}\n"
    "Every step of this skill reads and re-reads the same bytes — to hash them, "
    "to\nquote from them, and to re-check them later. A pipe, device or "
    "directory yields\nits contents once or never, so nothing downstream can "
    "confirm what was read.\nSave the text to an ordinary file and pass that."
)


def require_saved_file(path: Path, label: str = "Input") -> Path:
    """Refuse anything that cannot be read twice and hashed.

    A named pipe passes `exists()` and then blocks forever on read, which is
    indistinguishable from a slow script. Bounding it here means every consumer
    fails in the first second with a message that names the remedy.
    """
    if not path.is_file():
        raise IntegrityError(NOT_A_SAVED_FILE.format(label=label, path=path))
    return path


def canonical_text(text: str) -> str:
    return WHITESPACE.sub(" ", EDITORIAL_MARKER.sub(" ", text)).strip()


def canonical_sha256(text: str) -> str:
    return sha256_bytes(canonical_text(text).encode("utf-8"))


def identity(jurisdiction: str, profile: str, instrument_id: str) -> dict[str, str]:
    values = {
        "jurisdiction": jurisdiction.strip().upper(),
        "profile": profile.strip().lower(),
        "instrument_id": instrument_id.strip(),
    }
    missing = [name for name, value in values.items() if not value]
    if missing:
        raise IntegrityError("Instrument identity is missing: " + ", ".join(missing))
    return values


def require_identity(value: object) -> dict[str, str]:
    if not isinstance(value, dict):
        raise IntegrityError("No stable instrument identity is recorded.")
    return identity(*(str(value.get(field, "")) for field in IDENTITY_FIELDS))


DESTINATION_OCCUPIED = (
    "Destination is already occupied: {name}. "
    "Use a new destination so the earlier record remains unchanged."
)


def check_destination_free(destination: Path) -> None:
    """Refuse an occupied destination before any work is done for it."""
    if destination.exists():
        raise IntegrityError(DESTINATION_OCCUPIED.format(name=destination.name))


def reserve_destination(destination: Path) -> None:
    """Claim the destination atomically, immediately before the first write."""
    try:
        destination.mkdir(parents=True, exist_ok=False)
    except FileExistsError as error:
        raise IntegrityError(
            DESTINATION_OCCUPIED.format(name=destination.name)
        ) from error


def fetch_integrity(source_bytes: bytes, instrument_identity: dict[str, str]) -> dict:
    return {
        "schema": SCHEMA,
        "instrument": require_identity(instrument_identity),
        "publisher_bytes_sha256": sha256_bytes(source_bytes),
    }


def _read_json(path: Path, label: str) -> dict:
    if not path.is_file():
        raise IntegrityError(f"Missing {label}: {path.name}")
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, json.JSONDecodeError) as error:
        raise IntegrityError(f"Unreadable {label}: {path.name}: {error}") from error
    if not isinstance(value, dict):
        raise IntegrityError(f"Invalid {label}: {path.name}")
    return value


def validate_fetch(directory: Path) -> tuple[dict, Path, dict[str, str]]:
    fetch = _read_json(directory / "fetch.json", "fetch receipt")
    record = fetch.get("integrity")
    if not isinstance(record, dict) or record.get("schema") != SCHEMA:
        raise IntegrityError("The fetch receipt has no supported integrity record.")
    instrument_identity = require_identity(record.get("instrument"))
    saved_name = fetch.get("saved_as")
    if not isinstance(saved_name, str) or Path(saved_name).name != saved_name:
        raise IntegrityError("The fetch receipt has an invalid saved source name.")
    source_path = directory / saved_name
    actual = sha256_file(source_path)
    expected = record.get("publisher_bytes_sha256")
    if actual != expected or fetch.get("sha256") != expected:
        raise IntegrityError("Saved publisher text changed after retrieval.")
    return fetch, source_path, instrument_identity


def validate_extraction_input(directory: Path, instrument: Path) -> dict:
    fetch, publisher_path, instrument_identity = validate_fetch(directory)
    input_hash = sha256_file(instrument)
    publisher_hash = fetch["integrity"]["publisher_bytes_sha256"]
    if instrument.resolve() == publisher_path.resolve():
        transformation = {"kind": "direct", "input_name": instrument.name}
    else:
        receipt = _read_json(
            directory / "transformation.json", "transformation receipt"
        )
        if receipt.get("schema") != SCHEMA:
            raise IntegrityError(
                "The transformation receipt uses an unsupported schema."
            )
        if receipt.get("publisher_bytes_sha256") != publisher_hash:
            raise IntegrityError(
                "The transformation receipt names different publisher bytes."
            )
        if receipt.get("derived_text_sha256") != input_hash:
            raise IntegrityError("Derived text changed after transformation.")
        if receipt.get("derived_name") != instrument.name:
            raise IntegrityError(
                "The transformation receipt names a different derived file."
            )
        transformation = {
            "kind": "derived",
            "input_name": instrument.name,
            "receipt": receipt,
        }
    return {
        "schema": SCHEMA,
        "instrument": instrument_identity,
        "publisher_name": publisher_path.name,
        "publisher_bytes_sha256": publisher_hash,
        "extraction_input_sha256": input_hash,
        "transformation": transformation,
    }


def version_state(
    findings: list[dict], historical_effective: str | None, research_date: str | None
) -> str:
    if bool(historical_effective) != bool(research_date):
        raise IntegrityError("Historical selection requires both dates.")
    if historical_effective and research_date:
        try:
            effective, research = (
                date.fromisoformat(historical_effective),
                date.fromisoformat(research_date),
            )
        except ValueError as error:
            raise IntegrityError("Historical dates must be ISO YYYY-MM-DD.") from error
        if effective > research:
            raise IntegrityError(
                "Historical effective date is after the research date."
            )
    if any(item.get("severity") == "incomplete" for item in findings):
        return BLOCKED
    if not findings:
        return UNRESOLVED
    if historical_effective and research_date:
        return HISTORICAL_SELECTED
    if any(item.get("severity") == "superseded" for item in findings):
        return BLOCKED
    # Furniture/disclaimers do not identify the selected version.
    incidental = {
        "revised-may-not-be-current",
        "prospective-provisions",
        "ecfr-unofficial",
    }
    if all(item.get("marker") in incidental for item in findings):
        return UNRESOLVED
    return CONFIRMED


def validate_version(directory: Path, chain: dict) -> dict:
    version = _read_json(directory / "version.json", "version receipt")
    record = version.get("integrity")
    if not isinstance(record, dict) or record.get("schema") != SCHEMA:
        raise IntegrityError("The version receipt has no supported integrity record.")
    if require_identity(record.get("instrument")) != chain["instrument"]:
        raise IntegrityError("The version receipt belongs to a different instrument.")
    for field in ("publisher_bytes_sha256", "extraction_input_sha256"):
        if record.get(field) != chain[field]:
            raise IntegrityError(
                "The version receipt is bound to different source text."
            )
    if version.get("state") not in RELIANCE_STATES:
        raise IntegrityError(
            "The operative version is unresolved; downstream reliance is blocked."
        )
    expected_state = version_state(
        version.get("findings") or [],
        version.get("effective_date"),
        version.get("research_date"),
    )
    if expected_state != version["state"]:
        raise IntegrityError(
            "Version state conflicts with its findings or historical dates."
        )
    return version


def provision_set_sha256(provisions: list[dict]) -> str:
    records = []
    for provision in provisions:
        if "text" in provision and canonical_sha256(provision["text"]) != provision.get(
            "sha256"
        ):
            raise IntegrityError(
                "Extracted text changed for "
                f"{provision.get('id', 'unknown provision')}."
            )
        records.append(
            {
                key: provision.get(key)
                for key in ("id", "label", "heading", "class", "sha256")
            }
        )
    encoded = json.dumps(records, sort_keys=True, separators=(",", ":")).encode("utf-8")
    return sha256_bytes(encoded)


def extraction_integrity(chain: dict, version: dict, provisions: list[dict]) -> dict:
    return {
        **chain,
        "version_state": version["state"],
        "version_effective_date": version.get("effective_date"),
        "research_date": version.get("research_date"),
        "provision_set_sha256": provision_set_sha256(provisions),
    }


def validate_extraction(payload: dict, directory: Path) -> dict:
    record = payload.get("integrity")
    if not isinstance(record, dict) or record.get("schema") != SCHEMA:
        raise IntegrityError("The extraction has no supported integrity record.")
    input_name = record.get("transformation", {}).get("input_name")
    if not isinstance(input_name, str) or Path(input_name).name != input_name:
        raise IntegrityError("The extraction records an invalid source name.")
    chain = validate_extraction_input(directory, directory / input_name)
    version = validate_version(directory, chain)
    expected = extraction_integrity(chain, version, payload.get("provisions") or [])
    if record != expected:
        raise IntegrityError(
            "The extraction no longer matches its source and version chain."
        )
    return record


def require_same_instrument(before: dict, after: dict) -> None:
    if require_identity(before.get("instrument")) != require_identity(
        after.get("instrument")
    ):
        raise IntegrityError(
            "These records identify different instruments; no comparison was made."
        )