geointel/scripts/run_accuracy_phase4_benchmark.py

#!/usr/bin/env python3
"""Run the complete local Phase 4 evaluation workflow from frozen inputs."""

from __future__ import annotations

import argparse
import hashlib
import json
import math
import os
import platform
import re
import shutil
import subprocess
import sys
import tempfile
from datetime import datetime
from importlib import metadata as importlib_metadata
from pathlib import Path
from typing import Any, Callable

ROOT = Path(__file__).resolve().parents[1]
BACKEND_ROOT = ROOT / "backend"
for entry in (str(ROOT), str(BACKEND_ROOT), str(ROOT / "scripts")):
    if entry not in sys.path:
        sys.path.insert(0, entry)

from accuracy_phase4_evaluator import (  # noqa: E402
    EVALUATOR_VERSION,
    EXPECTED_PROTECTED_POLICY,
    SUBGROUP_MIN_CASE_SUPPORT,
    TASKS,
    canonical_hash,
    evaluate_cases,
    task_inventory,
)
from generate_accuracy_phase4_splits import (  # noqa: E402
    GENERATOR_VERSION,
    LeakageError,
    assert_training_inputs_safe,
    build_manifests,
)
from run_golden_qa_benchmark import run_benchmark  # noqa: E402

WORKFLOW_VERSION = "2.0.1"
BENCHMARK_ID = "geointel-p4-reference-harness-v2"
GATE_STATES = {"pass", "fail", "not_evaluable"}
PRODUCT_BASELINE_SCHEMA_VERSION = 2
PRODUCT_BASELINE_MANIFEST_TYPE = "geointel_governed_product_baseline"
PRODUCT_ARTIFACT_ROLES = (
    "configuration",
    "protected_split_manifest",
    "authoritative_reference_manifest",
    "inference_evidence",
    "raw_predictions",
    "metric_report",
    "human_review_ledger",
    "geometric_leakage_audit",
    "vault_access_evidence",
)
PROTECTED_SPLIT_ROLES = {"test", "background-test", "challenge"}
EVALUATED_PROTECTED_SPLIT_ROLES = {"test", "background-test"}
REQUIRED_SUBGROUP_DIMENSION_FIELDS = {
    "region": "region",
    "municipality": "municipality",
    "urbanity": "urbanity",
    "object_size": "object_size",
    "source": "source",
    "sensor": "sensor",
    "resolution": "resolution_m",
    "season": "season",
    "date": "date",
    "vegetation": "vegetation",
    "occlusion": "occlusion",
    "difficulty": "difficulty",
    "context": "context",
}
REQUIRED_SUBGROUP_DIMENSIONS = frozenset(REQUIRED_SUBGROUP_DIMENSION_FIELDS)
SUBGROUP_RELEASE_POLICY = {
    "minimum_case_support_per_task_stratum": SUBGROUP_MIN_CASE_SUPPORT,
    "minimum_distinct_strata_per_dimension": 2,
    "required_dimensions": sorted(REQUIRED_SUBGROUP_DIMENSIONS),
    "missing_or_insufficient_support": "fail",
    "targets_must_be_frozen_before_protected_access": True,
}
REQUIRED_EVALUATOR_TASK_FAMILIES = frozenset(TASKS)
SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$")

LOCAL_GATE_NAMES = frozenset(
    {
        "all_declared_evaluator_families_exercised",
        "implemented_capability_inventory",
        "normative_split_roles_and_leakage",
        "manifest_training_firewall_contract",
        "protected_operating_point_contract",
        "complete_raw_predictions_retained",
        "reference_implementation_baseline",
        "stratified_metric_contract",
        "undefined_metric_truth_table",
    }
)
PRODUCT_GATE_NAMES = frozenset(
    {
        "active_model_available_and_hash_verified",
        "authoritative_reference_portfolio_available",
        "human_review_complete",
        "split_independence",
        "phase3_leakage_resolved",
        "protected_storage_isolation",
        "executed_product_incumbent_baseline",
        "representative_product_subgroup_support",
    }
)
REQUIRED_AUTHORITY_REQUIREMENTS = (
    {"task": "building_validation", "zone": "flanders", "primary": "grb"},
    {"task": "building_validation", "zone": "wallonia", "primary": "picc"},
    {"task": "building_validation", "zone": "brussels", "primary": "urbis"},
    {"task": "terrain_height", "zone": "flanders", "primary": "dhmv"},
    {"task": "terrain_height", "zone": "wallonia", "primary": "spw_terrain"},
    {
        "task": "north_sea_bathymetry",
        "zone": "belgian_north_sea",
        "primary": "mdk",
    },
    {
        "task": "imagery_corroboration",
        "zone": "belgium",
        "primary": "official_orthophoto",
        "contextual": "sentinel-2",
    },
)


class EvidenceConflictError(RuntimeError):
    """Raised when an immutable evidence path already contains different bytes."""


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


def json_bytes(payload: Any) -> bytes:
    return (
        json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
    ).encode("utf-8")


def is_sha256(value: Any) -> bool:
    return isinstance(value, str) and SHA256_PATTERN.fullmatch(value) is not None


def parse_timestamp(value: Any) -> datetime | None:
    if not isinstance(value, str) or not value.strip():
        return None
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None
    return parsed if parsed.tzinfo is not None else None


def _raise_evidence_conflict(path: Path, detail: str = "different content") -> None:
    raise EvidenceConflictError(
        f"Refusing to overwrite immutable evidence with {detail}: {path}"
    )


def write_json_immutable(path: Path, payload: Any) -> None:
    """Create one immutable JSON file without a check-then-overwrite race."""

    content = json_bytes(payload)
    path.parent.mkdir(parents=True, exist_ok=True)
    if path.exists():
        if not path.is_file() or path.read_bytes() != content:
            _raise_evidence_conflict(path)
        return
    file_descriptor, temporary_name = tempfile.mkstemp(
        prefix=f".{path.name}.",
        suffix=".tmp",
        dir=path.parent,
    )
    temporary = Path(temporary_name)
    try:
        with os.fdopen(file_descriptor, "wb") as stream:
            stream.write(content)
            stream.flush()
            os.fsync(stream.fileno())
        try:
            os.link(temporary, path)
        except FileExistsError:
            if not path.is_file() or path.read_bytes() != content:
                _raise_evidence_conflict(path)
    finally:
        temporary.unlink(missing_ok=True)


def _assert_immutable_bundle(
    output_dir: Path,
    contents: dict[str, bytes],
) -> None:
    if not output_dir.is_dir():
        _raise_evidence_conflict(output_dir, "a non-directory target")
    observed_entries = {
        path.relative_to(output_dir).as_posix(): (
            "file" if path.is_file() else "directory" if path.is_dir() else "other"
        )
        for path in output_dir.rglob("*")
    }
    expected_entries = {name: "file" for name in contents}
    if observed_entries != expected_entries:
        _raise_evidence_conflict(
            output_dir,
            "an incomplete, nested or unexpected artifact set; "
            f"expected={sorted(expected_entries.items())}, "
            f"observed={sorted(observed_entries.items())}",
        )
    for name, content in contents.items():
        path = output_dir / name
        if not path.is_file() or path.read_bytes() != content:
            _raise_evidence_conflict(path)


def write_json_bundle_immutable(output_dir: Path, payloads: dict[str, Any]) -> None:
    """Publish a complete immutable evidence bundle with an atomic directory rename."""

    contents = {name: json_bytes(payload) for name, payload in payloads.items()}
    output_dir.parent.mkdir(parents=True, exist_ok=True)
    if output_dir.exists():
        _assert_immutable_bundle(output_dir, contents)
        return
    staging = Path(
        tempfile.mkdtemp(prefix=f".{output_dir.name}.", dir=output_dir.parent)
    )
    try:
        for name, content in sorted(contents.items()):
            path = staging / name
            with path.open("xb") as stream:
                stream.write(content)
                stream.flush()
                os.fsync(stream.fileno())
        try:
            staging.replace(output_dir)
        except FileExistsError:
            _assert_immutable_bundle(output_dir, contents)
    finally:
        if staging.exists():
            shutil.rmtree(staging)


def repository_commit(repo_root: Path) -> str | None:
    try:
        return subprocess.run(
            ["git", "rev-parse", "HEAD"],
            cwd=repo_root,
            check=True,
            capture_output=True,
            text=True,
        ).stdout.strip()
    except (OSError, subprocess.CalledProcessError):
        return None


def dependency_version(distribution: str) -> str | None:
    try:
        return importlib_metadata.version(distribution)
    except importlib_metadata.PackageNotFoundError:
        return None


def repository_file(repo_root: Path, relative_path: str) -> dict[str, Any]:
    path = repo_root / relative_path
    return {
        "path": relative_path,
        "sha256": sha256(path),
        "size_bytes": path.stat().st_size,
    }


def canonical_golden_baseline() -> dict[str, Any]:
    result = run_benchmark()
    scenarios = []
    for item in result["scenarios"]:
        normalized = dict(item)
        normalized.pop("quality_check_id", None)
        scenarios.append(normalized)
    return {
        "status": result["status"],
        "version": result["version"],
        "scenario_count": result["scenario_count"],
        "scenarios": scenarios,
        "persistence": result["persistence"],
        "implementation": "backend/app/services/qa_service.py via scripts/run_golden_qa_benchmark.py",
        "claim_boundary": "Reference implementation regression evidence; not production model accuracy.",
        "content_sha256": canonical_hash(scenarios),
    }


def _canonical_id_hash(values: set[str]) -> str:
    return canonical_hash(sorted(values))


def _load_governed_artifacts(
    repo_root: Path,
    governed_root: Path,
    manifest: dict[str, Any],
    violations: list[str],
) -> tuple[dict[str, dict[str, Any]], list[dict[str, Any]]]:
    payloads: dict[str, dict[str, Any]] = {}
    checked: list[dict[str, Any]] = []
    observed_paths: set[Path] = set()
    for role in PRODUCT_ARTIFACT_ROLES:
        descriptor = manifest.get(role)
        if not isinstance(descriptor, dict):
            violations.append(f"{role}:descriptor_not_object")
            continue
        descriptor_fields = {"path", "sha256", "size_bytes"}
        missing_fields = descriptor_fields - set(descriptor)
        unexpected_fields = set(descriptor) - descriptor_fields
        if missing_fields:
            violations.append(
                f"{role}:descriptor_missing:{','.join(sorted(missing_fields))}"
            )
        if unexpected_fields:
            violations.append(
                f"{role}:descriptor_unexpected:{','.join(sorted(unexpected_fields))}"
            )
        relative_path = descriptor.get("path")
        if (
            not isinstance(relative_path, str)
            or not relative_path.strip()
            or Path(relative_path).is_absolute()
        ):
            violations.append(f"{role}:path_not_relative")
            continue
        artifact_path = (repo_root / relative_path).resolve()
        try:
            artifact_path.relative_to(governed_root.resolve())
        except (OSError, ValueError):
            violations.append(f"{role}:outside_governed_evidence_root")
            continue
        if artifact_path in observed_paths:
            violations.append(f"{role}:artifact_path_reused")
            continue
        observed_paths.add(artifact_path)
        if not artifact_path.is_file():
            violations.append(f"{role}:missing")
            continue
        observed_hash = sha256(artifact_path)
        observed_size = artifact_path.stat().st_size
        checked.append(
            {
                "role": role,
                "path": artifact_path.relative_to(repo_root.resolve()).as_posix(),
                "sha256": observed_hash,
                "size_bytes": observed_size,
            }
        )
        if not is_sha256(descriptor.get("sha256")):
            violations.append(f"{role}:invalid_declared_sha256")
        elif observed_hash != descriptor["sha256"]:
            violations.append(f"{role}:hash_mismatch")
        if descriptor.get("size_bytes") != observed_size:
            violations.append(f"{role}:size_mismatch")
        try:
            payload = json.loads(artifact_path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError) as exc:
            violations.append(f"{role}:invalid_json:{type(exc).__name__}")
            continue
        if not isinstance(payload, dict):
            violations.append(f"{role}:payload_not_object")
            continue
        payloads[role] = payload
    return payloads, checked


def _validate_product_manifest_contract(
    manifest: dict[str, Any],
    active_model: dict[str, Any],
    evaluator_hash: str,
) -> list[str]:
    violations: list[str] = []
    required_fields = {
        "schema_version",
        "manifest_type",
        "baseline_id",
        "created_at",
        "status",
        "synthetic",
        "active_model",
        "active_model_sha256",
        "evaluator_sha256",
        "configuration_sha256",
        "development_split_manifest_sha256",
        "selection_isolation",
        "inference",
        *PRODUCT_ARTIFACT_ROLES,
    }
    missing_fields = required_fields - set(manifest)
    unexpected_fields = set(manifest) - required_fields
    if missing_fields:
        violations.append(f"manifest:missing:{','.join(sorted(missing_fields))}")
    if unexpected_fields:
        violations.append(f"manifest:unexpected:{','.join(sorted(unexpected_fields))}")
    if manifest.get("schema_version") != PRODUCT_BASELINE_SCHEMA_VERSION:
        violations.append("manifest:schema_version")
    if manifest.get("manifest_type") != PRODUCT_BASELINE_MANIFEST_TYPE:
        violations.append("manifest:type")
    if (
        not isinstance(manifest.get("baseline_id"), str)
        or not manifest["baseline_id"].strip()
    ):
        violations.append("manifest:baseline_id")
    if parse_timestamp(manifest.get("created_at")) is None:
        violations.append("manifest:created_at")
    if manifest.get("status") != "pass":
        violations.append("manifest:status_not_pass")
    if manifest.get("synthetic") is not False:
        violations.append("manifest:synthetic_or_unspecified")

    expected_identity = {
        key: active_model.get(key)
        for key in ("model_id", "model_version", "sha256", "size_bytes")
    }
    if manifest.get("active_model") != expected_identity:
        violations.append("manifest:active_model_identity_mismatch")
    expected_model_hash = active_model.get("sha256")
    if not is_sha256(expected_model_hash):
        violations.append("manifest:configured_active_model_sha256_invalid")
    if manifest.get("active_model_sha256") != expected_model_hash:
        violations.append("manifest:active_model_hash_mismatch")
    if manifest.get("evaluator_sha256") != evaluator_hash:
        violations.append("manifest:evaluator_hash_mismatch")
    for field in ("configuration_sha256", "development_split_manifest_sha256"):
        if not is_sha256(manifest.get(field)):
            violations.append(f"manifest:{field}_invalid")

    inference = manifest.get("inference")
    if not isinstance(inference, dict):
        violations.append("manifest:inference_not_object")
    else:
        if inference.get("executed") is not True:
            violations.append("manifest:inference_not_executed")
        if inference.get("test_used_for_selection") is not False:
            violations.append("manifest:protected_test_selection_policy_invalid")
        if re.fullmatch(r"cuda:\d+", str(inference.get("device") or "")) is None:
            violations.append("manifest:cuda_device_invalid")
        if (
            not isinstance(inference.get("execution_id"), str)
            or not inference["execution_id"].strip()
        ):
            violations.append("manifest:execution_id")

    isolation = manifest.get("selection_isolation")
    required_isolation = {
        "test_used_for_training": False,
        "test_used_for_threshold_selection": False,
        "test_used_for_model_selection": False,
        "test_used_for_iterative_error_correction": False,
        "challenge_labels_accessed": False,
        "operating_point_frozen_before_protected_inference": True,
        "configuration_sha256": manifest.get("configuration_sha256"),
    }
    if isolation != required_isolation:
        violations.append("manifest:selection_isolation_invalid")
    return violations


def _validate_configuration(
    payload: dict[str, Any],
    manifest: dict[str, Any],
) -> list[str]:
    violations: list[str] = []
    if payload.get("schema_version") != 2:
        violations.append("configuration:schema_version")
    if payload.get("artifact_role") != "frozen_inference_configuration":
        violations.append("configuration:artifact_role")
    if payload.get("active_model_sha256") != manifest.get("active_model_sha256"):
        violations.append("configuration:active_model_hash_mismatch")
    if payload.get("development_split_manifest_sha256") != manifest.get(
        "development_split_manifest_sha256"
    ):
        violations.append("configuration:development_split_hash_mismatch")
    if payload.get("frozen_before_protected_access") is not True:
        violations.append("configuration:not_frozen_before_protected_access")
    if parse_timestamp(payload.get("frozen_at")) is None:
        violations.append("configuration:frozen_at")
    if payload.get("protected_data_used") is not False:
        violations.append("configuration:protected_data_used")
    if payload.get("threshold_selection_source") not in {
        "validation_only",
        "calibration_only",
        "validation_and_calibration",
    }:
        violations.append("configuration:threshold_selection_source")
    parameters = payload.get("parameters_by_task")
    if (
        not isinstance(parameters, dict)
        or set(parameters) != REQUIRED_EVALUATOR_TASK_FAMILIES
    ):
        violations.append("configuration:parameters_by_task")
    elif any(not isinstance(value, dict) for value in parameters.values()):
        violations.append("configuration:task_parameters_not_objects")
    if payload.get("subgroup_release_policy") != SUBGROUP_RELEASE_POLICY:
        violations.append("configuration:subgroup_release_policy")
    targets = payload.get("subgroup_release_targets")
    if (
        not isinstance(targets, dict)
        or set(targets) != REQUIRED_EVALUATOR_TASK_FAMILIES
    ):
        violations.append("configuration:subgroup_release_targets")
    elif payload.get("subgroup_release_targets_sha256") != canonical_hash(targets):
        violations.append("configuration:subgroup_release_targets_sha256")
    return violations


def _validate_protected_split(
    payload: dict[str, Any],
) -> tuple[list[str], dict[str, dict[str, Any]]]:
    violations: list[str] = []
    samples_by_id: dict[str, dict[str, Any]] = {}
    if payload.get("schema_version") != 3:
        violations.append("protected_split:schema_version")
    if payload.get("artifact_role") != "protected_evaluation_split":
        violations.append("protected_split:artifact_role")
    expected_policy = {
        "immutable": True,
        "training_allowed": False,
        "threshold_selection_allowed": False,
        "model_selection_allowed": False,
        "iterative_error_correction_allowed": False,
        "challenge_labels_accessible": False,
    }
    if payload.get("protected_policy") != expected_policy:
        violations.append("protected_split:policy")
    if payload.get("evaluator_task_inventory_sha256") != canonical_hash(
        task_inventory()
    ):
        violations.append("protected_split:evaluator_task_inventory_sha256")

    samples = payload.get("samples")
    if not isinstance(samples, list) or not samples:
        violations.append("protected_split:samples")
        return violations, samples_by_id
    split_counts: dict[str, int] = {}
    for index, sample in enumerate(samples):
        prefix = f"protected_split:sample:{index}"
        if not isinstance(sample, dict):
            violations.append(f"{prefix}:not_object")
            continue
        sample_id = sample.get("sample_id")
        if (
            not isinstance(sample_id, str)
            or not sample_id
            or sample_id != sample_id.strip()
        ):
            violations.append(f"{prefix}:sample_id")
            continue
        if sample_id in samples_by_id:
            violations.append(f"{prefix}:duplicate_sample_id")
            continue
        samples_by_id[sample_id] = sample
        split = sample.get("split")
        if split not in PROTECTED_SPLIT_ROLES:
            violations.append(f"{prefix}:split")
        else:
            split_counts[split] = split_counts.get(split, 0) + 1
        if sample.get("task") not in REQUIRED_EVALUATOR_TASK_FAMILIES:
            violations.append(f"{prefix}:task")
        for field in ("zone", "aoi_id"):
            value = sample.get(field)
            if not isinstance(value, str) or not value.strip():
                violations.append(f"{prefix}:{field}")
        if not is_sha256(sample.get("content_sha256")):
            violations.append(f"{prefix}:content_sha256")
        subgroups = sample.get("subgroups")
        if not isinstance(subgroups, dict) or set(subgroups) != (
            REQUIRED_SUBGROUP_DIMENSIONS
        ):
            violations.append(f"{prefix}:subgroups")
        else:
            for dimension, value in subgroups.items():
                if dimension == "resolution":
                    if (
                        isinstance(value, bool)
                        or not isinstance(value, (int, float))
                        or not math.isfinite(float(value))
                        or float(value) <= 0
                    ):
                        violations.append(f"{prefix}:subgroup:resolution")
                elif not isinstance(value, str) or not value.strip():
                    violations.append(f"{prefix}:subgroup:{dimension}")
        scopes = sample.get("authority_scopes")
        if not isinstance(scopes, list) or not scopes:
            violations.append(f"{prefix}:authority_scopes")
        elif any(
            not isinstance(scope, dict)
            or set(scope) != {"task", "zone", "authority"}
            or any(
                not isinstance(scope.get(field), str) or not scope[field].strip()
                for field in ("task", "zone", "authority")
            )
            for scope in scopes
        ):
            violations.append(f"{prefix}:authority_scope_contract")

        if split in EVALUATED_PROTECTED_SPLIT_ROLES:
            if not is_sha256(sample.get("label_sha256")):
                violations.append(f"{prefix}:label_sha256")
            if not is_sha256(sample.get("case_input_sha256")):
                violations.append(f"{prefix}:case_input_sha256")
            if sample.get("labels_access_policy") != "evaluation_only":
                violations.append(f"{prefix}:labels_access_policy")
        elif split == "challenge":
            if sample.get("labels_sealed") is not True:
                violations.append(f"{prefix}:challenge_labels_not_sealed")
            if "label_sha256" in sample or "case_input_sha256" in sample:
                violations.append(f"{prefix}:challenge_label_evidence_exposed")

    if set(split_counts) != PROTECTED_SPLIT_ROLES:
        violations.append("protected_split:required_roles")
    if payload.get("split_counts") != dict(sorted(split_counts.items())):
        violations.append("protected_split:split_counts")
    all_ids = set(samples_by_id)
    evaluation_ids = {
        sample_id
        for sample_id, sample in samples_by_id.items()
        if sample.get("split") in EVALUATED_PROTECTED_SPLIT_ROLES
    }
    if payload.get("sample_ids_sha256") != _canonical_id_hash(all_ids):
        violations.append("protected_split:sample_ids_sha256")
    if payload.get("evaluation_sample_ids_sha256") != _canonical_id_hash(
        evaluation_ids
    ):
        violations.append("protected_split:evaluation_sample_ids_sha256")

    task_sample_ids = {
        task: sorted(
            sample_id
            for sample_id in evaluation_ids
            if samples_by_id[sample_id].get("task") == task
        )
        for task in sorted(REQUIRED_EVALUATOR_TASK_FAMILIES)
    }
    observed_tasks = {
        task for task, sample_ids in task_sample_ids.items() if sample_ids
    }
    if observed_tasks != REQUIRED_EVALUATOR_TASK_FAMILIES:
        violations.append("protected_split:evaluator_task_family_coverage")
    if payload.get("evaluated_task_families") != sorted(
        REQUIRED_EVALUATOR_TASK_FAMILIES
    ):
        violations.append("protected_split:evaluated_task_families")
    if payload.get("task_sample_ids") != task_sample_ids:
        violations.append("protected_split:task_sample_ids")
    if payload.get("task_sample_ids_canonical_json_sha256") != canonical_hash(
        task_sample_ids
    ):
        violations.append("protected_split:task_sample_ids_sha256")

    for dimension in sorted(REQUIRED_SUBGROUP_DIMENSIONS):
        values = {
            str(samples_by_id[sample_id].get("subgroups", {}).get(dimension))
            for sample_id in evaluation_ids
        }
        if (
            len(values)
            < SUBGROUP_RELEASE_POLICY["minimum_distinct_strata_per_dimension"]
        ):
            violations.append(f"protected_split:subgroup:{dimension}:not_stratified")
        for value in values:
            for task in sorted(REQUIRED_EVALUATOR_TASK_FAMILIES):
                group_ids = [
                    sample_id
                    for sample_id in evaluation_ids
                    if samples_by_id[sample_id].get("task") == task
                    and str(
                        samples_by_id[sample_id].get("subgroups", {}).get(dimension)
                    )
                    == value
                ]
                if len(group_ids) < SUBGROUP_MIN_CASE_SUPPORT:
                    violations.append(
                        f"protected_split:subgroup:{dimension}:{value}:{task}:sample_support"
                    )
                independent_aois = {
                    samples_by_id[sample_id].get("aoi_id") for sample_id in group_ids
                }
                if len(independent_aois) < SUBGROUP_MIN_CASE_SUPPORT:
                    violations.append(
                        f"protected_split:subgroup:{dimension}:{value}:{task}:aoi_support"
                    )
    return violations, samples_by_id


def probe_local_cuda_runtime() -> dict[str, Any]:
    """Independently observe the current local CUDA runtime; never trust a receipt."""

    try:
        import torch
    except (ImportError, OSError) as exc:
        return {
            "status": "not_evaluable",
            "reason": f"Local torch runtime is unavailable: {type(exc).__name__}",
        }
    try:
        if torch.cuda.is_available() is not True:
            return {
                "status": "not_evaluable",
                "reason": "torch.cuda.is_available() is false.",
            }
        device_index = int(torch.cuda.current_device())
        device = f"cuda:{device_index}"

        probe_value = float(
            (torch.ones(4, device=device, dtype=torch.float32) * 2).sum().item()
        )
        if probe_value != 8.0:
            return {
                "status": "not_evaluable",
                "reason": "The independent CUDA kernel probe returned an invalid value.",
            }
        smi = subprocess.run(
            [
                "nvidia-smi",
                f"--id={device_index}",
                "--query-gpu=uuid,name,driver_version",
                "--format=csv,noheader,nounits",
            ],
            check=True,
            capture_output=True,
            text=True,
            timeout=10,
        )
        rows = [row.strip() for row in smi.stdout.splitlines() if row.strip()]
        if len(rows) != 1:
            return {
                "status": "not_evaluable",
                "reason": "nvidia-smi did not return exactly one device row.",
            }
        parts = [part.strip() for part in rows[0].split(",")]
        if len(parts) != 3:
            return {
                "status": "not_evaluable",
                "reason": "nvidia-smi device evidence is malformed.",
            }
        gpu_uuid, device_name, driver_version = parts
        cuda_runtime = getattr(torch.version, "cuda", None)
        if (
            not gpu_uuid.startswith("GPU-")
            or not device_name
            or not driver_version
            or not isinstance(cuda_runtime, str)
            or not cuda_runtime
        ):
            return {
                "status": "not_evaluable",
                "reason": "Independent CUDA identity is incomplete.",
            }
        return {
            "status": "pass",
            "device": device,
            "device_name": device_name,
            "gpu_uuid": gpu_uuid,
            "driver_version": driver_version,
            "cuda_runtime_version": cuda_runtime,
            "torch_version": str(torch.__version__),
            "cuda_device_count": int(torch.cuda.device_count()),
            "kernel_execution_confirmed": True,
        }
    except (OSError, RuntimeError, subprocess.SubprocessError, ValueError) as exc:
        return {
            "status": "not_evaluable",
            "reason": f"Independent CUDA verification failed: {type(exc).__name__}",
        }


def _validate_inference_evidence(
    payload: dict[str, Any],
    manifest: dict[str, Any],
    configuration: dict[str, Any],
    protected_samples: dict[str, dict[str, Any]],
    protected_split_hash: Any,
    runtime_observation: dict[str, Any],
) -> tuple[list[str], str, dict[str, Any]]:
    violations: list[str] = []
    if payload.get("schema_version") != 2:
        violations.append("inference_evidence:schema_version")
    if payload.get("artifact_role") != "governed_cuda_inference_execution":
        violations.append("inference_evidence:artifact_role")
    bindings = {
        "active_model_sha256": manifest.get("active_model_sha256"),
        "configuration_sha256": manifest.get("configuration_sha256"),
        "evaluator_sha256": manifest.get("evaluator_sha256"),
        "protected_split_sha256": protected_split_hash,
    }
    for field, expected in bindings.items():
        if payload.get(field) != expected:
            violations.append(f"inference_evidence:{field}_mismatch")
    inference = manifest.get("inference")
    inference = inference if isinstance(inference, dict) else {}
    if payload.get("execution_id") != inference.get("execution_id"):
        violations.append("inference_evidence:execution_id_mismatch")
    if payload.get("executed") is not True or payload.get("exit_code") != 0:
        violations.append("inference_evidence:execution_not_successful")
    if payload.get("test_used_for_selection") is not False:
        violations.append("inference_evidence:test_used_for_selection")
    if payload.get("device_type") != "cuda":
        violations.append("inference_evidence:device_type")
    device = payload.get("device")
    if (
        device != inference.get("device")
        or re.fullmatch(r"cuda:\d+", str(device or "")) is None
    ):
        violations.append("inference_evidence:device")
    if payload.get("torch_cuda_is_available") is not True:
        violations.append("inference_evidence:torch_cuda_unavailable")
    device_count = payload.get("cuda_device_count")
    if (
        not isinstance(device_count, int)
        or isinstance(device_count, bool)
        or device_count < 1
    ):
        violations.append("inference_evidence:cuda_device_count")
    if payload.get("kernel_execution_confirmed") is not True:
        violations.append("inference_evidence:kernel_not_confirmed")
    if payload.get("batch_failure_count") != 0:
        violations.append("inference_evidence:batch_failures")
    for field in ("torch_version", "cuda_runtime_version", "driver_version"):
        if not isinstance(payload.get(field), str) or not payload[field].strip():
            violations.append(f"inference_evidence:{field}")
    started = parse_timestamp(payload.get("started_at"))
    finished = parse_timestamp(payload.get("finished_at"))
    frozen = parse_timestamp(configuration.get("frozen_at"))
    if started is None or finished is None or finished <= started:
        violations.append("inference_evidence:timestamps")
    if frozen is None or started is None or frozen >= started:
        violations.append("inference_evidence:configuration_not_pre_registered")

    nvidia_smi = payload.get("nvidia_smi")
    if not isinstance(nvidia_smi, dict):
        violations.append("inference_evidence:nvidia_smi")
        nvidia_smi = {}
    else:
        gpu_uuid = nvidia_smi.get("gpu_uuid")
        if (
            not isinstance(gpu_uuid, str)
            or not gpu_uuid.startswith("GPU-")
            or len(gpu_uuid) <= 4
        ):
            violations.append("inference_evidence:nvidia_smi_gpu_uuid")
        for field in ("device_name", "driver_version", "cuda_version"):
            if (
                not isinstance(nvidia_smi.get(field), str)
                or not nvidia_smi[field].strip()
            ):
                violations.append(f"inference_evidence:nvidia_smi_{field}")
        if not is_sha256(nvidia_smi.get("query_output_sha256")):
            violations.append("inference_evidence:nvidia_smi_output_sha256")

    expected_ids = {
        sample_id
        for sample_id, sample in protected_samples.items()
        if sample.get("split") in EVALUATED_PROTECTED_SPLIT_ROLES
    }
    processed_ids = payload.get("processed_sample_ids")
    if (
        not isinstance(processed_ids, list)
        or any(not isinstance(item, str) for item in processed_ids)
        or len(processed_ids) != len(set(processed_ids))
        or set(processed_ids) != expected_ids
    ):
        violations.append("inference_evidence:processed_sample_ids")
    if payload.get("processed_sample_ids_sha256") != _canonical_id_hash(expected_ids):
        violations.append("inference_evidence:processed_sample_ids_sha256")
    if payload.get("successful_sample_count") != len(expected_ids):
        violations.append("inference_evidence:successful_sample_count")

    runtime_status = runtime_observation.get("status")
    if runtime_status not in {"pass", "not_evaluable"}:
        violations.append("inference_evidence:independent_runtime_status")
        runtime_status = "fail"
    if runtime_status == "pass":
        expected_runtime = {
            "device": payload.get("device"),
            "device_name": nvidia_smi.get("device_name"),
            "gpu_uuid": nvidia_smi.get("gpu_uuid"),
            "driver_version": payload.get("driver_version"),
            "cuda_runtime_version": payload.get("cuda_runtime_version"),
            "torch_version": payload.get("torch_version"),
            "cuda_device_count": payload.get("cuda_device_count"),
            "kernel_execution_confirmed": True,
        }
        observed_runtime = {
            field: runtime_observation.get(field) for field in expected_runtime
        }
        if observed_runtime != expected_runtime:
            violations.append("inference_evidence:independent_runtime_mismatch")
    return violations, str(runtime_status), runtime_observation


def _validate_authoritative_reference_portfolio(
    payload: dict[str, Any],
    protected_samples: dict[str, dict[str, Any]],
    protected_split_hash: Any,
    authority_requirements: list[dict[str, Any]],
    evaluation: dict[str, Any] | None = None,
) -> tuple[list[str], dict[str, Any]]:
    """Validate authority scope coverage and bind every evaluator reference payload."""

    violations: list[str] = []
    if payload.get("schema_version") != 2:
        violations.append("authoritative_reference:schema_version")
    if payload.get("artifact_role") != "authoritative_reference_portfolio":
        violations.append("authoritative_reference:artifact_role")
    if payload.get("protected_split_sha256") != protected_split_hash:
        violations.append("authoritative_reference:protected_split_hash_mismatch")
    if (
        not isinstance(payload.get("portfolio_id"), str)
        or not payload["portfolio_id"].strip()
    ):
        violations.append("authoritative_reference:portfolio_id")

    required = {
        (item["task"], item["zone"]): item["primary"] for item in authority_requirements
    }
    expected_evaluation_ids = {
        sample_id
        for sample_id, sample in protected_samples.items()
        if sample.get("split") in EVALUATED_PROTECTED_SPLIT_ROLES
    }
    entries = payload.get("entries")
    observed: dict[tuple[str, str], str] = {}
    referenced_ids: set[str] = set()
    if not isinstance(entries, list) or not entries:
        violations.append("authoritative_reference:entries")
        entries = []
    elif payload.get("entries_canonical_json_sha256") != canonical_hash(entries):
        violations.append("authoritative_reference:entries_sha256")

    for index, entry in enumerate(entries):
        prefix = f"authoritative_reference:entry:{index}"
        if not isinstance(entry, dict):
            violations.append(f"{prefix}:not_object")
            continue
        key = (entry.get("task"), entry.get("zone"))
        authority = entry.get("authority")
        if not all(isinstance(value, str) and value.strip() for value in key):
            violations.append(f"{prefix}:task_zone")
        elif key in observed:
            violations.append(f"{prefix}:duplicate_task_zone")
        else:
            observed[(str(key[0]), str(key[1]))] = str(authority)
        if entry.get("source_classification") != "authoritative":
            violations.append(f"{prefix}:source_classification")
        if key in required and authority != required[key]:
            violations.append(f"{prefix}:authority_mismatch")
        if (
            not isinstance(entry.get("source_snapshot_id"), str)
            or not entry["source_snapshot_id"].strip()
        ):
            violations.append(f"{prefix}:source_snapshot_id")
        if not is_sha256(entry.get("source_snapshot_sha256")):
            violations.append(f"{prefix}:source_snapshot_sha256")
        sample_ids = entry.get("sample_ids")
        if (
            not isinstance(sample_ids, list)
            or not sample_ids
            or any(not isinstance(item, str) for item in sample_ids)
            or len(sample_ids) != len(set(sample_ids))
        ):
            violations.append(f"{prefix}:sample_ids")
            continue
        scope = {
            "task": entry.get("task"),
            "zone": entry.get("zone"),
            "authority": authority,
        }
        for sample_id in sample_ids:
            sample = protected_samples.get(sample_id)
            if (
                sample is None
                or sample.get("split") not in EVALUATED_PROTECTED_SPLIT_ROLES
            ):
                violations.append(f"{prefix}:unknown_or_unevaluated_sample:{sample_id}")
                continue
            referenced_ids.add(sample_id)
            if scope not in (sample.get("authority_scopes") or []):
                violations.append(
                    f"{prefix}:sample_authority_scope_mismatch:{sample_id}"
                )

    for key, authority in required.items():
        if observed.get(key) != authority:
            violations.append(
                f"authoritative_reference:missing_requirement:{key[0]}:{key[1]}:{authority}"
            )
    if referenced_ids != expected_evaluation_ids:
        violations.append(
            "authoritative_reference:sample_coverage:"
            f"missing={','.join(sorted(expected_evaluation_ids - referenced_ids))}:"
            f"unexpected={','.join(sorted(referenced_ids - expected_evaluation_ids))}"
        )

    sample_references = payload.get("sample_references")
    sample_references_by_id: dict[str, dict[str, Any]] = {}
    if not isinstance(sample_references, list):
        violations.append("authoritative_reference:sample_references")
        sample_references = []
    elif payload.get("sample_references_canonical_json_sha256") != canonical_hash(
        sample_references
    ):
        violations.append("authoritative_reference:sample_references_sha256")
    reference_fields = {
        "sample_id",
        "task",
        "reference_payload_sha256",
        "reference_lineage_sha256",
    }
    for index, item in enumerate(sample_references):
        prefix = f"authoritative_reference:sample_reference:{index}"
        if not isinstance(item, dict):
            violations.append(f"{prefix}:not_object")
            continue
        if set(item) != reference_fields:
            violations.append(f"{prefix}:fields")
        sample_id = item.get("sample_id")
        if (
            not isinstance(sample_id, str)
            or sample_id not in expected_evaluation_ids
            or sample_id in sample_references_by_id
        ):
            violations.append(f"{prefix}:sample_id")
            continue
        sample_references_by_id[sample_id] = item
        if item.get("task") != protected_samples[sample_id].get("task"):
            violations.append(f"{prefix}:task")
        for field in ("reference_payload_sha256", "reference_lineage_sha256"):
            if not is_sha256(item.get(field)):
                violations.append(f"{prefix}:{field}")
    if set(sample_references_by_id) != expected_evaluation_ids:
        violations.append("authoritative_reference:sample_reference_coverage")

    if evaluation is not None:
        results_by_id = {
            result.get("sample_id"): result
            for result in evaluation.get("results", [])
            if isinstance(result, dict) and isinstance(result.get("sample_id"), str)
        }
        for sample_id in sorted(expected_evaluation_ids):
            result = results_by_id.get(sample_id)
            provided = sample_references_by_id.get(sample_id)
            if result is None or provided is None:
                continue
            raw = result.get("raw") if isinstance(result.get("raw"), dict) else {}
            hashes = raw.get("hashes") if isinstance(raw.get("hashes"), dict) else {}
            lineage = (
                raw.get("input_lineage")
                if isinstance(raw.get("input_lineage"), dict)
                else {}
            )
            expected_reference = {
                "sample_id": sample_id,
                "task": result.get("task"),
                "reference_payload_sha256": hashes.get(
                    "references_canonical_json_sha256"
                ),
                "reference_lineage_sha256": canonical_hash(lineage.get("reference")),
            }
            if provided != expected_reference:
                violations.append(
                    f"authoritative_reference:sample_reference_mismatch:{sample_id}"
                )

    evidence = {
        "required": [
            {"task": task, "zone": zone, "authority": authority}
            for (task, zone), authority in sorted(required.items())
        ],
        "observed": [
            {"task": task, "zone": zone, "authority": authority}
            for (task, zone), authority in sorted(observed.items())
            if (task, zone) in required
        ],
        "covered_evaluation_sample_count": len(referenced_ids),
        "reference_bound_sample_count": len(sample_references_by_id),
        "required_evaluation_sample_count": len(expected_evaluation_ids),
    }
    return violations, evidence


def _evaluate_raw_product_portfolio(
    repo_root: Path,
    portfolio_path: Path,
    payload: dict[str, Any],
    manifest: dict[str, Any],
    configuration: dict[str, Any],
    protected_samples: dict[str, dict[str, Any]],
    protected_split_hash: Any,
    authority_reference_hash: Any,
    inference_evidence_hash: Any,
) -> tuple[list[str], dict[str, Any] | None]:
    """Cross-bind product cases and recompute metrics with the trusted evaluator."""

    violations: list[str] = []
    required_fields = {
        "schema_version",
        "portfolio_kind",
        "portfolio_id",
        "claim_boundary",
        "split_roles",
        "selection_policy",
        "protected_policy",
        "portfolio_lineage",
        "cases",
    }
    if set(payload) != required_fields:
        violations.append("raw_predictions:portfolio_fields")
    if payload.get("schema_version") != 2:
        violations.append("raw_predictions:schema_version")
    if payload.get("portfolio_kind") != "governed_product_baseline":
        violations.append("raw_predictions:portfolio_kind")
    if payload.get("portfolio_id") != manifest.get("baseline_id"):
        violations.append("raw_predictions:portfolio_id")
    claim = payload.get("claim_boundary")
    claim_lower = claim.lower() if isinstance(claim, str) else ""
    if "governed product baseline" not in claim_lower or "synthetic" in claim_lower:
        violations.append("raw_predictions:claim_boundary")
    if (
        payload.get("selection_policy")
        != "frozen_validation_calibration_only_no_protected_selection"
    ):
        violations.append("raw_predictions:selection_policy")
    if payload.get("split_roles") != ["test", "background-test"]:
        violations.append("raw_predictions:split_roles")
    if payload.get("protected_policy") != EXPECTED_PROTECTED_POLICY:
        violations.append("raw_predictions:protected_policy")
    try:
        source_path = (
            portfolio_path.resolve().relative_to(repo_root.resolve()).as_posix()
        )
    except (OSError, ValueError):
        source_path = str(portfolio_path)
    expected_lineage = {
        "origin": "governed_product_inference",
        "source_path": source_path,
        "version": manifest.get("baseline_id"),
        "active_model_sha256": manifest.get("active_model_sha256"),
        "configuration_sha256": manifest.get("configuration_sha256"),
        "protected_split_sha256": protected_split_hash,
        "authoritative_reference_sha256": authority_reference_hash,
        "inference_evidence_sha256": inference_evidence_hash,
    }
    if payload.get("portfolio_lineage") != expected_lineage:
        violations.append("raw_predictions:portfolio_lineage")

    expected_ids = {
        sample_id
        for sample_id, sample in protected_samples.items()
        if sample.get("split") in EVALUATED_PROTECTED_SPLIT_ROLES
    }
    cases = payload.get("cases")
    cases_by_id: dict[str, dict[str, Any]] = {}
    if not isinstance(cases, list) or not cases:
        violations.append("raw_predictions:cases")
        cases = []
    for index, case in enumerate(cases):
        prefix = f"raw_predictions:case:{index}"
        if not isinstance(case, dict):
            violations.append(f"{prefix}:not_object")
            continue
        sample_id = case.get("sample_id")
        if (
            not isinstance(sample_id, str)
            or sample_id not in expected_ids
            or sample_id in cases_by_id
        ):
            violations.append(f"{prefix}:sample_id")
            continue
        cases_by_id[sample_id] = case
        sample = protected_samples[sample_id]
        for field in ("task", "split"):
            if case.get(field) != sample.get(field):
                violations.append(f"{prefix}:{field}_mismatch")
        metadata = case.get("metadata")
        if not isinstance(metadata, dict):
            violations.append(f"{prefix}:metadata")
        else:
            for dimension, field in REQUIRED_SUBGROUP_DIMENSION_FIELDS.items():
                if metadata.get(field) != sample.get("subgroups", {}).get(dimension):
                    violations.append(f"{prefix}:metadata:{field}_mismatch")
        parameters = configuration.get("parameters_by_task")
        expected_config = (
            parameters.get(sample.get("task")) if isinstance(parameters, dict) else None
        )
        if case.get("config") != expected_config:
            violations.append(f"{prefix}:configuration_mismatch")
        if canonical_hash(case) != sample.get("case_input_sha256"):
            violations.append(f"{prefix}:case_input_sha256_mismatch")
    if set(cases_by_id) != expected_ids:
        violations.append(
            "raw_predictions:sample_coverage:"
            f"missing={','.join(sorted(expected_ids - set(cases_by_id)))}:"
            f"unexpected={','.join(sorted(set(cases_by_id) - expected_ids))}"
        )
    if {
        case.get("task") for case in cases_by_id.values()
    } != REQUIRED_EVALUATOR_TASK_FAMILIES:
        violations.append("raw_predictions:evaluator_task_family_coverage")

    evaluation: dict[str, Any] | None = None
    try:
        evaluation = evaluate_cases(portfolio_path, expected_ids)
    except (OSError, ValueError, TypeError, KeyError) as exc:
        violations.append(
            f"raw_predictions:evaluator_rejected:{type(exc).__name__}:{str(exc)}"
        )
    if evaluation is not None:
        if evaluation.get("portfolio_kind") != "governed_product_baseline":
            violations.append("raw_predictions:evaluation_portfolio_kind")
        if (
            set(evaluation.get("evaluated_task_families") or [])
            != REQUIRED_EVALUATOR_TASK_FAMILIES
        ):
            violations.append("raw_predictions:evaluation_task_family_coverage")
        if evaluation.get("task_count") != len(REQUIRED_EVALUATOR_TASK_FAMILIES):
            violations.append("raw_predictions:evaluation_task_count")
        if evaluation.get("case_count") != len(expected_ids):
            violations.append("raw_predictions:evaluation_case_count")
        if evaluation.get("task_inventory") != task_inventory():
            violations.append("raw_predictions:evaluation_task_inventory")
        for result in evaluation.get("results", []):
            if not isinstance(result, dict):
                violations.append("raw_predictions:evaluation_result_not_object")
                continue
            sample_id = result.get("sample_id")
            raw = result.get("raw") if isinstance(result.get("raw"), dict) else {}
            sample = protected_samples.get(str(sample_id), {})
            if raw.get("input_sha256") != sample.get("case_input_sha256"):
                violations.append(
                    f"raw_predictions:evaluation_input_sha256_mismatch:{sample_id}"
                )
    return violations, evaluation


def _append_metric_numeric_sanity_violations(
    value: Any,
    violations: list[str],
    path: str = "metric_report",
) -> None:
    """Reject impossible reported numbers before exact recomputation comparison."""

    if isinstance(value, dict):
        for key, child in value.items():
            child_path = f"{path}.{key}"
            if isinstance(child, bool):
                continue
            if isinstance(child, (int, float)):
                numeric = float(child)
                if not math.isfinite(numeric):
                    violations.append(f"metric_report:non_finite:{child_path}")
                    continue
                normalized = key.lower()
                count_like = (
                    normalized in {"support", "case_support", "minimum_case_support"}
                    or normalized.endswith("_count")
                    or normalized.endswith("_pixels")
                    or normalized
                    in {"true_positive", "false_positive", "false_negative"}
                )
                if count_like and (
                    numeric < 0.0
                    or (".macro." not in child_path and type(child) is not int)
                ):
                    violations.append(
                        f"metric_report:invalid_support_or_count:{child_path}"
                    )
                rate_like = normalized in {
                    "precision",
                    "recall",
                    "f1",
                    "iou",
                    "mean_iou",
                    "dice",
                    "mean_dice",
                    "boundary_f1",
                    "mean_boundary_f1",
                    "accuracy",
                    "prediction_coverage",
                    "reference_coverage",
                    "coverage",
                    "retained_prediction_coverage",
                    "risk",
                    "false_discovery_rate",
                    "miss_rate",
                    "ap50",
                    "ap50_95",
                    "map50",
                    "map50_95",
                    "ece",
                    "brier",
                }
                if rate_like and not 0.0 <= numeric <= 1.0:
                    violations.append(f"metric_report:impossible_rate:{child_path}")
            _append_metric_numeric_sanity_violations(child, violations, child_path)
        lower = value.get("lower")
        upper = value.get("upper")
        support = value.get("support")
        if isinstance(support, bool) or (
            support is not None and (not isinstance(support, int) or support < 0)
        ):
            violations.append(f"metric_report:invalid_ci_support:{path}")
        if (
            isinstance(lower, (int, float))
            and not isinstance(lower, bool)
            and isinstance(upper, (int, float))
            and not isinstance(upper, bool)
        ):
            if float(lower) > float(upper):
                violations.append(f"metric_report:invalid_ci_order:{path}")
            if "wilson" in path.lower() and (float(lower) < 0.0 or float(upper) > 1.0):
                violations.append(f"metric_report:impossible_wilson_ci:{path}")
    elif isinstance(value, list):
        for index, child in enumerate(value):
            _append_metric_numeric_sanity_violations(
                child, violations, f"{path}[{index}]"
            )
    elif isinstance(value, float) and not math.isfinite(value):
        violations.append(f"metric_report:non_finite:{path}")


def _validate_metric_report(
    payload: dict[str, Any],
    manifest: dict[str, Any],
    protected_samples: dict[str, dict[str, Any]],
    protected_split_hash: Any,
    raw_predictions_hash: Any,
    evaluation: dict[str, Any] | None,
    configuration: dict[str, Any],
) -> tuple[list[str], dict[str, Any]]:
    """Require exact equality with an in-process evaluator recomputation."""

    violations: list[str] = []
    _append_metric_numeric_sanity_violations(payload, violations)
    required_fields = {
        "schema_version",
        "artifact_role",
        "active_model_sha256",
        "configuration_sha256",
        "evaluator_sha256",
        "protected_split_sha256",
        "raw_predictions_sha256",
        "evaluator_version",
        "portfolio_kind",
        "portfolio_id",
        "portfolio_file_sha256",
        "portfolio_canonical_json_sha256",
        "evaluated_task_families",
        "task_count",
        "case_count",
        "task_inventory",
        "task_inventory_sha256",
        "results",
        "results_canonical_json_sha256",
        "portfolio_metrics",
        "portfolio_metrics_canonical_json_sha256",
        "subgroups",
        "subgroups_canonical_json_sha256",
        "failures",
        "failures_canonical_json_sha256",
        "failure_taxonomy",
        "failure_taxonomy_canonical_json_sha256",
        "subgroup_dimension_mapping",
        "subgroup_release_policy",
        "pre_registered_targets",
        "pre_registered_targets_sha256",
    }
    if set(payload) != required_fields:
        violations.append("metric_report:fields")
    if payload.get("schema_version") != 3:
        violations.append("metric_report:schema_version")
    if payload.get("artifact_role") != "protected_metric_report":
        violations.append("metric_report:artifact_role")
    bindings = {
        "active_model_sha256": manifest.get("active_model_sha256"),
        "configuration_sha256": manifest.get("configuration_sha256"),
        "evaluator_sha256": manifest.get("evaluator_sha256"),
        "protected_split_sha256": protected_split_hash,
        "raw_predictions_sha256": raw_predictions_hash,
    }
    for field, expected in bindings.items():
        if payload.get(field) != expected:
            violations.append(f"metric_report:{field}_mismatch")
    if evaluation is None:
        violations.append("metric_report:evaluator_result_unavailable")
        return violations, {
            "required_dimensions": sorted(REQUIRED_SUBGROUP_DIMENSIONS),
            "policy": SUBGROUP_RELEASE_POLICY,
            "status": "not_evaluable",
        }

    exact_fields = (
        "evaluator_version",
        "portfolio_kind",
        "portfolio_id",
        "portfolio_file_sha256",
        "portfolio_canonical_json_sha256",
        "evaluated_task_families",
        "task_count",
        "case_count",
        "task_inventory",
        "results",
        "results_canonical_json_sha256",
        "portfolio_metrics",
        "subgroups",
        "failures",
        "failure_taxonomy",
    )
    for field in exact_fields:
        if payload.get(field) != evaluation.get(field):
            violations.append(f"metric_report:{field}_recomputation_mismatch")
    for hash_field, value_field in {
        "task_inventory_sha256": "task_inventory",
        "portfolio_metrics_canonical_json_sha256": "portfolio_metrics",
        "subgroups_canonical_json_sha256": "subgroups",
        "failures_canonical_json_sha256": "failures",
        "failure_taxonomy_canonical_json_sha256": "failure_taxonomy",
    }.items():
        if payload.get(hash_field) != canonical_hash(evaluation.get(value_field)):
            violations.append(f"metric_report:{hash_field}")
    results = payload.get("results")
    if not isinstance(results, list) or payload.get(
        "results_canonical_json_sha256"
    ) != canonical_hash(results):
        violations.append("metric_report:results_sha256")
    if (
        set(payload.get("evaluated_task_families") or [])
        != REQUIRED_EVALUATOR_TASK_FAMILIES
    ):
        violations.append("metric_report:evaluator_task_family_coverage")
    expected_ids = {
        sample_id
        for sample_id, sample in protected_samples.items()
        if sample.get("split") in EVALUATED_PROTECTED_SPLIT_ROLES
    }
    result_ids = {
        item.get("sample_id")
        for item in payload.get("results", [])
        if isinstance(item, dict)
    }
    if result_ids != expected_ids:
        violations.append("metric_report:sample_coverage")

    reported_portfolio_metrics = payload.get("portfolio_metrics")
    if isinstance(reported_portfolio_metrics, dict):
        for task, metrics in reported_portfolio_metrics.items():
            support = (
                metrics.get("observation_support")
                if isinstance(metrics, dict)
                else None
            )
            support_values = (
                [value for value in support.values() if type(value) is int]
                if isinstance(support, dict)
                else []
            )
            if not support_values or sum(support_values) <= 0:
                violations.append(f"metric_report:task:{task}:empty_support")

    if payload.get("subgroup_dimension_mapping") != REQUIRED_SUBGROUP_DIMENSION_FIELDS:
        violations.append("metric_report:subgroup_dimension_mapping")
    if payload.get("subgroup_release_policy") != SUBGROUP_RELEASE_POLICY:
        violations.append("metric_report:subgroup_release_policy")
    subgroups = evaluation.get("subgroups") or {}
    dimension_reports = subgroups.get("dimensions")
    if not isinstance(dimension_reports, dict) or set(dimension_reports) != set(
        REQUIRED_SUBGROUP_DIMENSION_FIELDS.values()
    ):
        violations.append("metric_report:subgroup_dimension_coverage")
        dimension_reports = {}
    for dimension, evaluator_field in REQUIRED_SUBGROUP_DIMENSION_FIELDS.items():
        report = dimension_reports.get(evaluator_field)
        if not isinstance(report, dict):
            violations.append(f"metric_report:subgroup:{dimension}:missing")
            continue
        strata = report.get("strata")
        if (
            not isinstance(strata, dict)
            or len(strata)
            < SUBGROUP_RELEASE_POLICY["minimum_distinct_strata_per_dimension"]
        ):
            violations.append(f"metric_report:subgroup:{dimension}:not_stratified")
            continue
        for stratum, evidence in strata.items():
            task_metrics = (
                evidence.get("task_metrics") if isinstance(evidence, dict) else None
            )
            if (
                not isinstance(task_metrics, dict)
                or set(task_metrics) != REQUIRED_EVALUATOR_TASK_FAMILIES
            ):
                violations.append(
                    f"metric_report:subgroup:{dimension}:{stratum}:task_coverage"
                )
                continue
            for task, task_metric in task_metrics.items():
                if (
                    not isinstance(task_metric, dict)
                    or task_metric.get("status") != "evaluable"
                    or not isinstance(task_metric.get("case_support"), int)
                    or task_metric["case_support"] < SUBGROUP_MIN_CASE_SUPPORT
                    or (task_metric.get("primary_metric") or {}).get("value") is None
                ):
                    violations.append(
                        f"metric_report:subgroup:{dimension}:{stratum}:{task}:support"
                    )
        worst = report.get("worst_stratum_by_task")
        if (
            not isinstance(worst, dict)
            or set(worst) != REQUIRED_EVALUATOR_TASK_FAMILIES
        ):
            violations.append(f"metric_report:subgroup:{dimension}:worst_strata")
        elif any(
            not isinstance(item, dict) or item.get("status") != "computed"
            for item in worst.values()
        ):
            violations.append(f"metric_report:subgroup:{dimension}:worst_not_computed")

    portfolio_metrics = evaluation.get("portfolio_metrics")
    if (
        not isinstance(portfolio_metrics, dict)
        or set(portfolio_metrics) != REQUIRED_EVALUATOR_TASK_FAMILIES
    ):
        violations.append("metric_report:portfolio_metric_task_coverage")
        portfolio_metrics = {}
    for task, metrics in portfolio_metrics.items():
        support = (
            metrics.get("observation_support") if isinstance(metrics, dict) else {}
        )
        support_values = (
            [value for value in support.values() if type(value) is int]
            if isinstance(support, dict)
            else []
        )
        if not support_values or sum(support_values) <= 0:
            violations.append(f"metric_report:task:{task}:empty_support")
        if (
            type(metrics.get("case_support")) is not int
            or metrics["case_support"] < SUBGROUP_MIN_CASE_SUPPORT * 2
        ):
            violations.append(f"metric_report:task:{task}:case_support")

    targets = configuration.get("subgroup_release_targets")
    if payload.get("pre_registered_targets") != targets:
        violations.append("metric_report:pre_registered_targets")
    if payload.get("pre_registered_targets_sha256") != canonical_hash(targets):
        violations.append("metric_report:pre_registered_targets_sha256")
    if (
        not isinstance(targets, dict)
        or set(targets) != REQUIRED_EVALUATOR_TASK_FAMILIES
    ):
        violations.append("metric_report:release_target_coverage")
    else:
        for task, target in targets.items():
            primary = (portfolio_metrics.get(task) or {}).get("primary_metric") or {}
            if not isinstance(target, dict) or set(target) != {
                "metric",
                "direction",
                "threshold",
            }:
                violations.append(f"metric_report:release_target:{task}:contract")
                continue
            if target.get("metric") != primary.get("name"):
                violations.append(f"metric_report:release_target:{task}:metric")
            if target.get("direction") != primary.get("direction"):
                violations.append(f"metric_report:release_target:{task}:direction")
            threshold = target.get("threshold")
            current = primary.get("value")
            if (
                isinstance(threshold, bool)
                or not isinstance(threshold, (int, float))
                or not math.isfinite(float(threshold))
                or float(threshold) < 0.0
            ):
                violations.append(f"metric_report:release_target:{task}:threshold")
                continue
            if (
                isinstance(current, bool)
                or not isinstance(current, (int, float))
                or not math.isfinite(float(current))
            ):
                violations.append(f"metric_report:release_target:{task}:not_evaluable")
                continue
            direction = target.get("direction")
            if (
                direction == "higher_is_better" and float(current) < float(threshold)
            ) or (direction == "lower_is_better" and float(current) > float(threshold)):
                violations.append(f"metric_report:release_target_not_met:{task}")

    return violations, {
        "required_dimensions": sorted(REQUIRED_SUBGROUP_DIMENSIONS),
        "dimension_mapping": REQUIRED_SUBGROUP_DIMENSION_FIELDS,
        "policy": SUBGROUP_RELEASE_POLICY,
        "task_families": sorted(REQUIRED_EVALUATOR_TASK_FAMILIES),
        "evaluator_version": EVALUATOR_VERSION,
        "subgroups_sha256": canonical_hash(subgroups),
        "portfolio_metrics_sha256": canonical_hash(portfolio_metrics),
    }


def _validate_human_review_ledger(
    payload: dict[str, Any],
    protected_samples: dict[str, dict[str, Any]],
    protected_split_hash: Any,
    raw_predictions_hash: Any,
) -> tuple[list[str], dict[str, Any]]:
    violations: list[str] = []
    if payload.get("schema_version") != 1:
        violations.append("human_review:schema_version")
    if payload.get("artifact_role") != "human_review_ledger":
        violations.append("human_review:artifact_role")
    if payload.get("protected_split_sha256") != protected_split_hash:
        violations.append("human_review:protected_split_hash_mismatch")
    if payload.get("raw_predictions_sha256") != raw_predictions_hash:
        violations.append("human_review:raw_predictions_hash_mismatch")
    expected_ids = {
        sample_id
        for sample_id, sample in protected_samples.items()
        if sample.get("split") in EVALUATED_PROTECTED_SPLIT_ROLES
    }
    entries = payload.get("entries")
    if not isinstance(entries, list):
        violations.append("human_review:entries")
        entries = []
    elif payload.get("entries_canonical_json_sha256") != canonical_hash(entries):
        violations.append("human_review:entries_sha256")
    by_id: dict[str, dict[str, Any]] = {}
    fields = {
        "sample_id",
        "reviewer_id",
        "review_timestamp",
        "decision",
        "label_sha256",
        "case_input_sha256",
        "entry_canonical_json_sha256",
    }
    for index, entry in enumerate(entries):
        prefix = f"human_review:entry:{index}"
        if not isinstance(entry, dict) or set(entry) != fields:
            violations.append(f"{prefix}:contract")
            continue
        sample_id = entry.get("sample_id")
        if (
            not isinstance(sample_id, str)
            or sample_id not in expected_ids
            or sample_id in by_id
        ):
            violations.append(f"{prefix}:sample_id")
            continue
        by_id[sample_id] = entry
        sample = protected_samples[sample_id]
        if (
            not isinstance(entry.get("reviewer_id"), str)
            or not entry["reviewer_id"].strip()
            or entry.get("decision") != "accepted"
            or parse_timestamp(entry.get("review_timestamp")) is None
        ):
            violations.append(f"{prefix}:human_acceptance")
        if entry.get("label_sha256") != sample.get("label_sha256"):
            violations.append(f"{prefix}:label_sha256")
        if entry.get("case_input_sha256") != sample.get("case_input_sha256"):
            violations.append(f"{prefix}:case_input_sha256")
        hash_input = {
            key: value
            for key, value in entry.items()
            if key != "entry_canonical_json_sha256"
        }
        if entry.get("entry_canonical_json_sha256") != canonical_hash(hash_input):
            violations.append(f"{prefix}:entry_sha256")
    if set(by_id) != expected_ids:
        violations.append("human_review:sample_coverage")
    return violations, {
        "reviewed_sample_count": len(by_id),
        "required_sample_count": len(expected_ids),
        "ledger_sha256": canonical_hash(entries),
    }


def _validate_geometric_leakage_audit(
    payload: dict[str, Any],
    manifest: dict[str, Any],
    protected_samples: dict[str, dict[str, Any]],
    protected_split_hash: Any,
    raw_predictions_hash: Any,
) -> tuple[list[str], dict[str, Any]]:
    violations: list[str] = []
    if payload.get("schema_version") != 1:
        violations.append("geometric_leakage:schema_version")
    if payload.get("artifact_role") != "geometric_leakage_audit":
        violations.append("geometric_leakage:artifact_role")
    for field, expected in {
        "protected_split_sha256": protected_split_hash,
        "raw_predictions_sha256": raw_predictions_hash,
        "development_split_manifest_sha256": manifest.get(
            "development_split_manifest_sha256"
        ),
    }.items():
        if payload.get(field) != expected:
            violations.append(f"geometric_leakage:{field}_mismatch")
    expected_ids = {
        sample_id
        for sample_id, sample in protected_samples.items()
        if sample.get("split") in EVALUATED_PROTECTED_SPLIT_ROLES
    }
    if payload.get("evaluation_sample_ids_sha256") != _canonical_id_hash(expected_ids):
        violations.append("geometric_leakage:evaluation_sample_ids_sha256")
    if payload.get("distance_threshold_m") != 2000:
        violations.append("geometric_leakage:distance_threshold_m")
    crs = payload.get("projected_crs")
    if not isinstance(crs, str) or not crs.strip() or "4326" in crs:
        violations.append("geometric_leakage:projected_crs")
    algorithm = payload.get("algorithm")
    if not isinstance(algorithm, str) or "geometry" not in algorithm.lower():
        violations.append("geometric_leakage:algorithm")
    if (
        payload.get("below_threshold_pairs") != []
        or payload.get("below_threshold_pair_count") != 0
    ):
        violations.append("geometric_leakage:pairs_below_2000_m")
    minimum_distance = payload.get("minimum_observed_distance_m")
    if (
        isinstance(minimum_distance, bool)
        or not isinstance(minimum_distance, (int, float))
        or not math.isfinite(float(minimum_distance))
        or float(minimum_distance) < 2000.0
    ):
        violations.append("geometric_leakage:minimum_observed_distance_m")
    return violations, {
        "distance_threshold_m": 2000,
        "below_threshold_pair_count": payload.get("below_threshold_pair_count"),
        "minimum_observed_distance_m": minimum_distance,
        "evaluation_sample_ids_sha256": _canonical_id_hash(expected_ids),
    }


def _validate_vault_access_evidence(
    payload: dict[str, Any],
    manifest: dict[str, Any],
    protected_samples: dict[str, dict[str, Any]],
    protected_split_hash: Any,
    raw_predictions_hash: Any,
) -> tuple[list[str], dict[str, Any]]:
    violations: list[str] = []
    if payload.get("schema_version") != 1:
        violations.append("vault_access:schema_version")
    if payload.get("artifact_role") != "vault_access_evidence":
        violations.append("vault_access:artifact_role")
    if payload.get("protected_split_sha256") != protected_split_hash:
        violations.append("vault_access:protected_split_hash_mismatch")
    if payload.get("raw_predictions_sha256") != raw_predictions_hash:
        violations.append("vault_access:raw_predictions_hash_mismatch")
    inference = (
        manifest.get("inference") if isinstance(manifest.get("inference"), dict) else {}
    )
    if payload.get("execution_id") != inference.get("execution_id"):
        violations.append("vault_access:execution_id_mismatch")
    if payload.get("vault_mode") != "read_only_evaluation":
        violations.append("vault_access:vault_mode")
    if payload.get("challenge_labels_accessed") is not False:
        violations.append("vault_access:challenge_labels_accessed")
    expected_ids = {
        sample_id
        for sample_id, sample in protected_samples.items()
        if sample.get("split") in EVALUATED_PROTECTED_SPLIT_ROLES
    }
    challenge_ids = {
        sample_id
        for sample_id, sample in protected_samples.items()
        if sample.get("split") == "challenge"
    }
    entries = payload.get("access_log")
    if not isinstance(entries, list) or not entries:
        violations.append("vault_access:access_log")
        entries = []
    elif payload.get("access_log_canonical_json_sha256") != canonical_hash(entries):
        violations.append("vault_access:access_log_sha256")
    previous = "0" * 64
    accessed: set[str] = set()
    fields = {
        "sequence",
        "timestamp",
        "actor",
        "purpose",
        "operation",
        "sample_ids",
        "previous_entry_sha256",
        "entry_sha256",
    }
    for index, entry in enumerate(entries, start=1):
        prefix = f"vault_access:entry:{index}"
        if not isinstance(entry, dict) or set(entry) != fields:
            violations.append(f"{prefix}:contract")
            continue
        if entry.get("sequence") != index:
            violations.append(f"{prefix}:sequence")
        if parse_timestamp(entry.get("timestamp")) is None:
            violations.append(f"{prefix}:timestamp")
        if entry.get("actor") != "phase4-evaluator":
            violations.append(f"{prefix}:actor")
        if entry.get("purpose") != "evaluation_only":
            violations.append(f"{prefix}:purpose")
        if entry.get("operation") != "read":
            violations.append(f"{prefix}:operation")
        sample_ids = entry.get("sample_ids")
        if (
            not isinstance(sample_ids, list)
            or any(not isinstance(item, str) for item in sample_ids)
            or len(sample_ids) != len(set(sample_ids))
        ):
            violations.append(f"{prefix}:sample_ids")
            sample_ids = []
        if set(sample_ids) & challenge_ids:
            violations.append(f"{prefix}:challenge_access")
        if set(sample_ids) - expected_ids:
            violations.append(f"{prefix}:unexpected_sample")
        accessed.update(sample_ids)
        if entry.get("previous_entry_sha256") != previous:
            violations.append(f"{prefix}:previous_entry_sha256")
        hash_input = {
            key: value for key, value in entry.items() if key != "entry_sha256"
        }
        observed_hash = canonical_hash(hash_input)
        if entry.get("entry_sha256") != observed_hash:
            violations.append(f"{prefix}:entry_sha256")
        previous = observed_hash
    if accessed != expected_ids:
        violations.append("vault_access:sample_coverage")
    return violations, {
        "vault_mode": payload.get("vault_mode"),
        "accessed_sample_count": len(accessed),
        "required_sample_count": len(expected_ids),
        "access_log_sha256": canonical_hash(entries),
        "final_chain_sha256": previous,
    }


def _unavailable_product_baseline_gate(
    status: str,
    reason: str,
    **details: Any,
) -> dict[str, Any]:
    if status not in {"fail", "not_evaluable"}:
        raise ValueError("Unavailable product gates can only fail or be not_evaluable")
    derived = {
        name: {"status": status, "reason": reason}
        for name in (
            "authoritative_reference_portfolio_available",
            "representative_product_subgroup_support",
            "human_review_complete",
            "split_independence",
            "protected_storage_isolation",
        )
    }
    return {
        "status": status,
        "reason": reason,
        "checked_artifacts": [],
        "derived_gates": derived,
        **details,
    }


def product_baseline_manifest_gate(
    repo_root: Path,
    manifest_path: Path,
    active_model: dict[str, Any],
    authority_requirements: list[dict[str, Any]] | None = None,
    runtime_probe: Callable[[], dict[str, Any]] | None = None,
) -> dict[str, Any]:
    governed_root = (repo_root / "artifacts/evidence/accuracy/P4").resolve()
    resolved_manifest = manifest_path.resolve()
    try:
        relative = resolved_manifest.relative_to(repo_root.resolve())
        resolved_manifest.relative_to(governed_root)
    except (OSError, ValueError):
        return _unavailable_product_baseline_gate(
            "fail",
            "Product baseline manifest must reside inside artifacts/evidence/accuracy/P4.",
            path=str(manifest_path),
        )
    if not resolved_manifest.is_file():
        return _unavailable_product_baseline_gate(
            "not_evaluable",
            "No executed, hash-bound product incumbent baseline manifest is available.",
            expected_path=relative.as_posix(),
        )
    try:
        manifest = json.loads(resolved_manifest.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        return _unavailable_product_baseline_gate(
            "fail",
            f"Unreadable product baseline manifest: {exc}",
            path=relative.as_posix(),
        )
    if not isinstance(manifest, dict):
        return _unavailable_product_baseline_gate(
            "fail",
            "Product baseline manifest root must be an object.",
            path=relative.as_posix(),
        )

    evaluator_path = repo_root / "scripts/accuracy_phase4_evaluator.py"
    if not evaluator_path.is_file():
        return _unavailable_product_baseline_gate(
            "fail",
            "The hash-bound Phase-4 evaluator is unavailable.",
            path=relative.as_posix(),
        )
    evaluator_hash = sha256(evaluator_path)
    model_gate = active_model_availability_gate(active_model)
    manifest_violations = _validate_product_manifest_contract(
        manifest, active_model, evaluator_hash
    )
    artifact_violations: list[str] = []
    payloads, checked_artifacts = _load_governed_artifacts(
        repo_root, governed_root, manifest, artifact_violations
    )

    def descriptor(role: str) -> dict[str, Any]:
        value = manifest.get(role)
        return value if isinstance(value, dict) else {}

    if manifest.get("configuration_sha256") != descriptor("configuration").get(
        "sha256"
    ):
        artifact_violations.append("configuration:manifest_hash_binding")
    configuration = payloads.get("configuration", {})
    configuration_violations = (
        _validate_configuration(configuration, manifest) if configuration else []
    )
    protected_violations: list[str] = []
    protected_samples: dict[str, dict[str, Any]] = {}
    if "protected_split_manifest" in payloads:
        protected_violations, protected_samples = _validate_protected_split(
            payloads["protected_split_manifest"]
        )
    protected_hash = descriptor("protected_split_manifest").get("sha256")
    raw_hash = descriptor("raw_predictions").get("sha256")

    probe = runtime_probe or probe_local_cuda_runtime
    try:
        runtime_observation = probe()
    except (
        Exception
    ) as exc:  # pragma: no cover - defensive boundary around hardware probe
        runtime_observation = {
            "status": "not_evaluable",
            "reason": f"Independent CUDA probe raised {type(exc).__name__}.",
        }
    if not isinstance(runtime_observation, dict):
        runtime_observation = {
            "status": "not_evaluable",
            "reason": "Independent CUDA probe returned no structured evidence.",
        }
    inference_violations: list[str] = []
    runtime_status = str(runtime_observation.get("status"))
    if "inference_evidence" in payloads:
        inference_violations, runtime_status, runtime_observation = (
            _validate_inference_evidence(
                payloads["inference_evidence"],
                manifest,
                configuration,
                protected_samples,
                protected_hash,
                runtime_observation,
            )
        )

    raw_violations: list[str] = []
    evaluation: dict[str, Any] | None = None
    if "raw_predictions" in payloads:
        raw_descriptor = descriptor("raw_predictions")
        raw_path_value = raw_descriptor.get("path")
        raw_path = repo_root / str(raw_path_value or "")
        raw_violations, evaluation = _evaluate_raw_product_portfolio(
            repo_root,
            raw_path,
            payloads["raw_predictions"],
            manifest,
            configuration,
            protected_samples,
            protected_hash,
            descriptor("authoritative_reference_manifest").get("sha256"),
            descriptor("inference_evidence").get("sha256"),
        )

    requirements = authority_requirements or [
        dict(item) for item in REQUIRED_AUTHORITY_REQUIREMENTS
    ]
    authority_violations: list[str] = []
    authority_evidence: dict[str, Any] = {"required": requirements, "observed": []}
    if "authoritative_reference_manifest" in payloads:
        authority_violations, authority_evidence = (
            _validate_authoritative_reference_portfolio(
                payloads["authoritative_reference_manifest"],
                protected_samples,
                protected_hash,
                requirements,
                evaluation,
            )
        )

    metric_violations: list[str] = []
    subgroup_evidence: dict[str, Any] = {
        "required_dimensions": sorted(REQUIRED_SUBGROUP_DIMENSIONS),
        "status": "not_evaluable",
    }
    if "metric_report" in payloads:
        metric_violations, subgroup_evidence = _validate_metric_report(
            payloads["metric_report"],
            manifest,
            protected_samples,
            protected_hash,
            raw_hash,
            evaluation,
            configuration,
        )

    review_violations: list[str] = []
    review_evidence: dict[str, Any] = {}
    if "human_review_ledger" in payloads:
        review_violations, review_evidence = _validate_human_review_ledger(
            payloads["human_review_ledger"], protected_samples, protected_hash, raw_hash
        )
    leakage_violations: list[str] = []
    leakage_evidence: dict[str, Any] = {}
    if "geometric_leakage_audit" in payloads:
        leakage_violations, leakage_evidence = _validate_geometric_leakage_audit(
            payloads["geometric_leakage_audit"],
            manifest,
            protected_samples,
            protected_hash,
            raw_hash,
        )
    vault_violations: list[str] = []
    vault_evidence: dict[str, Any] = {}
    if "vault_access_evidence" in payloads:
        vault_violations, vault_evidence = _validate_vault_access_evidence(
            payloads["vault_access_evidence"],
            manifest,
            protected_samples,
            protected_hash,
            raw_hash,
        )

    model_violations = (
        [f"active_model:{model_gate.get('reason')}"]
        if model_gate.get("status") == "fail"
        else []
    )
    categories = {
        "active_model_integrity": model_violations,
        "manifest_contract": manifest_violations,
        "artifact_integrity": artifact_violations,
        "configuration": configuration_violations,
        "protected_split": protected_violations,
        "cuda_inference": inference_violations,
        "authoritative_reference": authority_violations,
        "raw_predictions_and_recomputation": raw_violations,
        "metrics_and_subgroups": metric_violations,
        "human_review_ledger": review_violations,
        "geometric_leakage_audit": leakage_violations,
        "vault_access_evidence": vault_violations,
    }
    all_violations = sorted({item for values in categories.values() for item in values})
    blockers: list[str] = []
    if model_gate.get("status") == "not_evaluable":
        blockers.append(
            str(model_gate.get("reason") or "Active model bytes are unavailable.")
        )
    if runtime_status == "not_evaluable":
        blockers.append(
            str(runtime_observation.get("reason") or "CUDA runtime is unavailable.")
        )
    common = sorted(
        {
            *model_violations,
            *manifest_violations,
            *artifact_violations,
            *configuration_violations,
            *protected_violations,
            *inference_violations,
        }
    )

    def derived(own: list[str], evidence: dict[str, Any]) -> dict[str, Any]:
        gate_violations = sorted({*common, *own})
        status = "fail" if gate_violations else "not_evaluable" if blockers else "pass"
        return {
            "status": status,
            "violations": gate_violations,
            "blockers": blockers,
            "evidence": evidence,
        }

    derived_gates = {
        "authoritative_reference_portfolio_available": derived(
            authority_violations, authority_evidence
        ),
        "representative_product_subgroup_support": derived(
            [*raw_violations, *metric_violations], subgroup_evidence
        ),
        "human_review_complete": derived(review_violations, review_evidence),
        "split_independence": derived(leakage_violations, leakage_evidence),
        "protected_storage_isolation": derived(vault_violations, vault_evidence),
    }
    status = "fail" if all_violations else "not_evaluable" if blockers else "pass"
    return {
        "status": status,
        "reason": "; ".join(blockers) if blockers else None,
        "path": relative.as_posix(),
        "manifest_sha256": sha256(resolved_manifest),
        "violations": all_violations,
        "blockers": blockers,
        "active_model_observation": model_gate,
        "runtime_observation": runtime_observation,
        "evaluation_results_sha256": evaluation.get("results_canonical_json_sha256")
        if evaluation
        else None,
        "validation_checks": {
            name: {
                "status": "fail" if values else "pass",
                "violations": sorted(set(values)),
            }
            for name, values in categories.items()
        },
        "checked_artifacts": checked_artifacts,
        "derived_gates": derived_gates,
        "evidence": (
            "Governed product evidence was independently model/runtime-verified and metrics were recomputed in-process."
            if status == "pass"
            else None
        ),
    }


def readiness_snapshot(repo_root: Path) -> dict[str, Any]:
    status_path = repo_root / "fixtures/accuracy/readiness/status.json"
    p3_path = repo_root / "fixtures/accuracy/readiness/full-scan-manifest.json"
    leakage_path = repo_root / "fixtures/accuracy/readiness/leakage-report.json"
    status = json.loads(status_path.read_text(encoding="utf-8"))
    p3 = json.loads(p3_path.read_text(encoding="utf-8"))
    leakage = json.loads(leakage_path.read_text(encoding="utf-8"))
    status_projection = {
        "schema_version": 1,
        "runtime": {"active_model": (status.get("runtime") or {}).get("active_model")},
    }
    return {
        "schema_version": 1,
        "source_paths": {
            "accuracy_status_projection": {
                "path": "fixtures/accuracy/readiness/status.json",
                "selected_json_pointers": ["/runtime/active_model"],
                "sha256": canonical_hash(status_projection),
            },
            "phase3_full_scan": repository_file(
                repo_root, "fixtures/accuracy/readiness/full-scan-manifest.json"
            ),
            "phase3_leakage": repository_file(
                repo_root, "fixtures/accuracy/readiness/leakage-report.json"
            ),
        },
        "active_model": status_projection["runtime"]["active_model"],
        "phase3_scan": {
            "scan_id": p3.get("scan_id"),
            "content_hash": p3.get("content_hash"),
            "grb_consistency": p3.get("grb_consistency"),
        },
        "phase3_leakage_status": leakage.get("status"),
        "authority_requirements": [
            dict(item) for item in REQUIRED_AUTHORITY_REQUIREMENTS
        ],
    }


def active_model_availability_gate(active_model: dict[str, Any]) -> dict[str, Any]:
    configured_path_value = active_model.get("path")
    configured_hash = active_model.get("sha256")
    result: dict[str, Any] = {
        "configured_path": str(configured_path_value or ""),
        "configured_sha256": configured_hash,
    }
    if not isinstance(configured_path_value, str) or not configured_path_value.strip():
        return {
            **result,
            "status": "fail",
            "reason": "Configured active model path is missing.",
        }
    if not is_sha256(configured_hash):
        return {
            **result,
            "status": "fail",
            "reason": "Configured active model SHA-256 is missing or invalid.",
        }
    configured_path = Path(configured_path_value)
    if not configured_path.is_file():
        return {
            **result,
            "status": "not_evaluable",
            "reason": "Configured active model is not locally accessible.",
        }
    observed_hash = sha256(configured_path)
    observed_size = configured_path.stat().st_size
    result.update(
        {
            "observed_sha256": observed_hash,
            "observed_size_bytes": observed_size,
        }
    )
    if observed_hash != configured_hash:
        return {
            **result,
            "status": "fail",
            "reason": "Configured active model checksum does not match the local model file.",
        }
    expected_size = active_model.get("size_bytes")
    if (
        not isinstance(expected_size, int)
        or isinstance(expected_size, bool)
        or expected_size != observed_size
    ):
        return {
            **result,
            "status": "fail",
            "reason": "Configured active model size does not match the local model file.",
        }
    return {**result, "status": "pass", "reason": None}


def product_gate_evidence(
    repo_root: Path,
    snapshot: dict[str, Any],
    product_baseline_manifest: Path,
    runtime_probe: Callable[[], dict[str, Any]] | None = None,
) -> dict[str, Any]:
    """Build mandatory product gates exclusively from governed artifacts."""

    active_model = snapshot.get("active_model") or {}
    baseline_gate = product_baseline_manifest_gate(
        repo_root,
        product_baseline_manifest,
        active_model,
        snapshot["authority_requirements"],
        runtime_probe=runtime_probe,
    )
    derived = baseline_gate.get("derived_gates") or {}
    fallback_status = (
        "fail" if baseline_gate.get("status") == "fail" else "not_evaluable"
    )
    fallback = {
        "status": fallback_status,
        "reason": baseline_gate.get("reason")
        or "A checksum-bound governed product artifact is unavailable.",
    }
    phase3_leakage = snapshot.get("phase3_leakage_status")
    return {
        "active_model_available_and_hash_verified": baseline_gate.get(
            "active_model_observation"
        )
        or active_model_availability_gate(active_model),
        "authoritative_reference_portfolio_available": derived.get(
            "authoritative_reference_portfolio_available", fallback
        ),
        "human_review_complete": derived.get("human_review_complete", fallback),
        "split_independence": derived.get("split_independence", fallback),
        "phase3_leakage_resolved": {
            "status": "pass"
            if phase3_leakage == "pass"
            else "not_evaluable"
            if phase3_leakage is None
            else "fail",
            "observed": phase3_leakage,
        },
        "protected_storage_isolation": derived.get(
            "protected_storage_isolation", fallback
        ),
        "executed_product_incumbent_baseline": baseline_gate,
        "representative_product_subgroup_support": derived.get(
            "representative_product_subgroup_support", fallback
        ),
    }


def build_release_gate_report(
    split_result: dict[str, Any],
    evaluation: dict[str, Any],
    portfolio: dict[str, Any],
    golden: dict[str, Any],
    firewall_checks: dict[str, bool],
    product_gates: dict[str, Any],
) -> dict[str, Any]:
    declared_families = {item["task"] for item in evaluation["task_inventory"]}
    observed_families = {item["task"] for item in evaluation["results"]}
    required_split_roles = {
        "train",
        "val",
        "calibration",
        "test",
        "background-test",
        "challenge",
    }
    observed_split_roles = set(split_result["leakage"]["split_counts"])
    required_raw_fields = {
        "references",
        "predictions_pre_filter",
        "predictions_post_filter",
        "config",
        "input_lineage",
        "portfolio_lineage",
    }
    raw_violations = [
        item["sample_id"]
        for item in evaluation["results"]
        if not required_raw_fields <= set(item.get("raw") or {})
    ]
    protected_policy = portfolio.get("protected_policy") or {}
    selection_contract_valid = (
        protected_policy.get("operating_point_selection_allowed") is False
        and protected_policy.get("diagnostic_curves_select_operating_point") is False
        and protected_policy.get("test_feedback_allowed") is False
        and protected_policy.get("threshold_selection_source")
        == "pre_registered_configuration_only"
        and all(
            isinstance((item.get("raw") or {}).get("config"), dict)
            for item in evaluation["results"]
        )
    )
    empty_case = next(
        (
            item
            for item in evaluation["results"]
            if item["sample_id"] == "background-test-pure-empty"
        ),
        None,
    )
    empty_metrics = (empty_case or {}).get("metrics") or {}
    null_semantics_valid = (
        empty_case is not None
        and empty_metrics.get("reference_count") == 0
        and empty_metrics.get("prediction_count") == 0
        and empty_metrics.get("precision") is None
        and empty_metrics.get("recall") is None
        and empty_metrics.get("f1") is None
    )
    subgroup_report = evaluation.get("subgroups") or {}
    subgroup_contract_valid = (
        subgroup_report.get("overall_status")
        in {"not_evaluable", "evaluable_no_release_target"}
        and isinstance(subgroup_report.get("dimensions"), dict)
        and bool(subgroup_report.get("dimensions"))
        and all(
            isinstance(dimension.get("strata"), dict)
            and isinstance(dimension.get("worst_stratum_by_task"), dict)
            for dimension in subgroup_report["dimensions"].values()
        )
    )
    capability_inventory = evaluation.get("task_inventory") or []
    capability_contract_valid = bool(capability_inventory) and all(
        item.get("capability_id")
        and item.get("implementation_paths")
        and item.get("suitable_metrics")
        and item.get("evaluation_status")
        in {
            "synthetic_contract_case_only",
            "covered_by_family_not_separately_benchmarked",
            "not_separately_benchmarked",
            "no_independent_accuracy_score_underlying_tool_results_are_authoritative",
            "synthetic_metric_contract_only_no_generic_learned_classifier_claim",
        }
        for item in capability_inventory
    )
    local_gates = {
        "all_declared_evaluator_families_exercised": {
            "status": "pass" if declared_families == observed_families else "fail",
            "declared": sorted(declared_families),
            "observed": sorted(observed_families),
        },
        "implemented_capability_inventory": {
            "status": "pass" if capability_contract_valid else "fail",
            "capability_count": len(capability_inventory),
        },
        "normative_split_roles_and_leakage": {
            "status": (
                "pass"
                if required_split_roles <= observed_split_roles
                and split_result["leakage"]["status"] == "pass"
                else "fail"
            ),
            "required_roles": sorted(required_split_roles),
            "observed_roles": sorted(observed_split_roles),
            "leakage_status": split_result["leakage"]["status"],
        },
        "manifest_training_firewall_contract": {
            "status": "pass"
            if firewall_checks and all(firewall_checks.values())
            else "fail",
            "checks": firewall_checks,
        },
        "protected_operating_point_contract": {
            "status": "pass" if selection_contract_valid else "fail",
            "evidence": (
                "Protected cases carry pre-registered configurations. Fixed AP/risk-coverage "
                "diagnostics cannot select an operating point or feed back into training."
            ),
        },
        "complete_raw_predictions_retained": {
            "status": "pass" if not raw_violations else "fail",
            "violating_samples": raw_violations,
        },
        "reference_implementation_baseline": {
            "status": "pass" if golden.get("status") == "passed" else "fail",
        },
        "stratified_metric_contract": {
            "status": "pass" if subgroup_contract_valid else "fail",
            "observed_overall_status": subgroup_report.get("overall_status"),
        },
        "undefined_metric_truth_table": {
            "status": "pass" if null_semantics_valid else "fail",
            "sample_id": "background-test-pure-empty",
        },
    }
    gate_families = {"local": local_gates, "product": product_gates}
    required_names = {
        "local": LOCAL_GATE_NAMES,
        "product": PRODUCT_GATE_NAMES,
    }
    missing_gate_names = {
        family: sorted(required_names[family] - set(gates))
        for family, gates in gate_families.items()
    }
    unexpected_gate_names = {
        family: sorted(set(gates) - required_names[family])
        for family, gates in gate_families.items()
    }
    invalid_gate_states: dict[str, Any] = {}
    for family, gates in gate_families.items():
        for name, item in gates.items():
            state = item.get("status") if isinstance(item, dict) else None
            if state not in GATE_STATES:
                invalid_gate_states[f"{family}.{name}"] = state

    def family_status(family: str) -> str:
        gates = gate_families[family]
        if (
            missing_gate_names[family]
            or unexpected_gate_names[family]
            or any(key.startswith(f"{family}.") for key in invalid_gate_states)
        ):
            return "fail"
        states = {gates[name]["status"] for name in required_names[family]}
        if "fail" in states:
            return "fail"
        if "not_evaluable" in states:
            return "not_evaluable"
        return "pass"

    local_status = family_status("local")
    product_status = family_status("product")
    family_states = {local_status, product_status}
    overall_status = (
        "fail"
        if "fail" in family_states
        else "not_evaluable"
        if "not_evaluable" in family_states
        else "pass"
    )
    return {
        "schema_version": 2,
        "gate_policy": "geointel-p4-evaluation-harness-v2",
        "status": overall_status,
        "phase_decision": "ready_for_phase5" if overall_status == "pass" else "blocked",
        "local_harness_status": local_status,
        "product_benchmark_status": product_status,
        "promotion_allowed": False,
        "numeric_model_release_targets": "not_frozen_without_reviewed_representative_incumbent_baseline",
        "local_gates": local_gates,
        "product_gates": product_gates,
        "invalid_gate_states": invalid_gate_states,
        "missing_gate_names": missing_gate_names,
        "unexpected_gate_names": unexpected_gate_names,
        "critical_subgroup_policy": (
            "Any required subgroup with insufficient support, missing metrics, a failed "
            "non-inferiority comparison or regression blocks promotion; averages cannot override it."
        ),
        "decision": (
            "All Phase 4 completion gates pass."
            if overall_status == "pass"
            else "At least one mandatory Phase 4 gate failed; Phase 5 remains blocked."
            if overall_status == "fail"
            else (
                "All evaluated Phase 4 gates pass, but mandatory product evidence remains "
                "not evaluable; Phase 5 remains blocked."
            )
        ),
    }


def build_evidence_manifest(
    artifacts: dict[str, Any],
    benchmark_manifest: dict[str, Any],
) -> dict[str, Any]:
    retained = []
    for name, payload in sorted(artifacts.items()):
        content = json_bytes(payload)
        retained.append(
            {
                "path": name,
                "sha256": hashlib.sha256(content).hexdigest(),
                "size_bytes": len(content),
            }
        )
    return {
        "schema_version": 2,
        "phase": "P4",
        "benchmark_id": benchmark_manifest["benchmark_id"],
        "artifacts": retained,
        "artifact_count": len(retained),
        "claim_boundary": (
            "Immutable local reference-harness evidence only; product accuracy and release "
            "remain blocked while product gates are not evaluable or fail."
        ),
    }


def firewall_contract_checks(
    split_result: dict[str, Any],
    protected_cases_path: Path,
) -> dict[str, bool]:
    development = split_result["development"]["samples"]
    protected = split_result["protected"]
    train = [item for item in development if item["split"] == "train"]
    validation = next(item for item in development if item["split"] == "val")
    protected_item = protected["samples"][0]
    checks: dict[str, bool] = {}
    try:
        assert_training_inputs_safe([], train, protected, trusted_fixture_mode=True)
    except LeakageError:
        checks["clean_train_allowed"] = False
    else:
        checks["clean_train_allowed"] = True
    for name, paths, records in (
        ("non_train_role_blocked", [], [validation]),
        ("protected_path_blocked", [protected_cases_path], []),
        (
            "renamed_protected_lineage_blocked",
            [],
            [{**train[0], "source_family": protected_item["source_family"]}],
        ),
    ):
        try:
            assert_training_inputs_safe(
                paths, records, protected, trusted_fixture_mode=True
            )
        except LeakageError:
            checks[name] = True
        else:
            checks[name] = False
    return checks


def runtime_identity() -> dict[str, Any]:
    return {
        "python": platform.python_version(),
        "python_implementation": platform.python_implementation(),
        "platform": platform.platform(),
        "dependencies": {
            "numpy": dependency_version("numpy"),
            "pyproj": dependency_version("pyproj"),
            "shapely": dependency_version("shapely"),
        },
        "execution_device": "CPU deterministic evaluator arithmetic; no production model inference",
        "cuda_used_for_reference_harness": False,
    }


def metric_results_without_raw(evaluation: dict[str, Any]) -> list[dict[str, Any]]:
    return [
        {key: value for key, value in item.items() if key not in {"raw", "failures"}}
        for item in evaluation["results"]
    ]


def build_input_manifest(
    repo_root: Path,
    snapshot: dict[str, Any],
    product_baseline_manifest: Path,
    product_gates: dict[str, Any],
) -> dict[str, Any]:
    code_paths = [
        "scripts/accuracy_phase4_evaluator.py",
        "scripts/generate_accuracy_phase4_splits.py",
        "scripts/run_accuracy_phase4_benchmark.py",
        "scripts/run_golden_qa_benchmark.py",
        "backend/app/services/qa_service.py",
    ]
    input_paths = [
        "fixtures/accuracy/p4/split-source-manifest.json",
        "fixtures/accuracy/p4/protected-baseline-cases.json",
        "fixtures/golden/golden_qa_benchmarks.json",
        "docs/ACCURACY.md",
        "docs/DATA_SOURCES.md",
        "fixtures/accuracy/readiness/full-scan-manifest.json",
        "fixtures/accuracy/readiness/leakage-report.json",
    ]
    baseline_gate = product_gates.get("executed_product_incumbent_baseline") or {}
    manifest_binding: dict[str, Any] | None = None
    resolved_manifest = product_baseline_manifest.resolve()
    try:
        requested_path = resolved_manifest.relative_to(repo_root.resolve()).as_posix()
    except (OSError, ValueError):
        requested_path = str(product_baseline_manifest)
    if resolved_manifest.is_file():
        try:
            relative_manifest = resolved_manifest.relative_to(repo_root.resolve())
        except (OSError, ValueError):
            relative_manifest = None
        if relative_manifest is not None:
            manifest_binding = {
                "path": relative_manifest.as_posix(),
                "sha256": sha256(resolved_manifest),
                "size_bytes": resolved_manifest.stat().st_size,
            }
    product_binding = {
        "requested_path": requested_path,
        "validation_status": baseline_gate.get("status"),
        "manifest": manifest_binding,
        "artifacts": sorted(
            baseline_gate.get("checked_artifacts") or [],
            key=lambda item: str(item.get("role")),
        ),
        "product_gate_evidence_sha256": canonical_hash(product_gates),
    }
    return {
        "schema_version": 2,
        "repository_commit": repository_commit(repo_root),
        "code": [repository_file(repo_root, path) for path in code_paths],
        "inputs": [repository_file(repo_root, path) for path in input_paths],
        "readiness_snapshot": snapshot,
        "readiness_snapshot_sha256": canonical_hash(snapshot),
        "runtime": runtime_identity(),
        "product_baseline": product_binding,
        "model_execution": {
            "status": baseline_gate.get("status", "fail"),
            "reason": baseline_gate.get("reason"),
            "configured_active_model": snapshot.get("active_model"),
        },
    }


def run_workflow(
    repo_root: Path,
    output_dir: Path | None,
    product_baseline_manifest: Path | None = None,
) -> dict[str, Any]:
    source_path = repo_root / "fixtures/accuracy/p4/split-source-manifest.json"
    cases_path = repo_root / "fixtures/accuracy/p4/protected-baseline-cases.json"
    source = json.loads(source_path.read_text(encoding="utf-8"))
    development, protected, leakage = build_manifests(source, trusted_fixture_mode=True)
    if leakage["status"] != "pass":
        raise LeakageError(
            f"Leakage gate failed with {leakage['finding_count']} findings"
        )
    split_result = {
        "development": development,
        "protected": protected,
        "leakage": leakage,
        "generation_status": {
            "schema_version": 1,
            "generator_version": GENERATOR_VERSION,
            "status": "pass",
            "source_manifest_sha256": leakage["source_manifest_sha256"],
            "development_manifest_sha256": development["manifest_sha256"],
            "protected_manifest_sha256": protected["manifest_sha256"],
        },
    }
    protected_evaluation_ids = {
        item["sample_id"]
        for item in protected["samples"]
        if item["split"] in {"test", "background-test"}
    }
    portfolio = json.loads(cases_path.read_text(encoding="utf-8"))
    evaluation = evaluate_cases(cases_path, protected_evaluation_ids)
    firewall_checks = firewall_contract_checks(split_result, cases_path)
    golden = canonical_golden_baseline()
    snapshot = readiness_snapshot(repo_root)
    baseline_path = (
        product_baseline_manifest
        if product_baseline_manifest is not None
        else repo_root / "artifacts/evidence/accuracy/P4/product-baseline-manifest.json"
    )
    product_gates = product_gate_evidence(repo_root, snapshot, baseline_path)
    gate_report = build_release_gate_report(
        split_result,
        evaluation,
        portfolio,
        golden,
        firewall_checks,
        product_gates,
    )
    input_manifest = build_input_manifest(
        repo_root, snapshot, baseline_path, product_gates
    )
    evaluation_contract = {
        "schema_version": 2,
        "benchmark_id": BENCHMARK_ID,
        "workflow_version": WORKFLOW_VERSION,
        "evaluator_version": EVALUATOR_VERSION,
        "split_generator_version": GENERATOR_VERSION,
        "evaluator_families": sorted(evaluation["evaluated_task_families"]),
        "implemented_capabilities": evaluation["task_inventory"],
        "metric_contract": repository_file(
            repo_root, "docs/ACCURACY.md"
        ),
        "gate_states": sorted(GATE_STATES),
        "protected_policy": evaluation["protected_policy"],
        "undefined_value_policy": (
            "Undefined denominators are null with numerator, denominator and support; "
            "they are never coerced to a perfect score."
        ),
        "claim_boundary": evaluation["claim_boundary"],
    }
    raw_items = [item["raw"] for item in evaluation["results"]]
    raw_predictions = {
        "schema_version": 2,
        "evaluator_version": EVALUATOR_VERSION,
        "portfolio_file_sha256": evaluation["portfolio_file_sha256"],
        "items": raw_items,
        "items_canonical_json_sha256": canonical_hash(raw_items),
        "hash_specification": evaluation["hash_specification"],
    }
    metric_results = metric_results_without_raw(evaluation)
    metric_report = {
        key: value
        for key, value in evaluation.items()
        if key not in {"results", "failures"}
    }
    metric_report["results"] = metric_results
    metric_report["metric_results_canonical_json_sha256"] = canonical_hash(
        metric_results
    )
    metric_report["full_results_canonical_json_sha256"] = evaluation[
        "results_canonical_json_sha256"
    ]
    failure_gallery = {
        "schema_version": 2,
        "taxonomy": "docs/ACCURACY.md",
        "failure_count": len(evaluation["failures"]),
        "items": evaluation["failures"],
        "items_canonical_json_sha256": canonical_hash(evaluation["failures"]),
        "rendering_status": (
            "machine_readable_examples_retained; a visual production gallery requires "
            "controlled access to protected imagery"
        ),
    }
    taxonomy_entries = sorted(
        {(item["error_code"], item["kind"]) for item in evaluation["failures"]}
    )
    error_taxonomy = {
        "schema_version": 2,
        "source": "docs/ACCURACY.md",
        "observed_codes": [
            {"error_code": code, "kind": kind} for code, kind in taxonomy_entries
        ],
        "observed_failure_count": len(evaluation["failures"]),
        "claim_boundary": evaluation["claim_boundary"],
    }
    object_task_names = {
        "object_detection",
        "footprint_segmentation",
        "vector_comparison",
        "change_detection",
        "geospatial_data_validation",
    }
    object_metrics = {
        "schema_version": 2,
        "status": "fixture_contract_only",
        "results": [
            item for item in metric_results if item["task"] in object_task_names
        ],
    }
    tile_metrics = {
        "schema_version": 2,
        "status": "fixture_contract_only",
        "results": [
            item
            for item in metric_results
            if item["task"] in {"raster_classification", "terrain_interpretation"}
        ],
    }
    aoi_metrics = {
        "schema_version": 2,
        "status": "not_evaluable",
        "reason": (
            "Synthetic single-case fixtures do not provide independent product AOI clusters. "
            "AOI micro/macro and cluster-bootstrap evidence requires the protected product corpus."
        ),
        "required_future_outputs": [
            "per-AOI primary metrics",
            "micro and macro aggregation",
            "paired candidate-minus-incumbent deltas",
            "cluster-bootstrap confidence intervals",
        ],
    }
    stratified_metrics = evaluation["subgroups"]
    calibration_items = []
    for item in metric_results:
        calibration = item["metrics"].get("calibration")
        coverage_risk = item["metrics"].get("coverage_risk")
        if calibration is not None or coverage_risk is not None:
            calibration_items.append(
                {
                    "sample_id": item["sample_id"],
                    "task": item["task"],
                    "calibration": calibration,
                    "coverage_risk": coverage_risk,
                }
            )
    calibration_metrics = {
        "schema_version": 2,
        "status": "fixture_diagnostic_only",
        "selection_allowed": False,
        "items": calibration_items,
        "note": (
            "Fixed diagnostic bins and risk thresholds test metric arithmetic; they do not "
            "select or change any operating point."
        ),
    }
    latency_reliability = {
        "schema_version": 2,
        "status": "not_evaluable",
        "model_inference_executed": False,
        "reason": (
            "The reference harness performs deterministic evaluator arithmetic only. "
            "GPU latency, VRAM, throughput and failure-rate gates require the real active model."
        ),
    }
    human_review_summary = {
        "schema_version": 2,
        "status": product_gates["human_review_complete"]["status"],
        "reviewed": product_gates["human_review_complete"].get("observed"),
        "required": product_gates["human_review_complete"].get("required"),
        "source": "readiness-snapshot.json bound to Phase-1/3 evidence",
        "ai_review_is_human_signoff": False,
    }
    candidate_vs_incumbent = {
        "schema_version": 2,
        "status": "not_evaluable",
        "reason": (
            "Phase 4 has no valid real incumbent product baseline and no pre-registered "
            "candidate; synthetic fixture values cannot define non-inferiority."
        ),
        "future_gate_contract": {
            "unit": "paired independent AOI",
            "global_and_critical_subgroups_required": True,
            "missing_or_insufficient_support": "not_evaluable",
            "aggregate_improvement_may_mask_subgroup_regression": False,
            "numeric_margin": "to_be_frozen_before_protected_access",
        },
    }
    input_manifest_file_sha256 = hashlib.sha256(json_bytes(input_manifest)).hexdigest()
    benchmark_manifest = {
        "schema_version": 2,
        "benchmark_id": BENCHMARK_ID,
        "workflow_version": WORKFLOW_VERSION,
        "evaluator_version": EVALUATOR_VERSION,
        "split_generator_version": GENERATOR_VERSION,
        "repository_commit": input_manifest["repository_commit"],
        "input_manifest": {
            "path": "input-manifest.json",
            "sha256": input_manifest_file_sha256,
        },
        "inputs": {
            "split_source": repository_file(
                repo_root, "fixtures/accuracy/p4/split-source-manifest.json"
            ),
            "protected_cases": repository_file(
                repo_root, "fixtures/accuracy/p4/protected-baseline-cases.json"
            ),
            "golden_qa_manifest": repository_file(
                repo_root, "fixtures/golden/golden_qa_benchmarks.json"
            ),
            "phase3_full_scan": repository_file(
                repo_root, "fixtures/accuracy/readiness/full-scan-manifest.json"
            ),
            "phase3_leakage": repository_file(
                repo_root, "fixtures/accuracy/readiness/leakage-report.json"
            ),
        },
        "code": input_manifest["code"],
        "runtime": input_manifest["runtime"],
        "product_baseline": input_manifest["product_baseline"],
        "product_gate_evidence_sha256": canonical_hash(product_gates),
        "split_manifests": {
            "development_sha256": development["manifest_sha256"],
            "protected_sha256": protected["manifest_sha256"],
            "leakage_status": leakage["status"],
        },
        "inference_and_selection": {
            "synthetic_reference_harness": True,
            "production_model_inference_executed": product_gates[
                "executed_product_incumbent_baseline"
            ]["status"]
            == "pass",
            "test_used_for_selection": False,
            "background_test_used_for_selection": False,
            "challenge_labels_available": False,
            "raw_predictions_retained": True,
            "threshold_source": "pre_registered_configuration_only",
        },
        "evaluation_results_canonical_json_sha256": evaluation[
            "results_canonical_json_sha256"
        ],
        "reference_baseline_sha256": golden["content_sha256"],
        "claim_boundary": evaluation["claim_boundary"],
    }
    run_fingerprint = canonical_hash(
        {
            "workflow_version": WORKFLOW_VERSION,
            "input_manifest_sha256": input_manifest_file_sha256,
            "product_gate_evidence_sha256": canonical_hash(product_gates),
            "evaluation_results_sha256": evaluation["results_canonical_json_sha256"],
            "development_split_sha256": development["manifest_sha256"],
            "protected_split_sha256": protected["manifest_sha256"],
        }
    )
    evidence_run_id = f"p4-{WORKFLOW_VERSION}-{run_fingerprint[:20]}"
    benchmark_manifest["evidence_run_id"] = evidence_run_id
    benchmark_manifest["manifest_sha256"] = canonical_hash(benchmark_manifest)
    gate_report["benchmark_manifest_sha256"] = benchmark_manifest["manifest_sha256"]
    target_output_dir = output_dir or (
        repo_root / "artifacts/evidence/accuracy/P4/runs" / evidence_run_id
    )
    try:
        evidence_path = (
            target_output_dir.resolve().relative_to(repo_root.resolve()).as_posix()
        )
    except (OSError, ValueError):
        evidence_path = str(target_output_dir)
    workflow_summary = {
        "schema_version": 2,
        "status": gate_report["status"],
        "phase_decision": gate_report["phase_decision"],
        "local_harness_status": gate_report["local_harness_status"],
        "product_benchmark_status": gate_report["product_benchmark_status"],
        "benchmark_manifest_sha256": benchmark_manifest["manifest_sha256"],
        "evidence_run_id": evidence_run_id,
        "evidence_path": evidence_path,
        "product_gate_evidence_sha256": canonical_hash(product_gates),
        "split_counts": leakage["split_counts"],
        "task_family_count": evaluation["task_count"],
        "implemented_capability_count": len(evaluation["task_inventory"]),
        "case_count": evaluation["case_count"],
        "failure_example_count": len(evaluation["failures"]),
        "evaluation_results_canonical_json_sha256": evaluation[
            "results_canonical_json_sha256"
        ],
        "reference_baseline_sha256": golden["content_sha256"],
        "promotion_allowed": False,
        "phase4_done": gate_report["status"] == "pass",
        "phase5_ready": gate_report["status"] == "pass",
    }
    artifacts: dict[str, Any] = {
        "acceptance-gates.json": gate_report,
        "aoi-metrics.json": aoi_metrics,
        "baseline-raw-predictions.json": raw_predictions,
        "benchmark-manifest.json": benchmark_manifest,
        "calibration-metrics.json": calibration_metrics,
        "candidate-vs-incumbent.json": candidate_vs_incumbent,
        "development-split-manifest.json": development,
        "error-taxonomy.json": error_taxonomy,
        "evaluation-contract.json": evaluation_contract,
        "failure-gallery.json": failure_gallery,
        "generation-status.json": split_result["generation_status"],
        "human-review-summary.json": human_review_summary,
        "input-manifest.json": input_manifest,
        "latency-and-reliability.json": latency_reliability,
        "leakage-gate-report.json": leakage,
        "metric-report.json": metric_report,
        "object-metrics.json": object_metrics,
        "protected-split-manifest.json": protected,
        "reference-implementation-baseline.json": golden,
        "release-gate-report.json": gate_report,
        "split-and-leakage-audit.json": leakage,
        "stratified-metrics.json": stratified_metrics,
        "tile-metrics.json": tile_metrics,
        "workflow-summary.json": workflow_summary,
    }
    evidence = build_evidence_manifest(artifacts, benchmark_manifest)
    evidence_bundle = {**artifacts, "evidence-manifest.json": evidence}
    write_json_bundle_immutable(target_output_dir, evidence_bundle)
    return workflow_summary


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repo-root", type=Path, default=ROOT)
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=None,
        help=(
            "Override the default content-addressed artifacts/evidence/accuracy/P4/runs/<run-id> directory."
        ),
    )
    parser.add_argument(
        "--product-baseline-manifest",
        type=Path,
        help=(
            "Optional governed product incumbent manifest. It can pass only when real "
            "active-model inference and all referenced artifacts validate."
        ),
    )
    parser.add_argument(
        "--allow-product-blocked",
        action="store_true",
        help=(
            "Return zero when the local harness passes while product evidence remains "
            "fail/not_evaluable. This never changes a gate or phase decision."
        ),
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    try:
        summary = run_workflow(
            args.repo_root.resolve(),
            args.output_dir.resolve() if args.output_dir else None,
            args.product_baseline_manifest.resolve()
            if args.product_baseline_manifest
            else None,
        )
    except Exception as exc:  # noqa: BLE001 - workflow evidence must fail closed
        print(
            json.dumps(
                {"status": "fail", "error": f"{type(exc).__name__}: {exc}"},
                indent=2,
            )
        )
        return 2
    print(json.dumps(summary, indent=2, sort_keys=True))
    if summary["status"] == "pass":
        return 0
    return (
        0
        if args.allow_product_blocked and summary["local_harness_status"] == "pass"
        else 2
    )


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