Downloads and Reproducibility

#!/usr/bin/env python3
"""Provide strict JSON, hashing and primitive validation helpers.

Purpose: give every evaluator component the same fail-closed rules for
canonical data, finite numbers, identifiers and timestamps before evidence is
measured, signed or compared.

Author: Sergei Sychev
Website: www.triz-ri.pro
"""

from __future__ import annotations

import hashlib
import json
import math
import pathlib
import re
from datetime import datetime, timezone
from typing import Any


SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
UTC_RE = re.compile(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$")


class IntegrityError(ValueError):
    """Report data that cannot safely enter the evaluator's trust boundary.

    Purpose: distinguish integrity failures from ordinary runtime errors so
    callers can reject malformed or ambiguous evidence deterministically.
    """


def canonical_json_bytes(value: Any) -> bytes:
    """Serialise a value into the evaluator's canonical UTF-8 JSON form.

    Purpose: ensure that equivalent evidence always produces the same bytes
    before hashing or signature verification.
    """
    try:
        text = json.dumps(
            value,
            ensure_ascii=False,
            allow_nan=False,
            sort_keys=True,
            separators=(",", ":"),
        )
    except (TypeError, ValueError) as exc:
        raise IntegrityError(f"value is not canonical JSON: {exc}") from exc
    return text.encode("utf-8")


def canonical_sha256(value: Any) -> str:
    """Return the SHA-256 digest of a value's canonical JSON bytes.

    Purpose: bind policies, observations and events to content rather than to
    filenames or formatting choices.
    """
    return hashlib.sha256(canonical_json_bytes(value)).hexdigest()


def file_sha256(path: pathlib.Path) -> str:
    """Calculate the SHA-256 digest of a file without loading it all at once.

    Purpose: verify large package artifacts with bounded memory use and the
    same digest algorithm used by publication manifests.
    """
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def load_json_strict(path: pathlib.Path) -> Any:
    """Load JSON while rejecting non-standard non-finite numeric constants.

    Purpose: prevent NaN or infinity from entering calculations through parser
    extensions that ordinary JSON does not permit.
    """

    def reject_constant(token: str) -> None:
        """Reject a non-finite token accepted by Python's permissive parser.

        Purpose: keep every loaded number inside the evaluator's finite,
        portable JSON domain.
        """
        raise IntegrityError(f"non-finite JSON number is not allowed: {token}")

    try:
        return json.loads(
            path.read_text(encoding="utf-8"),
            parse_constant=reject_constant,
        )
    except (OSError, json.JSONDecodeError) as exc:
        raise IntegrityError(f"cannot read strict JSON from {path}: {exc}") from exc


def write_json_strict(path: pathlib.Path, value: Any) -> None:
    """Write readable UTF-8 JSON while refusing non-finite numbers.

    Purpose: ensure generated evidence remains standards-compliant and can be
    reproduced by independent JSON implementations.
    """
    try:
        text = json.dumps(
            value,
            ensure_ascii=False,
            allow_nan=False,
            indent=2,
        )
    except (TypeError, ValueError) as exc:
        raise IntegrityError(f"cannot write strict JSON to {path}: {exc}") from exc
    path.write_text(text + "\n", encoding="utf-8")


def require_object(value: Any, label: str) -> dict[str, Any]:
    """Return a value only when it is a JSON object.

    Purpose: stop type confusion before field-level validation begins.
    """
    if not isinstance(value, dict):
        raise IntegrityError(f"{label} must be an object")
    return value


def require_array(value: Any, label: str) -> list[Any]:
    """Return a value only when it is a JSON array.

    Purpose: make collection cardinality and item validation unambiguous.
    """
    if not isinstance(value, list):
        raise IntegrityError(f"{label} must be an array")
    return value


def require_exact_keys(
    value: dict[str, Any],
    required: set[str],
    label: str,
) -> None:
    """Require an object to contain exactly the declared field names.

    Purpose: reject missing controls and unrecognised fields instead of
    silently accepting a weaker or differently interpreted record.
    """
    actual = set(value)
    missing = sorted(required - actual)
    unknown = sorted(actual - required)
    if missing or unknown:
        details = []
        if missing:
            details.append("missing=" + ",".join(missing))
        if unknown:
            details.append("unknown=" + ",".join(unknown))
        raise IntegrityError(f"{label} has invalid fields: {'; '.join(details)}")


def require_string(value: Any, label: str) -> str:
    """Return a value only when it is a non-empty string.

    Purpose: prevent absent identifiers and labels from being treated as valid
    bindings between evaluator artifacts.
    """
    if not isinstance(value, str) or not value:
        raise IntegrityError(f"{label} must be a non-empty string")
    return value


def require_boolean(value: Any, label: str) -> bool:
    """Return a value only when it is a literal JSON boolean.

    Purpose: prevent truthy numbers or strings from authorising gates that
    require an explicit true-or-false decision.
    """
    if type(value) is not bool:
        raise IntegrityError(f"{label} must be a boolean")
    return value


def require_integer(
    value: Any,
    label: str,
    *,
    minimum: int | None = None,
) -> int:
    """Validate an integer and an optional inclusive lower bound.

    Purpose: protect counters, sequence numbers and time budgets from boolean,
    fractional or out-of-domain values.
    """
    if type(value) is not int:
        raise IntegrityError(f"{label} must be an integer")
    if minimum is not None and value < minimum:
        raise IntegrityError(f"{label} must be at least {minimum}")
    return value


def require_number(
    value: Any,
    label: str,
    *,
    minimum: float | None = None,
    maximum: float | None = None,
) -> float:
    """Validate a finite JSON number within optional inclusive bounds.

    Purpose: keep calculations inside the metric domain fixed by policy and
    prevent NaN, infinity and unchecked magnitude inflation.
    """
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise IntegrityError(f"{label} must be a JSON number")
    result = float(value)
    if not math.isfinite(result):
        raise IntegrityError(f"{label} must be finite")
    if minimum is not None and result < minimum:
        raise IntegrityError(f"{label} must be at least {minimum}")
    if maximum is not None and result > maximum:
        raise IntegrityError(f"{label} must be at most {maximum}")
    return result


def require_sha256(value: Any, label: str) -> str:
    """Validate a lowercase hexadecimal SHA-256 digest.

    Purpose: ensure artifact bindings have one unambiguous textual form.
    """
    result = require_string(value, label)
    if not SHA256_RE.fullmatch(result):
        raise IntegrityError(f"{label} must be a lowercase SHA-256 digest")
    return result


def require_utc_timestamp(value: Any, label: str) -> datetime:
    """Parse a second-resolution UTC timestamp in the frozen wire format.

    Purpose: make signed observation intervals comparable without local
    timezone, locale or formatting ambiguity.
    """
    result = require_string(value, label)
    if not UTC_RE.fullmatch(result):
        raise IntegrityError(f"{label} must use YYYY-MM-DDTHH:MM:SSZ")
    try:
        parsed = datetime.strptime(result, "%Y-%m-%dT%H:%M:%SZ")
    except ValueError as exc:
        raise IntegrityError(f"{label} is not a valid UTC timestamp") from exc
    return parsed.replace(tzinfo=timezone.utc)

#!/usr/bin/env python3
"""Verify canonical JSON evidence with a frozen Minisign public key.

Purpose: ensure that authorization records used by the evaluator came from the
declared authority and were not changed after signing.

Author: Sergei Sychev
Website: www.triz-ri.pro
"""

from __future__ import annotations

import base64
import binascii
import pathlib
import subprocess
import tempfile
from typing import Any

from integrity import IntegrityError, canonical_json_bytes, require_string


def validate_minisign_public_key(value: Any, label: str) -> str:
    """Validate the encoded shape of a Minisign Ed25519 public key.

    Purpose: reject malformed or substituted key material before invoking the
    external verifier.
    """
    public_key = require_string(value, label)
    try:
        decoded = base64.b64decode(public_key, validate=True)
    except (binascii.Error, ValueError) as exc:
        raise IntegrityError(f"{label} must be a base64 Minisign public key") from exc
    if len(decoded) != 42 or decoded[:2] not in {b"Ed", b"ED"}:
        raise IntegrityError(f"{label} is not a valid Minisign Ed25519 public key")
    return public_key


def verify_minisign_json(
    value: Any,
    signature_path: pathlib.Path,
    public_key: str,
) -> None:
    """Verify a detached signature over a value's canonical JSON bytes.

    Purpose: authenticate authorization evidence independently of its original
    whitespace or object-key ordering while failing closed if Minisign is
    unavailable, times out or rejects the signature.
    """
    public_key = validate_minisign_public_key(public_key, "Minisign public key")
    try:
        signature = signature_path.read_text(encoding="utf-8")
    except OSError as exc:
        raise IntegrityError(f"cannot read Minisign signature: {exc}") from exc
    if not signature.strip():
        raise IntegrityError("Minisign signature must not be empty")

    # Verification uses a temporary canonical representation so signatures bind
    # to data semantics, not to the formatting of the supplied JSON file.
    with tempfile.TemporaryDirectory(prefix="triz-ri-minisign-") as directory:
        canonical_path = pathlib.Path(directory) / "canonical.json"
        canonical_path.write_bytes(canonical_json_bytes(value))
        try:
            completed = subprocess.run(
                [
                    "minisign",
                    "-V",
                    "-q",
                    "-P",
                    public_key,
                    "-m",
                    str(canonical_path),
                    "-x",
                    str(signature_path),
                ],
                check=False,
                capture_output=True,
                text=True,
                timeout=10,
            )
        except FileNotFoundError as exc:
            raise IntegrityError(
                "Minisign 0.12 or later is required to verify authorization"
            ) from exc
        except subprocess.TimeoutExpired as exc:
            raise IntegrityError("Minisign verification timed out") from exc
    if completed.returncode != 0:
        details = (completed.stderr or completed.stdout).strip()
        raise IntegrityError(
            "authorization signature verification failed"
            + (f": {details}" if details else "")
        )

#!/usr/bin/env python3
"""Calculate a bounded result-based signal from frozen policy and evidence.

Purpose: reward verified task performance only after all mandatory result,
quality and external gates have passed, while keeping allocation release under
the separate escrow protocol.

Author: Sergei Sychev
Website: www.triz-ri.pro
"""

from __future__ import annotations

import json
import math
import pathlib
import sys
from typing import Any

from integrity import (
    IntegrityError,
    canonical_sha256,
    load_json_strict,
    require_array,
    require_boolean,
    require_exact_keys,
    require_number,
    require_object,
    require_sha256,
    require_string,
)
from minisign_verifier import validate_minisign_public_key


POLICY_KEYS = {
    "schema_version",
    "policy_id",
    "pilot_id",
    "protocol_version",
    "base_unit",
    "combination_method",
    "result_blocks",
    "quality",
    "required_external_checks",
    "escrow",
}
OBSERVATION_KEYS = {
    "schema_version",
    "run_id",
    "attempt_id",
    "lineage_id",
    "pilot_id",
    "policy_sha256",
    "source_hashes",
    "result_values",
    "quality_value",
    "critical_violations",
    "external_checks",
}
BLOCK_KEYS = {"name", "minimum", "maximum", "threshold", "required"}
QUALITY_KEYS = {"minimum", "maximum", "threshold"}
ESCROW_KEYS = {"preliminary_fraction", "release_requirements", "authorization"}
RELEASE_REQUIREMENT_KEYS = {
    "minimum_observation_seconds",
    "minimum_independent_control_cycles",
    "require_source_complete",
    "require_late_evidence_clear",
}
AUTHORIZATION_KEYS = {
    "authorizer_id",
    "minisign_public_key",
    "trusted_time_sources",
}
SOURCE_HASH_KEYS = {
    "candidate",
    "scorer_output",
    "critical_violations",
    "external_checks",
}


def validate_policy(policy_value: Any) -> dict[str, Any]:
    """Validate the complete frozen signal policy and return it unchanged.

    Purpose: ensure that metric domains, thresholds, required checks and escrow
    authority are fixed and internally consistent before any result is scored.
    """
    policy = require_object(policy_value, "policy")
    require_exact_keys(policy, POLICY_KEYS, "policy")
    if policy["schema_version"] != "signal-policy-0.9":
        raise IntegrityError("policy.schema_version must be signal-policy-0.9")
    require_string(policy["policy_id"], "policy.policy_id")
    require_string(policy["pilot_id"], "policy.pilot_id")
    require_string(policy["protocol_version"], "policy.protocol_version")
    require_number(policy["base_unit"], "policy.base_unit", minimum=0.0)
    if policy["combination_method"] != "product":
        raise IntegrityError("policy.combination_method must be product")

    blocks = require_array(policy["result_blocks"], "policy.result_blocks")
    if not blocks:
        raise IntegrityError("policy.result_blocks must not be empty")
    names: set[str] = set()
    for index, block_value in enumerate(blocks):
        block = require_object(block_value, f"policy.result_blocks[{index}]")
        require_exact_keys(block, BLOCK_KEYS, f"policy.result_blocks[{index}]")
        name = require_string(block["name"], f"policy.result_blocks[{index}].name")
        if name in names:
            raise IntegrityError(f"duplicate policy result block: {name}")
        names.add(name)
        minimum = require_number(block["minimum"], f"policy.result_blocks[{index}].minimum")
        maximum = require_number(block["maximum"], f"policy.result_blocks[{index}].maximum")
        threshold = require_number(block["threshold"], f"policy.result_blocks[{index}].threshold")
        if minimum > maximum or not minimum <= threshold <= maximum:
            raise IntegrityError(f"invalid range for policy result block: {name}")
        if minimum < 0.0 or maximum > 1.0:
            raise IntegrityError(
                f"product result block must use a normalised 0-to-1 domain: {name}"
            )
        require_boolean(block["required"], f"policy.result_blocks[{index}].required")

    quality = require_object(policy["quality"], "policy.quality")
    require_exact_keys(quality, QUALITY_KEYS, "policy.quality")
    q_minimum = require_number(quality["minimum"], "policy.quality.minimum")
    q_maximum = require_number(quality["maximum"], "policy.quality.maximum")
    q_threshold = require_number(quality["threshold"], "policy.quality.threshold")
    if q_minimum > q_maximum or not q_minimum <= q_threshold <= q_maximum:
        raise IntegrityError("invalid policy.quality range")

    checks = require_array(
        policy["required_external_checks"],
        "policy.required_external_checks",
    )
    check_names = [
        require_string(value, "policy.required_external_checks item")
        for value in checks
    ]
    if not check_names or len(check_names) != len(set(check_names)):
        raise IntegrityError("policy.required_external_checks must be non-empty and unique")

    escrow = require_object(policy["escrow"], "policy.escrow")
    require_exact_keys(escrow, ESCROW_KEYS, "policy.escrow")
    require_number(
        escrow["preliminary_fraction"],
        "policy.escrow.preliminary_fraction",
        minimum=0.0,
        maximum=1.0,
    )
    release = require_object(
        escrow["release_requirements"],
        "policy.escrow.release_requirements",
    )
    require_exact_keys(
        release,
        RELEASE_REQUIREMENT_KEYS,
        "policy.escrow.release_requirements",
    )
    if (
        type(release["minimum_observation_seconds"]) is not int
        or release["minimum_observation_seconds"] < 0
    ):
        raise IntegrityError("minimum_observation_seconds must be a non-negative integer")
    if (
        type(release["minimum_independent_control_cycles"]) is not int
        or release["minimum_independent_control_cycles"] < 1
    ):
        raise IntegrityError("minimum_independent_control_cycles must be a positive integer")
    require_boolean(release["require_source_complete"], "require_source_complete")
    require_boolean(release["require_late_evidence_clear"], "require_late_evidence_clear")
    authorization = require_object(escrow["authorization"], "policy.escrow.authorization")
    require_exact_keys(
        authorization,
        AUTHORIZATION_KEYS,
        "policy.escrow.authorization",
    )
    require_string(
        authorization["authorizer_id"],
        "policy.escrow.authorization.authorizer_id",
    )
    validate_minisign_public_key(
        authorization["minisign_public_key"],
        "policy.escrow.authorization.minisign_public_key",
    )
    trusted_sources = require_array(
        authorization["trusted_time_sources"],
        "policy.escrow.authorization.trusted_time_sources",
    )
    source_names = [
        require_string(
            value,
            "policy.escrow.authorization.trusted_time_sources item",
        )
        for value in trusted_sources
    ]
    if not source_names or len(source_names) != len(set(source_names)):
        raise IntegrityError(
            "policy.escrow.authorization.trusted_time_sources "
            "must be non-empty and unique"
        )
    return policy


def validate_observations(
    observation_value: Any,
    policy: dict[str, Any],
) -> dict[str, Any]:
    """Validate observations against the exact fields and domains in policy.

    Purpose: prevent a candidate from inventing, omitting or rescaling the
    measurements and external checks used to calculate its signal.
    """
    observations = require_object(observation_value, "observations")
    require_exact_keys(observations, OBSERVATION_KEYS, "observations")
    if observations["schema_version"] != "signal-observations-0.9":
        raise IntegrityError("observations.schema_version must be signal-observations-0.9")
    for field in ("run_id", "attempt_id", "lineage_id", "pilot_id"):
        require_string(observations[field], f"observations.{field}")
    if observations["pilot_id"] != policy["pilot_id"]:
        raise IntegrityError("observations.pilot_id does not match policy.pilot_id")
    expected_policy_hash = canonical_sha256(policy)
    if (
        require_sha256(
            observations["policy_sha256"],
            "observations.policy_sha256",
        )
        != expected_policy_hash
    ):
        raise IntegrityError("observations.policy_sha256 does not match the frozen policy")

    source_hashes = require_object(
        observations["source_hashes"],
        "observations.source_hashes",
    )
    require_exact_keys(source_hashes, SOURCE_HASH_KEYS, "observations.source_hashes")
    for name, digest in source_hashes.items():
        require_sha256(digest, f"observations.source_hashes.{name}")

    expected_blocks = {block["name"]: block for block in policy["result_blocks"]}
    values = require_object(observations["result_values"], "observations.result_values")
    require_exact_keys(values, set(expected_blocks), "observations.result_values")
    validated_values: dict[str, float] = {}
    for name, block in expected_blocks.items():
        validated_values[name] = require_number(
            values[name],
            f"observations.result_values.{name}",
            minimum=float(block["minimum"]),
            maximum=float(block["maximum"]),
        )

    quality = policy["quality"]
    require_number(
        observations["quality_value"],
        "observations.quality_value",
        minimum=float(quality["minimum"]),
        maximum=float(quality["maximum"]),
    )
    violations = require_array(
        observations["critical_violations"],
        "observations.critical_violations",
    )
    for index, violation in enumerate(violations):
        require_string(violation, f"observations.critical_violations[{index}]")
    if len(violations) != len(set(violations)):
        raise IntegrityError("observations.critical_violations contains duplicates")

    checks = require_object(
        observations["external_checks"],
        "observations.external_checks",
    )
    require_exact_keys(
        checks,
        set(policy["required_external_checks"]),
        "observations.external_checks",
    )
    for name, value in checks.items():
        require_boolean(value, f"observations.external_checks.{name}")
    return observations


def calculate(policy_value: Any, observation_value: Any) -> dict[str, Any]:
    """Calculate eligibility, performance and provisional signal allocations.

    Purpose: turn verified measurements into a deterministic result-based
    signal without allowing an eligible-looking metric to bypass mandatory
    quality, result or external gates.
    """
    policy = validate_policy(policy_value)
    observations = validate_observations(observation_value, policy)
    blocks = {block["name"]: block for block in policy["result_blocks"]}
    result_values = {
        name: float(observations["result_values"][name])
        for name in blocks
    }

    # Gates are evaluated before multiplication. A high product cannot
    # compensate for a mandatory failure.
    reasons: list[str] = []
    if observations["critical_violations"]:
        reasons.append("critical_violation")
    if float(observations["quality_value"]) < float(policy["quality"]["threshold"]):
        reasons.append("quality_gate_failed")
    for name, block in blocks.items():
        if block["required"] and result_values[name] < float(block["threshold"]):
            reasons.append("result_threshold_failed:" + name)
    incomplete = sorted(
        name
        for name in policy["required_external_checks"]
        if observations["external_checks"][name] is not True
    )
    if incomplete:
        reasons.append("external_checks_incomplete:" + ",".join(incomplete))

    # Every component has already been restricted to its frozen 0-to-1 domain,
    # so multiplication cannot create an inflated or non-finite score.
    perf_overall = math.prod(result_values[name] for name in blocks)
    if not math.isfinite(perf_overall):
        raise IntegrityError("calculated Perf_overall is not finite")
    eligible = not reasons
    base_unit = float(policy["base_unit"])
    gross = base_unit + base_unit * perf_overall if eligible else 0.0
    if not math.isfinite(gross):
        raise IntegrityError("calculated gross signal is not finite")
    preliminary_fraction = float(policy["escrow"]["preliminary_fraction"])
    preliminary = gross * preliminary_fraction

    return {
        "schema_version": "signal-output-0.9",
        "formula_model": (
            "mandatory quality and external gates; "
            "S=B+B*product(Perf_i); release is controlled separately"
        ),
        "policy_id": policy["policy_id"],
        "policy_sha256": canonical_sha256(policy),
        "observations_sha256": canonical_sha256(observations),
        "run_id": observations["run_id"],
        "attempt_id": observations["attempt_id"],
        "lineage_id": observations["lineage_id"],
        "pilot_id": observations["pilot_id"],
        "source_hashes": observations["source_hashes"],
        "base_unit": base_unit,
        "Perf_overall": perf_overall,
        "eligible": eligible,
        "blocking_reasons": reasons,
        "gross_signal": gross,
        "preliminary_allocation": preliminary,
        "held_allocation": gross - preliminary,
        "released_signal": 0.0,
        "final_signal": None,
        "escrow_state": "pending",
    }


def main(argv: list[str]) -> int:
    """Run the calculator as a command-line JSON transformer.

    Purpose: provide a reproducible entry point that emits either the full
    signal record or a machine-readable fail-closed validation result.
    """
    if len(argv) != 3:
        raise SystemExit(
            "usage: signal_calculator.py signal_policy.json signal_observations.json"
        )
    try:
        policy = load_json_strict(pathlib.Path(argv[1]))
        observations = load_json_strict(pathlib.Path(argv[2]))
        result = calculate(policy, observations)
        print(json.dumps(result, ensure_ascii=False, allow_nan=False, indent=2))
    except IntegrityError as exc:
        print(
            json.dumps(
                {
                    "schema_version": "signal-output-0.9",
                    "eligible": False,
                    "validation_error": str(exc),
                },
                ensure_ascii=False,
                allow_nan=False,
                indent=2,
            )
        )
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))

#!/usr/bin/env python3
"""Control provisional and final signal release through signed evidence.

Purpose: keep a calculated reward from becoming final until an authorised,
cryptographically bound record proves that the required observation period,
independent controls and late-evidence review have been completed.

Author: Sergei Sychev
Website: www.triz-ri.pro
"""

from __future__ import annotations

import json
import pathlib
import sys
from typing import Any

from integrity import (
    IntegrityError,
    canonical_sha256,
    load_json_strict,
    require_array,
    require_boolean,
    require_exact_keys,
    require_number,
    require_object,
    require_sha256,
    require_string,
    require_utc_timestamp,
)
from minisign_verifier import verify_minisign_json
from signal_calculator import calculate, validate_policy


EVENT_KEYS = {
    "schema_version",
    "event_id",
    "event_type",
    "authorization_evidence_sha256",
    "policy_sha256",
    "observations_sha256",
    "signal_output_sha256",
    "run_id",
    "attempt_id",
    "lineage_id",
    "pilot_id",
    "previous_ledger_sha256",
}
AUTHORIZATION_EVIDENCE_KEYS = {
    "schema_version",
    "event_id",
    "event_type",
    "effective_at_utc",
    "authorizer_id",
    "trusted_time_source",
    "policy_sha256",
    "observations_sha256",
    "signal_output_sha256",
    "run_id",
    "attempt_id",
    "lineage_id",
    "pilot_id",
    "previous_ledger_sha256",
    "observation_started_at_utc",
    "observation_completed_at_utc",
    "independent_control_cycle_ids",
    "source_complete",
    "late_evidence_clear",
}
LEDGER_KEYS = {
    "schema_version",
    "policy_sha256",
    "observations_sha256",
    "signal_output_sha256",
    "run_id",
    "attempt_id",
    "lineage_id",
    "pilot_id",
    "gross_signal",
    "preliminary_allocation",
    "held_allocation",
    "released_signal",
    "final_signal",
    "escrow_state",
    "applied_event_ids",
}
SIGNAL_KEYS = {
    "schema_version",
    "formula_model",
    "policy_id",
    "policy_sha256",
    "observations_sha256",
    "run_id",
    "attempt_id",
    "lineage_id",
    "pilot_id",
    "source_hashes",
    "base_unit",
    "Perf_overall",
    "gross_signal",
    "preliminary_allocation",
    "held_allocation",
    "released_signal",
    "final_signal",
    "escrow_state",
    "eligible",
    "blocking_reasons",
}
EVENT_BINDING_FIELDS = (
    "event_id",
    "event_type",
    "policy_sha256",
    "observations_sha256",
    "signal_output_sha256",
    "run_id",
    "attempt_id",
    "lineage_id",
    "pilot_id",
    "previous_ledger_sha256",
)


def validate_signal_output(
    policy_value: Any,
    observation_value: Any,
    signal_value: Any,
) -> tuple[dict[str, Any], dict[str, Any]]:
    """Recalculate and validate a supplied signal output.

    Purpose: prevent escrow from trusting a signal file that was altered,
    generated from different observations or calculated under another policy.
    """
    policy = validate_policy(policy_value)
    signal = require_object(signal_value, "signal output")
    require_exact_keys(signal, SIGNAL_KEYS, "signal output")
    expected_signal = calculate(policy, observation_value)
    if canonical_sha256(signal) != canonical_sha256(expected_signal):
        raise IntegrityError(
            "signal output does not match a fresh calculation from policy and observations"
        )
    if signal["schema_version"] != "signal-output-0.9":
        raise IntegrityError("signal output schema_version must be signal-output-0.9")
    if signal["policy_sha256"] != canonical_sha256(policy):
        raise IntegrityError("signal output does not match policy")
    require_sha256(signal["observations_sha256"], "signal output.observations_sha256")
    for field in ("run_id", "attempt_id", "lineage_id", "pilot_id"):
        require_string(signal[field], f"signal output.{field}")
    if signal["pilot_id"] != policy["pilot_id"]:
        raise IntegrityError("signal output pilot_id does not match policy")
    eligible = require_boolean(signal["eligible"], "signal output.eligible")
    gross = require_number(signal["gross_signal"], "signal output.gross_signal", minimum=0.0)
    preliminary = require_number(
        signal["preliminary_allocation"],
        "signal output.preliminary_allocation",
        minimum=0.0,
    )
    held = require_number(signal["held_allocation"], "signal output.held_allocation", minimum=0.0)
    if abs((preliminary + held) - gross) > 1e-12:
        raise IntegrityError("signal output allocations do not equal gross signal")
    if not eligible and gross != 0.0:
        raise IntegrityError("ineligible signal output must have zero gross signal")
    return policy, signal


def initialize_ledger(
    policy_value: Any,
    observation_value: Any,
    signal_value: Any,
) -> dict[str, Any]:
    """Create the initial pending ledger from a freshly validated signal.

    Purpose: bind the escrow balance and release state to the exact policy,
    observations, attempt and calculated signal before any event can act on it.
    """
    _, signal = validate_signal_output(
        policy_value,
        observation_value,
        signal_value,
    )
    gross = float(signal["gross_signal"])
    preliminary = float(signal["preliminary_allocation"])
    held = float(signal["held_allocation"])
    return {
        "schema_version": "escrow-ledger-0.9",
        "policy_sha256": signal["policy_sha256"],
        "observations_sha256": signal["observations_sha256"],
        "signal_output_sha256": canonical_sha256(signal),
        "run_id": signal["run_id"],
        "attempt_id": signal["attempt_id"],
        "lineage_id": signal["lineage_id"],
        "pilot_id": signal["pilot_id"],
        "gross_signal": gross,
        "preliminary_allocation": preliminary,
        "held_allocation": held,
        "released_signal": 0.0,
        "final_signal": None,
        "escrow_state": "pending",
        "applied_event_ids": [],
    }


def validate_ledger(ledger_value: Any) -> dict[str, Any]:
    """Validate the complete structure and numeric state of an escrow ledger.

    Purpose: reject corrupted, ambiguous or replay-prone state before applying
    a new release or forfeiture event.
    """
    ledger = require_object(ledger_value, "escrow ledger")
    require_exact_keys(ledger, LEDGER_KEYS, "escrow ledger")
    if ledger["schema_version"] != "escrow-ledger-0.9":
        raise IntegrityError("escrow ledger schema_version must be escrow-ledger-0.9")
    for field in (
        "policy_sha256",
        "observations_sha256",
        "signal_output_sha256",
    ):
        require_sha256(ledger[field], f"escrow ledger.{field}")
    for field in ("run_id", "attempt_id", "lineage_id", "pilot_id"):
        require_string(ledger[field], f"escrow ledger.{field}")
    for field in (
        "gross_signal",
        "preliminary_allocation",
        "held_allocation",
        "released_signal",
    ):
        require_number(ledger[field], f"escrow ledger.{field}", minimum=0.0)
    if ledger["final_signal"] is not None:
        require_number(ledger["final_signal"], "escrow ledger.final_signal", minimum=0.0)
    if ledger["escrow_state"] not in {"pending", "preliminary_released", "released", "forfeited"}:
        raise IntegrityError("escrow ledger has invalid state")
    event_ids = ledger["applied_event_ids"]
    if not isinstance(event_ids, list):
        raise IntegrityError("escrow ledger.applied_event_ids must be an array")
    for index, event_id in enumerate(event_ids):
        require_string(event_id, f"escrow ledger.applied_event_ids[{index}]")
    if len(event_ids) != len(set(event_ids)):
        raise IntegrityError("escrow ledger contains duplicate event IDs")
    return ledger


def validate_event(event_value: Any) -> dict[str, Any]:
    """Validate an escrow event and all of its immutable artifact bindings.

    Purpose: ensure that only a declared transition for one exact ledger,
    policy, observation set and attempt can reach the authorization stage.
    """
    event = require_object(event_value, "escrow event")
    require_exact_keys(event, EVENT_KEYS, "escrow event")
    if event["schema_version"] != "escrow-event-0.9":
        raise IntegrityError("escrow event schema_version must be escrow-event-0.9")
    for field in ("event_id", "run_id", "attempt_id", "lineage_id", "pilot_id"):
        require_string(event[field], f"escrow event.{field}")
    if event["event_type"] not in {
        "release_preliminary",
        "release_remaining",
        "forfeit_remaining",
    }:
        raise IntegrityError("escrow event has invalid event_type")
    for field in (
        "authorization_evidence_sha256",
        "policy_sha256",
        "observations_sha256",
        "signal_output_sha256",
        "previous_ledger_sha256",
    ):
        require_sha256(event[field], f"escrow event.{field}")
    return event


def validate_authorization_evidence(
    policy: dict[str, Any],
    event: dict[str, Any],
    evidence_value: Any,
    signature_path: pathlib.Path,
) -> dict[str, Any]:
    """Authenticate and interpret the evidence authorising an escrow event.

    Purpose: derive release facts from signed trusted timestamps and control
    cycle identifiers rather than from unsigned event claims supplied by the
    evaluated system.
    """
    evidence = require_object(evidence_value, "authorization evidence")
    require_exact_keys(
        evidence,
        AUTHORIZATION_EVIDENCE_KEYS,
        "authorization evidence",
    )
    if evidence["schema_version"] != "escrow-authorization-evidence-0.9":
        raise IntegrityError(
            "authorization evidence schema_version "
            "must be escrow-authorization-evidence-0.9"
        )
    for field in (
        "event_id",
        "run_id",
        "attempt_id",
        "lineage_id",
        "pilot_id",
        "authorizer_id",
        "trusted_time_source",
    ):
        require_string(evidence[field], f"authorization evidence.{field}")
    if evidence["event_type"] not in {
        "release_preliminary",
        "release_remaining",
        "forfeit_remaining",
    }:
        raise IntegrityError("authorization evidence has invalid event_type")
    for field in (
        "policy_sha256",
        "observations_sha256",
        "signal_output_sha256",
        "previous_ledger_sha256",
    ):
        require_sha256(evidence[field], f"authorization evidence.{field}")
    # Every operational field is duplicated inside the signed evidence. Exact
    # equality prevents a valid signature from being replayed for another
    # event, attempt, ledger or signal.
    for field in EVENT_BINDING_FIELDS:
        if evidence[field] != event[field]:
            raise IntegrityError(
                f"authorization evidence {field} does not match escrow event"
            )
    if event["authorization_evidence_sha256"] != canonical_sha256(evidence):
        raise IntegrityError("escrow event does not match authorization evidence")

    authorization = policy["escrow"]["authorization"]
    if evidence["authorizer_id"] != authorization["authorizer_id"]:
        raise IntegrityError("authorization evidence has an untrusted authorizer")
    if evidence["trusted_time_source"] not in authorization["trusted_time_sources"]:
        raise IntegrityError("authorization evidence has an untrusted time source")

    started = require_utc_timestamp(
        evidence["observation_started_at_utc"],
        "authorization evidence.observation_started_at_utc",
    )
    completed = require_utc_timestamp(
        evidence["observation_completed_at_utc"],
        "authorization evidence.observation_completed_at_utc",
    )
    effective = require_utc_timestamp(
        evidence["effective_at_utc"],
        "authorization evidence.effective_at_utc",
    )
    if completed < started:
        raise IntegrityError("authorization observation interval is negative")
    if effective < completed:
        raise IntegrityError(
            "authorization cannot take effect before observation is complete"
        )
    cycle_ids = require_array(
        evidence["independent_control_cycle_ids"],
        "authorization evidence.independent_control_cycle_ids",
    )
    validated_cycle_ids = [
        require_string(
            value,
            "authorization evidence.independent_control_cycle_ids item",
        )
        for value in cycle_ids
    ]
    if len(validated_cycle_ids) != len(set(validated_cycle_ids)):
        raise IntegrityError("authorization evidence has duplicate control cycle IDs")
    source_complete = require_boolean(
        evidence["source_complete"],
        "authorization evidence.source_complete",
    )
    late_evidence_clear = require_boolean(
        evidence["late_evidence_clear"],
        "authorization evidence.late_evidence_clear",
    )
    # Signature verification occurs only after structural and policy checks, so
    # the trusted key and canonical payload are both unambiguous.
    verify_minisign_json(
        evidence,
        signature_path,
        authorization["minisign_public_key"],
    )
    return {
        "observation_seconds": int((completed - started).total_seconds()),
        "independent_control_cycles": len(validated_cycle_ids),
        "source_complete": source_complete,
        "late_evidence_clear": late_evidence_clear,
    }


def apply_event(
    policy_value: Any,
    observation_value: Any,
    signal_value: Any,
    ledger_value: Any,
    event_value: Any,
    authorization_evidence_value: Any,
    authorization_signature_path: pathlib.Path,
) -> dict[str, Any]:
    """Apply one authorised transition to a validated escrow ledger.

    Purpose: release or forfeit allocations only when the event, current state,
    signal and signed authorization evidence all refer to the same immutable
    evaluation lineage.
    """
    policy, signal = validate_signal_output(
        policy_value,
        observation_value,
        signal_value,
    )
    ledger = validate_ledger(ledger_value)
    event = validate_event(event_value)
    if ledger["policy_sha256"] != canonical_sha256(policy):
        raise IntegrityError("ledger does not match policy")
    if event["previous_ledger_sha256"] != canonical_sha256(ledger):
        raise IntegrityError("escrow event does not bind to the current ledger")
    if ledger["signal_output_sha256"] != canonical_sha256(signal):
        raise IntegrityError("ledger does not match the supplied signal output")
    for field in (
        "policy_sha256",
        "observations_sha256",
        "run_id",
        "attempt_id",
        "lineage_id",
        "pilot_id",
    ):
        if ledger[field] != signal[field]:
            raise IntegrityError(f"ledger {field} does not match signal output")
    for field in ("gross_signal", "preliminary_allocation"):
        if float(ledger[field]) != float(signal[field]):
            raise IntegrityError(f"ledger {field} does not match signal output")
    for field in (
        "policy_sha256",
        "observations_sha256",
        "signal_output_sha256",
        "run_id",
        "attempt_id",
        "lineage_id",
        "pilot_id",
    ):
        if event[field] != ledger[field]:
            raise IntegrityError(f"escrow event {field} does not match ledger")
    if event["event_id"] in ledger["applied_event_ids"]:
        raise IntegrityError("escrow event has already been applied")
    if ledger["escrow_state"] in {"released", "forfeited"}:
        raise IntegrityError("escrow ledger is already terminal")

    evidence = validate_authorization_evidence(
        policy,
        event,
        authorization_evidence_value,
        authorization_signature_path,
    )
    # Event IDs are retained in the next ledger state to make each transition
    # one-time and auditable.
    result = dict(ledger)
    result["applied_event_ids"] = list(ledger["applied_event_ids"]) + [event["event_id"]]
    event_type = event["event_type"]
    if event_type == "release_preliminary":
        if ledger["escrow_state"] != "pending":
            raise IntegrityError("preliminary release is only valid from pending")
        result["released_signal"] = float(ledger["preliminary_allocation"])
        result["escrow_state"] = "preliminary_released"
        return result

    if ledger["escrow_state"] not in {"pending", "preliminary_released"}:
        raise IntegrityError("remaining escrow cannot be changed from the current state")
    if event_type == "release_remaining":
        # Final release is conditional on facts derived from signed evidence,
        # never on boolean flags supplied by the escrow event itself.
        requirements = policy["escrow"]["release_requirements"]
        if evidence["observation_seconds"] < requirements["minimum_observation_seconds"]:
            raise IntegrityError("minimum observation window has not been completed")
        if (
            evidence["independent_control_cycles"]
            < requirements["minimum_independent_control_cycles"]
        ):
            raise IntegrityError("minimum independent control cycles have not been completed")
        if requirements["require_source_complete"] and not evidence["source_complete"]:
            raise IntegrityError("source completeness requirement has not been met")
        if (
            requirements["require_late_evidence_clear"]
            and not evidence["late_evidence_clear"]
        ):
            raise IntegrityError("late-evidence review has not been completed")
        result["released_signal"] = float(ledger["gross_signal"])
        result["held_allocation"] = 0.0
        result["final_signal"] = float(ledger["gross_signal"])
        result["escrow_state"] = "released"
        return result

    result["held_allocation"] = 0.0
    result["final_signal"] = float(ledger["released_signal"])
    result["escrow_state"] = "forfeited"
    return result


def main(argv: list[str]) -> int:
    """Run ledger initialisation or event application from the command line.

    Purpose: expose one deterministic, fail-closed interface for producing the
    next escrow state as a reviewable JSON artifact.
    """
    if len(argv) not in {6, 10}:
        raise SystemExit(
            "usage: escrow_controller.py init policy.json observations.json "
            "signal_output.json output.json\n"
            "   or: escrow_controller.py apply policy.json observations.json "
            "signal_output.json ledger.json event.json authorization_evidence.json "
            "authorization_evidence.minisig output.json"
        )
    command = argv[1]
    try:
        if command == "init" and len(argv) == 6:
            result = initialize_ledger(
                load_json_strict(pathlib.Path(argv[2])),
                load_json_strict(pathlib.Path(argv[3])),
                load_json_strict(pathlib.Path(argv[4])),
            )
            output_path = pathlib.Path(argv[5])
        elif command == "apply" and len(argv) == 10:
            result = apply_event(
                load_json_strict(pathlib.Path(argv[2])),
                load_json_strict(pathlib.Path(argv[3])),
                load_json_strict(pathlib.Path(argv[4])),
                load_json_strict(pathlib.Path(argv[5])),
                load_json_strict(pathlib.Path(argv[6])),
                load_json_strict(pathlib.Path(argv[7])),
                pathlib.Path(argv[8]),
            )
            output_path = pathlib.Path(argv[9])
        else:
            raise IntegrityError("invalid escrow controller command")
        output_path.write_text(
            json.dumps(result, ensure_ascii=False, allow_nan=False, indent=2) + "\n",
            encoding="utf-8",
        )
    except IntegrityError as exc:
        print(f"escrow validation failed: {exc}", file=sys.stderr)
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))

#!/usr/bin/env python3
"""Validate signed collector logs and their rate-control invariants.

Purpose: ensure that attempts, measurements, alerts and provisional releases
are counted from trusted receipt times in one authenticated, hash-chained event
lineage rather than from agent-supplied timestamps.

Author: Sergei Sychev
Website: www.triz-ri.pro
"""

from __future__ import annotations

import hashlib
import json
import pathlib
import sys
from collections import defaultdict, deque
from typing import Any

from integrity import (
    IntegrityError,
    canonical_sha256,
    load_json_strict,
    require_array,
    require_exact_keys,
    require_integer,
    require_object,
    require_sha256,
    require_string,
)
from minisign_verifier import validate_minisign_public_key, verify_minisign_json


ZERO_SHA256 = "0" * 64
POLICY_KEYS = {
    "schema_version",
    "pilot_id",
    "contract_sha256",
    "collector_id",
    "minisign_public_key",
    "trusted_time_sources",
    "event_window_seconds",
    "maximum_events_per_attempt_window",
    "lineage_window_seconds",
    "maximum_attempts_per_lineage_window",
    "maximum_preliminary_releases_per_lineage",
}
LOG_KEYS = {
    "schema_version",
    "collector_id",
    "trusted_time_source",
    "sealed_at_epoch",
    "events",
}
EVENT_KEYS = {
    "sequence",
    "event_id",
    "event_type",
    "received_at_epoch",
    "attempt_id",
    "lineage_id",
    "payload_sha256",
    "previous_event_sha256",
}


def derive_lineage_id(pilot_id: str, contract_sha256: str) -> str:
    """Derive the stable lineage identifier for one pilot contract.

    Purpose: bind every control event to the same frozen pilot and contract
    without relying on a caller-selected lineage label.
    """
    return hashlib.sha256(
        f"{pilot_id}|{contract_sha256}".encode("utf-8")
    ).hexdigest()


def validate_control_policy(value: Any) -> dict[str, Any]:
    """Validate the collector identity, trust anchors and rate limits.

    Purpose: freeze who may attest receipt times and how much event or attempt
    activity is allowed before any submitted log is examined.
    """
    policy = require_object(value, "control policy")
    require_exact_keys(policy, POLICY_KEYS, "control policy")
    if policy["schema_version"] != "control-policy-0.9":
        raise IntegrityError("control policy schema_version must be control-policy-0.9")
    require_string(policy["pilot_id"], "control policy.pilot_id")
    require_sha256(policy["contract_sha256"], "control policy.contract_sha256")
    require_string(policy["collector_id"], "control policy.collector_id")
    validate_minisign_public_key(
        policy["minisign_public_key"],
        "control policy.minisign_public_key",
    )
    trusted_time_sources = require_array(
        policy["trusted_time_sources"],
        "control policy.trusted_time_sources",
    )
    source_names = [
        require_string(value, "control policy.trusted_time_sources item")
        for value in trusted_time_sources
    ]
    if not source_names or len(source_names) != len(set(source_names)):
        raise IntegrityError(
            "control policy.trusted_time_sources must be non-empty and unique"
        )
    for field, minimum in (
        ("event_window_seconds", 1),
        ("maximum_events_per_attempt_window", 1),
        ("lineage_window_seconds", 1),
        ("maximum_attempts_per_lineage_window", 1),
        ("maximum_preliminary_releases_per_lineage", 0),
    ):
        require_integer(policy[field], f"control policy.{field}", minimum=minimum)
    return policy


def validate_control_log(
    policy_value: Any,
    log_value: Any,
    signature_path: pathlib.Path,
) -> dict[str, Any]:
    """Authenticate a control log and validate its complete event sequence.

    Purpose: detect fabricated timing, event reordering, replay, lineage
    switching and activity floods before control evidence can support a signal
    or escrow decision.
    """
    policy = validate_control_policy(policy_value)
    log = require_object(log_value, "control log")
    require_exact_keys(log, LOG_KEYS, "control log")
    if log["schema_version"] != "control-log-0.9":
        raise IntegrityError("control log schema_version must be control-log-0.9")
    if require_string(log["collector_id"], "control log.collector_id") != policy["collector_id"]:
        raise IntegrityError("control log collector does not match frozen policy")
    trusted_time_source = require_string(
        log["trusted_time_source"],
        "control log.trusted_time_source",
    )
    if trusted_time_source not in policy["trusted_time_sources"]:
        raise IntegrityError("control log has an untrusted time source")
    sealed_at = require_integer(
        log["sealed_at_epoch"],
        "control log.sealed_at_epoch",
        minimum=0,
    )
    # The signature authenticates the collector's receipt times and the entire
    # ordered event list before individual rate and lineage checks begin.
    verify_minisign_json(log, signature_path, policy["minisign_public_key"])

    events = require_array(log["events"], "control log.events")
    expected_lineage = derive_lineage_id(
        policy["pilot_id"],
        policy["contract_sha256"],
    )
    seen_event_ids: set[str] = set()
    registered_attempts: dict[str, str] = {}
    event_windows: dict[str, deque[int]] = defaultdict(deque)
    attempt_windows: dict[str, deque[int]] = defaultdict(deque)
    preliminary_releases: dict[str, int] = defaultdict(int)
    previous_time = -1
    previous_event_sha256 = ZERO_SHA256

    for index, event_value in enumerate(events):
        label = f"control log.events[{index}]"
        event = require_object(event_value, label)
        require_exact_keys(event, EVENT_KEYS, label)
        sequence = require_integer(
            event["sequence"],
            f"{label}.sequence",
            minimum=1,
        )
        if sequence != index + 1:
            raise IntegrityError("control event sequence is not contiguous")
        event_id = require_string(event["event_id"], f"{label}.event_id")
        if event_id in seen_event_ids:
            raise IntegrityError(f"duplicate control event ID: {event_id}")
        seen_event_ids.add(event_id)
        event_type = require_string(event["event_type"], f"{label}.event_type")
        if event_type not in {
            "attempt_registered",
            "measurement",
            "alert",
            "preliminary_release",
            "late_evidence",
        }:
            raise IntegrityError(f"unsupported control event type: {event_type}")
        received_at = require_integer(
            event["received_at_epoch"],
            f"{label}.received_at_epoch",
            minimum=0,
        )
        if received_at < previous_time:
            raise IntegrityError("control events must be in receipt-time order")
        if received_at > sealed_at:
            raise IntegrityError("control event receipt time is later than log sealing time")
        previous_time = received_at
        attempt_id = require_string(event["attempt_id"], f"{label}.attempt_id")
        lineage_id = require_sha256(event["lineage_id"], f"{label}.lineage_id")
        require_sha256(event["payload_sha256"], f"{label}.payload_sha256")
        supplied_previous = require_sha256(
            event["previous_event_sha256"],
            f"{label}.previous_event_sha256",
        )
        # The canonical hash chain makes deletion, insertion and reordering of
        # signed events detectable during replay.
        if supplied_previous != previous_event_sha256:
            raise IntegrityError("control event hash chain is broken")
        previous_event_sha256 = canonical_sha256(event)
        if lineage_id != expected_lineage:
            raise IntegrityError("control event lineage does not match the frozen contract")

        # Sliding windows use collector receipt time, not a timestamp supplied
        # by the evaluated agent.
        queue = event_windows[attempt_id]
        while queue and received_at - queue[0] >= policy["event_window_seconds"]:
            queue.popleft()
        queue.append(received_at)
        if len(queue) > policy["maximum_events_per_attempt_window"]:
            raise IntegrityError("event-rate limit exceeded for attempt " + attempt_id)

        if event_type == "attempt_registered":
            if attempt_id in registered_attempts:
                raise IntegrityError("attempt was registered more than once: " + attempt_id)
            registered_attempts[attempt_id] = lineage_id
            attempt_queue = attempt_windows[lineage_id]
            while (
                attempt_queue
                and received_at - attempt_queue[0] >= policy["lineage_window_seconds"]
            ):
                attempt_queue.popleft()
            attempt_queue.append(received_at)
            if len(attempt_queue) > policy["maximum_attempts_per_lineage_window"]:
                raise IntegrityError("attempt-rate limit exceeded for lineage " + lineage_id)
        elif registered_attempts.get(attempt_id) != lineage_id:
            raise IntegrityError("control event references an unregistered attempt")

        if event_type == "preliminary_release":
            preliminary_releases[lineage_id] += 1
            if (
                preliminary_releases[lineage_id]
                > policy["maximum_preliminary_releases_per_lineage"]
            ):
                raise IntegrityError(
                    "preliminary-release limit exceeded for lineage " + lineage_id
                )

    return {
        "schema_version": "control-log-validation-0.9",
        "valid": True,
        "event_count": len(events),
        "attempt_count": len(registered_attempts),
        "lineage_id": expected_lineage,
        "last_event_sha256": previous_event_sha256,
    }


def main(argv: list[str]) -> int:
    """Run signed control-log validation as a command-line operation.

    Purpose: emit a reproducible validation record, including the final chain
    hash, or a machine-readable fail-closed error for downstream review.
    """
    if len(argv) != 4:
        raise SystemExit(
            "usage: control_log_validator.py control_policy.json "
            "control_log.json control_log.minisig"
        )
    try:
        result = validate_control_log(
            load_json_strict(pathlib.Path(argv[1])),
            load_json_strict(pathlib.Path(argv[2])),
            pathlib.Path(argv[3]),
        )
        print(json.dumps(result, ensure_ascii=False, allow_nan=False, indent=2))
    except IntegrityError as exc:
        print(
            json.dumps(
                {
                    "schema_version": "control-log-validation-0.9",
                    "valid": False,
                    "validation_error": str(exc),
                },
                ensure_ascii=False,
                allow_nan=False,
                indent=2,
            )
        )
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))