Jens/geointel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
885e86b823c2a7e600ff236b4aaaceec7d698b33
geointel/scripts/run_accuracy_phase4_benchmark.py
T

3067 lines
121 KiB
Python
Raw Blame History

#!/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 / "docs/accuracy-program/status.json"
p3_path = repo_root / "artifacts/evidence/accuracy/P3/full-scan-manifest.json"
leakage_path = repo_root / "artifacts/evidence/accuracy/P3/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": "docs/accuracy-program/status.json",
"selected_json_pointers": ["/runtime/active_model"],
"sha256": canonical_hash(status_projection),
},
"phase3_full_scan": repository_file(
repo_root, "artifacts/evidence/accuracy/P3/full-scan-manifest.json"
),
"phase3_leakage": repository_file(
repo_root, "artifacts/evidence/accuracy/P3/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-program/05-metric-framework.md",
"docs/accuracy-program/07-source-authority-matrix.md",
"artifacts/evidence/accuracy/P3/full-scan-manifest.json",
"artifacts/evidence/accuracy/P3/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-program/05-metric-framework.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-program/05-metric-framework.md section 4",
"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-program/05-metric-framework.md section 4",
"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, "artifacts/evidence/accuracy/P3/full-scan-manifest.json"
),
"phase3_leakage": repository_file(
repo_root, "artifacts/evidence/accuracy/P3/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())
Reference in New Issue View Git Blame Copy Permalink