#!/usr/bin/env python3 """Generate immutable P4 split manifests and fail closed on leakage.""" from __future__ import annotations import argparse import hashlib import json import math import re import unicodedata from collections import Counter, defaultdict from datetime import date, datetime from pathlib import Path from typing import Any, Iterable from pyproj import CRS SCHEMA_VERSION = 1 GENERATOR_VERSION = "1.3.0" TRAIN_SPLITS = {"train"} SELECTION_SPLITS = {"val", "calibration"} DEVELOPMENT_SPLITS = TRAIN_SPLITS | SELECTION_SPLITS PROTECTED_SPLITS = {"test", "background-test", "challenge"} ALL_SPLITS = DEVELOPMENT_SPLITS | PROTECTED_SPLITS MANDATORY_SPLIT_ROLES = ( "train", "val", "calibration", "test", "background-test", "challenge", ) NORMATIVE_SPLITS = frozenset(MANDATORY_SPLIT_ROLES) EVALUATOR_PROTECTED_SPLITS = frozenset({"test", "background-test"}) MIN_INDEPENDENCE_BUFFER_M = 2000.0 MIN_PERCEPTUAL_HAMMING_THRESHOLD = 4 MIN_LABEL_GEOMETRY_HAMMING_THRESHOLD = 2 TRUSTED_FIXTURE_POLICY = { "dataset_version": "geointel-p4-harness-fixture-v2", "claim_boundary": ( "Synthetic contract fixtures for evaluator regression only; " "never production accuracy evidence." ), "policy_id": "geointel-p4-synthetic-fixture-v2", } SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") FINGERPRINT_PATTERN = re.compile(r"^[0-9a-f]{16}$") P3_ITEM_ID_PATTERN = re.compile(r"^[0-9a-f]{20}$") PROVENANCE_MANIFEST_TYPE = "geointel_phase4_source_provenance" ASSET_BINDINGS = { "raw_image": ("raw_image_path", "raw_image_sha256"), "processed_image": ("processed_image_path", "processed_image_sha256"), "label": ("label_path", "label_sha256"), "label_geometry": ("label_geometry_path", "label_geometry_hash"), } FIXTURE_ACTIVATION_FIELDS = { "trusted_fixture_mode", "fixture_mode", "allow_synthetic_fixture", "test_fixture_mode", } class LeakageError(ValueError): """Raised when split isolation is not demonstrably safe.""" def canonical_bytes(value: Any) -> bytes: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":") ).encode("utf-8") def canonical_hash(value: Any) -> str: return hashlib.sha256(canonical_bytes(value)).hexdigest() def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def canonical_identifier(value: Any, *, field: str, sample_id: str) -> str: if not isinstance(value, str): raise LeakageError(f"{sample_id}: {field} must be a string identifier") if any(unicodedata.category(character) == "Cc" for character in value): raise LeakageError(f"{sample_id}: {field} contains control characters") normalized = " ".join(unicodedata.normalize("NFKC", value).split()).casefold() if not normalized: raise LeakageError(f"{sample_id}: {field} must be non-empty") return normalized def normalized_identifier_list(values: Any, *, field: str, sample_id: str) -> list[str]: if not isinstance(values, list): raise LeakageError(f"{sample_id}: {field} must be a list") normalized: list[str] = [] originals_by_identity: dict[str, set[str]] = defaultdict(set) for value in values: identity = canonical_identifier(value, field=field, sample_id=sample_id) originals_by_identity[identity].add(str(value)) normalized.append(identity) ambiguous = { identity: sorted(originals) for identity, originals in originals_by_identity.items() if len(originals) > 1 or normalized.count(identity) > 1 } if ambiguous: raise LeakageError( f"{sample_id}: {field} contains ambiguous canonical identities: {ambiguous}" ) return sorted(normalized) def validated_policy_thresholds(source: dict[str, Any]) -> tuple[float, int, int]: try: buffer_m = float(source.get("independence_buffer_m")) perceptual = int(source.get("perceptual_hamming_threshold")) geometry = int(source.get("label_geometry_hamming_threshold")) except (TypeError, ValueError) as exc: raise LeakageError("Split isolation thresholds must be numeric") from exc if not math.isfinite(buffer_m) or buffer_m < MIN_INDEPENDENCE_BUFFER_M: raise LeakageError( "independence_buffer_m cannot weaken the code-owned minimum " f"of {MIN_INDEPENDENCE_BUFFER_M:g} m" ) if not MIN_PERCEPTUAL_HAMMING_THRESHOLD <= perceptual < 64: raise LeakageError( "perceptual_hamming_threshold cannot weaken the code-owned minimum " f"of {MIN_PERCEPTUAL_HAMMING_THRESHOLD}" ) if not MIN_LABEL_GEOMETRY_HAMMING_THRESHOLD <= geometry < 64: raise LeakageError( "label_geometry_hamming_threshold cannot weaken the code-owned minimum " f"of {MIN_LABEL_GEOMETRY_HAMMING_THRESHOLD}" ) return buffer_m, perceptual, geometry def write_json(path: Path, payload: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) content = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n" temporary = path.with_name(f".{path.name}.tmp") temporary.write_text(content, encoding="utf-8") temporary.replace(path) def validate_metric_crs(value: Any) -> str: try: crs = CRS.from_user_input(value) except Exception as exc: # noqa: BLE001 - pyproj exposes multiple parser errors raise LeakageError(f"Invalid CRS: {value!r}") from exc axes = crs.axis_info if crs.is_geographic or not axes: raise LeakageError(f"CRS must be projected in metres: {value!r}") if any( abs(float(axis.unit_conversion_factor or 0.0) - 1.0) > 1e-12 for axis in axes[:2] ): raise LeakageError(f"CRS axes must use metres: {value!r}") return crs.to_string() def require_hex( sample_id: str, field: str, value: Any, pattern: re.Pattern[str] ) -> str: normalized = str(value or "").lower() if not pattern.fullmatch(normalized): raise LeakageError(f"{sample_id}: {field} has an invalid fingerprint") return normalized def fingerprint_distance(left: str, right: str) -> int: return (int(left, 16) ^ int(right, 16)).bit_count() def bbox_distance(left: list[float], right: list[float]) -> float: dx = max(left[0] - right[2], right[0] - left[2], 0.0) dy = max(left[1] - right[3], right[1] - left[3], 0.0) return math.hypot(dx, dy) def is_trusted_fixture_source(source: dict[str, Any]) -> bool: return ( source.get("dataset_version") == TRUSTED_FIXTURE_POLICY["dataset_version"] and source.get("claim_boundary") == TRUSTED_FIXTURE_POLICY["claim_boundary"] ) def _resolve_evidence_path(value: Any, source_root: Path | None) -> Path: if not isinstance(value, str) or not value.strip(): raise LeakageError("Governance evidence path must be non-empty") path = Path(value) if not path.is_absolute(): if source_root is None: raise LeakageError( "Relative governance evidence paths require the source manifest root" ) path = source_root / path return path.resolve() def _required_string(value: Any, *, field: str) -> str: if not isinstance(value, str) or not value.strip() or value != value.strip(): raise LeakageError(f"{field} must be a non-empty, trimmed string") if any(unicodedata.category(character) == "Cc" for character in value): raise LeakageError(f"{field} contains control characters") return value def _canonical_bound_path(value: Any, *, field: str) -> str: return Path(_required_string(value, field=field)).as_posix() def _load_evidence_object( binding: Any, *, key: str, source_root: Path | None, from_protected_manifest: bool = False, ) -> tuple[dict[str, Any], dict[str, Any]]: required = ( {"path", "sha256", "resolved_path"} if from_protected_manifest else { "path", "sha256", } ) if not isinstance(binding, dict) or set(binding) != required: raise LeakageError( f"governance_evidence.{key} must contain exactly {sorted(required)}" ) expected = require_hex( "", f"governance_evidence.{key}.sha256", binding.get("sha256"), SHA256_PATTERN, ) path_value = ( binding["resolved_path"] if from_protected_manifest else binding["path"] ) path = _resolve_evidence_path(path_value, source_root) if not path.is_file(): raise LeakageError(f"Governance evidence is unavailable: {path}") actual = sha256_file(path) if actual != expected: raise LeakageError( f"Governance evidence checksum mismatch for {key}: {actual} != {expected}" ) try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise LeakageError(f"Governance evidence is not readable JSON: {path}") from exc if not isinstance(payload, dict) or not payload: raise LeakageError( f"Governance evidence must be a non-empty JSON object: {path}" ) return payload, { "path": str(binding["path"]), "resolved_path": str(path), "sha256": actual, } def _validate_p3_manifest(payload: dict[str, Any]) -> dict[str, dict[str, Any]]: required_root = { "schema_version", "scan_id", "scanner_version", "completed_at", "items", "reconciliation", } missing = sorted(required_root - set(payload)) if missing or payload.get("schema_version") != 1: raise LeakageError(f"P3 scan manifest has an invalid schema; missing={missing}") _required_string(payload["scan_id"], field="P3 scan_id") _required_string(payload["scanner_version"], field="P3 scanner_version") try: completed = datetime.fromisoformat( _required_string(payload["completed_at"], field="P3 completed_at") ) except ValueError as exc: raise LeakageError("P3 completed_at must be ISO-8601") from exc if completed.tzinfo is None: raise LeakageError("P3 completed_at must include a timezone") reconciliation = payload["reconciliation"] count_fields = ("examined", "skipped", "unreachable", "inventory_total") if ( not isinstance(reconciliation, dict) or reconciliation.get("reconciles") is not True ): raise LeakageError("P3 reconciliation must explicitly pass") counts: dict[str, int] = {} for field in count_fields: value = reconciliation.get(field) if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise LeakageError(f"P3 reconciliation.{field} is invalid") counts[field] = value if sum(counts[field] for field in count_fields[:3]) != counts["inventory_total"]: raise LeakageError("P3 reconciliation counts do not add up") items = payload["items"] if ( not isinstance(items, list) or not items or len(items) != counts["inventory_total"] ): raise LeakageError("P3 item count does not match the reconciled inventory") required_item = { "item_id", "path", "sha256", "size_bytes", "status", "read_status", "recommended_action", "empty_content", "schema_conformity", "anomalies", } indexed: dict[str, dict[str, Any]] = {} for position, item in enumerate(items): if not isinstance(item, dict) or not required_item <= set(item): raise LeakageError(f"P3 item {position} has an invalid schema") item_id = require_hex( f"", "item_id", item["item_id"], P3_ITEM_ID_PATTERN ) if item_id in indexed: raise LeakageError(f"P3 scan manifest contains duplicate item_id {item_id}") _canonical_bound_path(item["path"], field=f"P3 item {item_id}.path") size = item["size_bytes"] if size is not None and ( isinstance(size, bool) or not isinstance(size, int) or size < 0 ): raise LeakageError(f"P3 item {item_id}.size_bytes is invalid") if item["sha256"] is not None: require_hex(item_id, "sha256", item["sha256"], SHA256_PATTERN) if not isinstance(item["anomalies"], list): raise LeakageError(f"P3 item {item_id}.anomalies must be a list") indexed[item_id] = item return indexed def _validate_provenance_manifest( payload: dict[str, Any], ) -> dict[str, dict[str, Any]]: root_fields = { "schema_version", "manifest_type", "status", "records", "records_canonical_json_sha256", } if ( set(payload) != root_fields or payload.get("schema_version") != 1 or payload.get("manifest_type") != PROVENANCE_MANIFEST_TYPE or payload.get("status") != "pass" ): raise LeakageError("Source provenance manifest has an invalid schema or status") records = payload["records"] expected_hash = require_hex( "", "records_canonical_json_sha256", payload["records_canonical_json_sha256"], SHA256_PATTERN, ) if ( not isinstance(records, list) or not records or canonical_hash(records) != expected_hash ): raise LeakageError("Source provenance records are empty or checksum-invalid") record_fields = { "record_id", "sample_id", "status", "lineage_status", "training_allowed", "perceptual_image_hash", "label_geometry_fingerprint", "assets", } asset_fields = {"path", "sha256", "size_bytes", "p3_item_id"} indexed: dict[str, dict[str, Any]] = {} seen_samples: set[str] = set() for position, record in enumerate(records): if not isinstance(record, dict) or set(record) != record_fields: raise LeakageError( f"Source provenance record {position} has an invalid schema" ) record_id = _required_string( record["record_id"], field=f"provenance record {position}.record_id" ) sample_id = canonical_identifier( record["sample_id"], field="sample_id", sample_id=record_id ) if ( sample_id != record["sample_id"] or record_id in indexed or sample_id in seen_samples ): raise LeakageError("Source provenance manifest has ambiguous identities") if ( record["status"] != "accepted" or record["lineage_status"] != "complete" or record["training_allowed"] is not True ): raise LeakageError(f"Source provenance record {record_id} is not accepted") for field in ("perceptual_image_hash", "label_geometry_fingerprint"): require_hex(sample_id, field, record[field], FINGERPRINT_PATTERN) assets = record["assets"] if not isinstance(assets, dict) or set(assets) != set(ASSET_BINDINGS): raise LeakageError( f"Source provenance record {record_id} has incomplete assets" ) for role, asset in assets.items(): if not isinstance(asset, dict) or set(asset) != asset_fields: raise LeakageError( f"Source provenance record {record_id}.{role} has an invalid schema" ) _canonical_bound_path(asset["path"], field=f"{record_id}.{role}.path") require_hex(sample_id, f"{role}.sha256", asset["sha256"], SHA256_PATTERN) size = asset["size_bytes"] if isinstance(size, bool) or not isinstance(size, int) or size < 0: raise LeakageError(f"{record_id}.{role}.size_bytes is invalid") require_hex( sample_id, f"{role}.p3_item_id", asset["p3_item_id"], P3_ITEM_ID_PATTERN, ) indexed[record_id] = record seen_samples.add(sample_id) return indexed def _validated_governance_bundle( evidence: Any, *, source_root: Path | None, from_protected_manifest: bool = False, ) -> tuple[dict[str, Any], dict[str, dict[str, Any]], dict[str, dict[str, Any]]]: keys = {"p3_scan_manifest", "source_provenance_manifest"} if not isinstance(evidence, dict) or set(evidence) != keys: raise LeakageError(f"governance_evidence must contain exactly {sorted(keys)}") p3, p3_binding = _load_evidence_object( evidence["p3_scan_manifest"], key="p3_scan_manifest", source_root=source_root, from_protected_manifest=from_protected_manifest, ) provenance, provenance_binding = _load_evidence_object( evidence["source_provenance_manifest"], key="source_provenance_manifest", source_root=source_root, from_protected_manifest=from_protected_manifest, ) p3_items = _validate_p3_manifest(p3) provenance_records = _validate_provenance_manifest(provenance) verified = { "p3_scan_manifest": { **p3_binding, "scan_id": p3["scan_id"], "scanner_version": p3["scanner_version"], }, "source_provenance_manifest": { **provenance_binding, "manifest_type": provenance["manifest_type"], "records_canonical_json_sha256": provenance[ "records_canonical_json_sha256" ], }, } return verified, p3_items, provenance_records def validate_source_trust( source: dict[str, Any], source_root: Path | None, *, trusted_fixture_mode: bool = False, ) -> dict[str, Any]: activation = sorted(FIXTURE_ACTIVATION_FIELDS & set(source)) if activation: raise LeakageError( f"Fixture mode cannot be enabled by source metadata: {activation}" ) if trusted_fixture_mode: if not is_trusted_fixture_source(source): raise LeakageError( "Explicit fixture mode only accepts the code-owned fixture policy" ) return { "mode": "synthetic_fixture", "policy_id": TRUSTED_FIXTURE_POLICY["policy_id"], "production_accuracy_use_allowed": False, } if is_trusted_fixture_source(source): raise LeakageError( "Synthetic fixture source requires explicit trusted_fixture_mode=True" ) verified, p3_items, provenance_records = _validated_governance_bundle( source.get("governance_evidence"), source_root=source_root ) return { "mode": "governed_production", "verified_evidence": verified, "production_accuracy_use_allowed": True, "_p3_items_by_id": p3_items, "_provenance_records_by_id": provenance_records, } def _verified_content_hashes( sample: dict[str, Any], *, sample_id: str, source_root: Path | None, source_trust: dict[str, Any], ) -> tuple[dict[str, str], dict[str, dict[str, Any]]]: statuses: dict[str, str] = {} bindings: dict[str, dict[str, Any]] = {} provenance_record: dict[str, Any] | None = None provenance_id: str | None = None p3_ids: dict[str, Any] = {} if source_trust["mode"] == "governed_production": governance = sample.get("governance_binding") required = {"source_provenance_record_id", "p3_item_ids"} if not isinstance(governance, dict) or set(governance) != required: raise LeakageError( f"{sample_id}: governance_binding must contain exactly {sorted(required)}" ) provenance_id = _required_string( governance["source_provenance_record_id"], field=f"{sample_id}.source_provenance_record_id", ) p3_ids = governance["p3_item_ids"] if not isinstance(p3_ids, dict) or set(p3_ids) != set(ASSET_BINDINGS): raise LeakageError(f"{sample_id}: p3_item_ids must bind every asset") provenance_record = source_trust["_provenance_records_by_id"].get(provenance_id) if provenance_record is None or provenance_record["sample_id"] != sample_id: raise LeakageError(f"{sample_id}: exact provenance record is unavailable") for field in ("perceptual_image_hash", "label_geometry_fingerprint"): if provenance_record[field] != str(sample[field]).lower(): raise LeakageError(f"{sample_id}: provenance {field} mismatch") for role, (path_field, hash_field) in ASSET_BINDINGS.items(): expected = require_hex( sample_id, hash_field, sample[hash_field], SHA256_PATTERN ) path_value = sample.get(path_field) if path_value is None: if source_trust["mode"] == "governed_production": raise LeakageError( f"{sample_id}: governed production requires accessible {path_field}" ) statuses[hash_field] = "synthetic_fixture_bound" continue path = _resolve_evidence_path(path_value, source_root) if not path.is_file(): raise LeakageError(f"{sample_id}: content path is unavailable: {path}") actual = sha256_file(path) size = path.stat().st_size if actual != expected: raise LeakageError( f"{sample_id}: {hash_field} checksum mismatch: {actual} != {expected}" ) statuses[hash_field] = "recomputed_from_accessible_file" if source_trust["mode"] != "governed_production": continue assert provenance_record is not None and provenance_id is not None p3_id = require_hex( sample_id, f"p3_item_ids.{role}", p3_ids[role], P3_ITEM_ID_PATTERN ) asset = provenance_record["assets"][role] canonical_path = _canonical_bound_path(path_value, field=path_field) if ( _canonical_bound_path(asset["path"], field=f"{provenance_id}.{role}.path") != canonical_path or asset["sha256"] != actual or asset["size_bytes"] != size or asset["p3_item_id"] != p3_id ): raise LeakageError(f"{sample_id}: provenance binding mismatch for {role}") p3_item = source_trust["_p3_items_by_id"].get(p3_id) if p3_item is None or ( _canonical_bound_path(p3_item["path"], field=f"P3 {p3_id}.path") != canonical_path or p3_item["sha256"] != actual or p3_item["size_bytes"] != size or p3_item["status"] != "examined" or p3_item["read_status"] != "readable" or p3_item["recommended_action"] != "accept" or p3_item["empty_content"] is not False or p3_item["schema_conformity"] not in {"conformant", "not_applicable"} or p3_item["anomalies"] ): raise LeakageError(f"{sample_id}: P3 record is not accepted for {role}") bindings[role] = { "path": str(path_value), "resolved_path": str(path), "sha256": actual, "size_bytes": size, "p3_item_id": p3_id, "source_provenance_record_id": provenance_id, } return statuses, bindings def normalized_sample( sample: dict[str, Any], *, source_root: Path | None = None, source_trust: dict[str, Any] | None = None, ) -> dict[str, Any]: required = { "sample_id", "task", "split", "group_id", "source_family", "temporal_family", "object_ids", "bbox", "raw_image_sha256", "processed_image_sha256", "label_sha256", "perceptual_image_hash", "label_geometry_hash", "label_geometry_fingerprint", "native_feature_ids", "parent_raster_id", "acquisition_id", "acquisition_date", } missing = sorted(required - set(sample)) if missing: raise LeakageError( f"{sample.get('sample_id', '')}: missing fields {missing}" ) raw_sample_id = str(sample["sample_id"]) sample_id = canonical_identifier( sample["sample_id"], field="sample_id", sample_id=raw_sample_id ) split = canonical_identifier(sample["split"], field="split", sample_id=sample_id) if split not in ALL_SPLITS: raise LeakageError(f"{sample_id}: unsupported split {split!r}") bbox = sample["bbox"] if not isinstance(bbox, list) or len(bbox) != 4: raise LeakageError(f"{sample_id}: bbox must contain four values") try: values = [float(value) for value in bbox] except (TypeError, ValueError) as exc: raise LeakageError(f"{sample_id}: bbox values must be numeric") from exc if ( not all(math.isfinite(value) for value in values) or values[0] >= values[2] or values[1] >= values[3] ): raise LeakageError(f"{sample_id}: invalid bbox") if source_trust is None: raise LeakageError(f"{sample_id}: explicit source trust is required") trust = source_trust result = dict(sample) result["sample_id"] = sample_id result["task"] = canonical_identifier( sample["task"], field="task", sample_id=sample_id ) result["split"] = split result["bbox"] = values hash_verification, content_bindings = _verified_content_hashes( sample, sample_id=sample_id, source_root=source_root, source_trust=trust, ) result["content_hash_verification"] = hash_verification if content_bindings: result["content_path_bindings"] = content_bindings for field in ( "raw_image_sha256", "processed_image_sha256", "label_sha256", "label_geometry_hash", ): result[field] = require_hex(sample_id, field, sample[field], SHA256_PATTERN) for field in ("perceptual_image_hash", "label_geometry_fingerprint"): result[field] = require_hex( sample_id, field, sample[field], FINGERPRINT_PATTERN ) for field in ("object_ids", "native_feature_ids"): result[field] = normalized_identifier_list( sample[field], field=field, sample_id=sample_id ) for field in ( "group_id", "source_family", "temporal_family", "parent_raster_id", "acquisition_id", ): result[field] = canonical_identifier( sample[field], field=field, sample_id=sample_id ) try: result["acquisition_date"] = date.fromisoformat( str(sample["acquisition_date"]) ).isoformat() except (TypeError, ValueError) as exc: raise LeakageError(f"{sample_id}: acquisition_date must be ISO-8601") from exc if trust["mode"] == "governed_production": binding = sample["governance_binding"] result["governance_binding"] = { "source_provenance_record_id": binding["source_provenance_record_id"], "p3_item_ids": { role: binding["p3_item_ids"][role] for role in ASSET_BINDINGS }, } record_without_split = { key: value for key, value in result.items() if key != "split" } result["record_sha256"] = canonical_hash(record_without_split) return result def assign_split_roles( source: dict[str, Any], *, source_root: Path | None = None, source_trust: dict[str, Any] | None = None, ) -> list[dict[str, Any]]: """Assign complete leakage groups deterministically, or validate a frozen assignment.""" raw_samples = [dict(item) for item in source.get("samples", [])] mode = canonical_identifier( source.get("assignment_mode") or "preassigned", field="assignment_mode", sample_id="", ) split_presence = [bool(item.get("split")) for item in raw_samples] if mode == "preassigned": if raw_samples and not all(split_presence): raise LeakageError("Preassigned mode requires a split on every sample") return raw_samples if mode != "deterministic_grouped": raise LeakageError(f"Unsupported assignment_mode: {mode!r}") if any(split_presence): raise LeakageError( "deterministic_grouped mode refuses partially or fully preassigned splits" ) config = source.get("split_assignment") or {} requested_roles = config.get("roles") if requested_roles is not None and tuple(requested_roles) != MANDATORY_SPLIT_ROLES: raise LeakageError( "split_assignment.roles cannot change the code-owned mandatory role order" ) roles = list(MANDATORY_SPLIT_ROLES) weights = { role: float((config.get("weights") or {}).get(role, 1.0)) for role in roles } if any(not math.isfinite(value) or value <= 0 for value in weights.values()): raise LeakageError("split_assignment weights must be positive finite values") placeholder = [ normalized_sample( {**item, "split": "train"}, source_root=source_root, source_trust=source_trust, ) for item in raw_samples ] parent = list(range(len(placeholder))) def find(index: int) -> int: while parent[index] != index: parent[index] = parent[parent[index]] index = parent[index] return index def union(left: int, right: int) -> None: left_root = find(left) right_root = find(right) if left_root != right_root: parent[max(left_root, right_root)] = min(left_root, right_root) identity_fields = ( "group_id", "source_family", "temporal_family", "acquisition_date", "raw_image_sha256", "processed_image_sha256", "label_sha256", "label_geometry_hash", "parent_raster_id", "acquisition_id", ) seen: dict[tuple[str, str], int] = {} for index, sample in enumerate(placeholder): identities = [(field, str(sample[field])) for field in identity_fields] identities.extend(("object_id", str(value)) for value in sample["object_ids"]) identities.extend( ("native_feature_id", str(value)) for value in sample["native_feature_ids"] ) for identity in identities: if identity in seen: union(index, seen[identity]) else: seen[identity] = index buffer_m, image_threshold, geometry_threshold = validated_policy_thresholds(source) for index, left in enumerate(placeholder): for right_index, right in enumerate(placeholder[index + 1 :], start=index + 1): if ( fingerprint_distance( left["perceptual_image_hash"], right["perceptual_image_hash"] ) <= image_threshold or fingerprint_distance( left["label_geometry_fingerprint"], right["label_geometry_fingerprint"], ) <= geometry_threshold or bbox_distance(left["bbox"], right["bbox"]) < buffer_m ): union(index, right_index) components: dict[int, list[int]] = defaultdict(list) for index in range(len(placeholder)): components[find(index)].append(index) if len(components) < len(roles): raise LeakageError( f"Not enough independent groups for required roles: {len(components)} < {len(roles)}" ) seed = str(config.get("seed") or "geointel-p4-group-split-v1") groups = sorted( components.values(), key=lambda indices: canonical_hash( { "seed": seed, "sample_ids": sorted( placeholder[index]["sample_id"] for index in indices ), } ), ) stratify_by = tuple(config.get("stratify_by") or ["task"]) assigned_counts: Counter[str] = Counter() stratum_counts: dict[str, Counter[str]] = defaultdict(Counter) assignment: dict[int, str] = {} for group_index, indices in enumerate(groups): strata = { f"{field}={placeholder[index].get(field) or (placeholder[index].get('metadata') or {}).get(field)}" for index in indices for field in stratify_by } candidates = ( roles[group_index : group_index + 1] if group_index < len(roles) else roles ) role = min( candidates, key=lambda candidate: ( assigned_counts[candidate] / weights[candidate] + sum( stratum_counts[stratum][candidate] / weights[candidate] for stratum in strata ), candidate, ), ) for index in indices: assignment[index] = role assigned_counts[role] += len(indices) for stratum in strata: stratum_counts[stratum][role] += len(indices) group_role = {indices[0]: assignment[indices[0]] for indices in groups} role_group_counts = Counter(group_role.values()) for task in sorted({sample["task"] for sample in placeholder}): task_groups = [ indices for indices in groups if any(placeholder[index]["task"] == task for index in indices) ] if any( assignment[indices[0]] in EVALUATOR_PROTECTED_SPLITS for indices in task_groups ): continue movable = [ indices for indices in task_groups if role_group_counts[assignment[indices[0]]] > 1 ] if not movable: raise LeakageError( f"Cannot provide evaluator-protected coverage for task {task!r}" ) indices = min( movable, key=lambda candidate: canonical_hash( { "seed": seed, "protected_task": task, "sample_ids": sorted( placeholder[index]["sample_id"] for index in candidate ), } ), ) old_role = assignment[indices[0]] new_role = min( EVALUATOR_PROTECTED_SPLITS, key=lambda role: (assigned_counts[role], role), ) for index in indices: assignment[index] = new_role role_group_counts[old_role] -= 1 role_group_counts[new_role] += 1 assigned_counts[old_role] -= len(indices) assigned_counts[new_role] += len(indices) return [ {**sample, "split": assignment[index]} for index, sample in enumerate(raw_samples) ] def leakage_findings( samples: list[dict[str, Any]], independence_buffer_m: float, perceptual_hamming_threshold: int, label_geometry_hamming_threshold: int, ) -> list[dict[str, Any]]: findings: list[dict[str, Any]] = [] identity_fields = ( "sample_id", "group_id", "source_family", "temporal_family", "acquisition_date", "raw_image_sha256", "processed_image_sha256", "label_sha256", "label_geometry_hash", "parent_raster_id", "acquisition_id", "record_sha256", ) identities: dict[str, dict[str, set[str]]] = { key: defaultdict(set) for key in identity_fields } object_splits: dict[str, set[str]] = defaultdict(set) native_feature_splits: dict[str, set[str]] = defaultdict(set) for sample in samples: for key, index in identities.items(): index[str(sample[key])].add(sample["split"]) for object_id in sample["object_ids"]: object_splits[str(object_id)].add(sample["split"]) for feature_id in sample["native_feature_ids"]: native_feature_splits[str(feature_id)].add(sample["split"]) code_by_key = { "sample_id": "S-SAMPLE-IDENTITY", "group_id": "S-SPATIAL-GROUP", "source_family": "S-SOURCE-FAMILY", "temporal_family": "S-TEMPORAL-FAMILY", "acquisition_date": "S-ACQUISITION-DATE", "raw_image_sha256": "S-RAW-IMAGE-DUPLICATE", "processed_image_sha256": "S-PROCESSED-IMAGE-DUPLICATE", "label_sha256": "S-LABEL-DUPLICATE", "label_geometry_hash": "S-LABEL-GEOMETRY-DUPLICATE", "parent_raster_id": "S-PARENT-RASTER", "acquisition_id": "S-ACQUISITION", "record_sha256": "S-EXACT-DUPLICATE", } for key, index in identities.items(): for value, splits in sorted(index.items()): if len(splits) > 1: findings.append( { "code": code_by_key[key], "identity": value, "splits": sorted(splits), } ) for object_id, splits in sorted(object_splits.items()): if len(splits) > 1: findings.append( { "code": "S-OBJECT-INSTANCE", "identity": object_id, "splits": sorted(splits), } ) for feature_id, splits in sorted(native_feature_splits.items()): if len(splits) > 1: findings.append( { "code": "S-NATIVE-FEATURE", "identity": feature_id, "splits": sorted(splits), } ) for index, left in enumerate(samples): for right in samples[index + 1 :]: if left["split"] == right["split"]: continue image_distance = fingerprint_distance( left["perceptual_image_hash"], right["perceptual_image_hash"] ) if image_distance <= perceptual_hamming_threshold: findings.append( { "code": "S-PERCEPTUAL-IMAGE-NEAR-DUPLICATE", "left": left["sample_id"], "right": right["sample_id"], "distance": image_distance, "maximum_allowed_distance": perceptual_hamming_threshold, } ) geometry_distance = fingerprint_distance( left["label_geometry_fingerprint"], right["label_geometry_fingerprint"] ) if geometry_distance <= label_geometry_hamming_threshold: findings.append( { "code": "S-LABEL-GEOMETRY-NEAR-DUPLICATE", "left": left["sample_id"], "right": right["sample_id"], "distance": geometry_distance, "maximum_allowed_distance": label_geometry_hamming_threshold, } ) distance = bbox_distance(left["bbox"], right["bbox"]) if distance < independence_buffer_m: findings.append( { "code": "S-SPATIAL-OVERLAP", "left": left["sample_id"], "right": right["sample_id"], "left_split": left["split"], "right_split": right["split"], "distance_m": distance, "required_distance_m": independence_buffer_m, } ) return findings def build_protected_task_coverage( samples: list[dict[str, Any]], source: dict[str, Any] ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: implemented_tasks = sorted({item["task"] for item in samples}) exemptions_raw = source.get("protected_task_exemptions") or {} if not isinstance(exemptions_raw, dict): raise LeakageError("protected_task_exemptions must be an object") exemptions: dict[str, str] = {} for raw_task, raw_reason in exemptions_raw.items(): task = canonical_identifier(raw_task, field="task", sample_id="") if task in exemptions: raise LeakageError(f"Ambiguous protected task exemption for {task!r}") reason = " ".join(str(raw_reason).split()) if len(reason) < 20: raise LeakageError( f"Protected task exemption for {task!r} requires a concrete reason" ) exemptions[task] = reason unknown = sorted(set(exemptions) - set(implemented_tasks)) if unknown: raise LeakageError( f"Protected task exemptions reference unknown tasks: {unknown}" ) coverage: list[dict[str, Any]] = [] findings: list[dict[str, Any]] = [] for task in implemented_tasks: counts = Counter(item["split"] for item in samples if item["task"] == task) evaluator_count = sum(counts[role] for role in EVALUATOR_PROTECTED_SPLITS) status = "covered" if evaluator_count else "not_evaluable" row = { "task": task, "status": status, "test_count": counts["test"], "background_test_count": counts["background-test"], "sealed_challenge_count": counts["challenge"], "exemption_reason": exemptions.get(task), } coverage.append(row) if evaluator_count: continue finding = { "code": ( "S-PROTECTED-TASK-COVERAGE-EXEMPTED" if task in exemptions else "S-PROTECTED-TASK-COVERAGE-MISSING" ), "task": task, "status": "not_evaluable", } if task in exemptions: finding["reason"] = exemptions[task] findings.append(finding) return coverage, findings def seal_challenge_sample(sample: dict[str, Any]) -> dict[str, Any]: sealed_fields = { "label_sha256", "label_geometry_hash", "label_path", "label_geometry_path", "label_geometry_fingerprint", "object_ids", "native_feature_ids", "record_sha256", "governance_binding", "content_path_bindings", } result = {key: value for key, value in sample.items() if key not in sealed_fields} result["sealed"] = True result["withheld_fields"] = sorted(sealed_fields) return result def redact_challenge_findings(findings: list[dict[str, Any]]) -> list[dict[str, Any]]: sensitive_codes = { "S-LABEL-DUPLICATE", "S-LABEL-GEOMETRY-DUPLICATE", "S-OBJECT-INSTANCE", "S-NATIVE-FEATURE", "S-EXACT-DUPLICATE", } redacted: list[dict[str, Any]] = [] for finding in findings: item = dict(finding) if item.get("code") in sensitive_codes and "challenge" in item.get( "splits", [] ): item.pop("identity", None) item["identity_withheld"] = "sealed_challenge_identity" redacted.append(item) return redacted def build_manifests( source: dict[str, Any], *, source_root: Path | None = None, trusted_fixture_mode: bool = False, ) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: if source.get("schema_version") != 1: raise LeakageError("Unsupported source manifest schema_version") if ( not isinstance(source.get("dataset_version"), str) or not source["dataset_version"].strip() ): raise LeakageError("dataset_version must be non-empty") metric_crs = validate_metric_crs(source.get("crs")) source_trust = validate_source_trust( source, source_root, trusted_fixture_mode=trusted_fixture_mode ) assigned_source_samples = assign_split_roles( source, source_root=source_root, source_trust=source_trust ) samples = [ normalized_sample( item, source_root=source_root, source_trust=source_trust, ) for item in assigned_source_samples ] if not samples: raise LeakageError("Source manifest has no samples") if len({item["sample_id"] for item in samples}) != len(samples): raise LeakageError( "Duplicate or ambiguous canonical sample_id in source manifest" ) buffer_m, perceptual_threshold, geometry_threshold = validated_policy_thresholds( source ) split_counts = Counter(item["split"] for item in samples) declared_required = source.get("required_splits") if ( declared_required is not None and tuple(declared_required) != MANDATORY_SPLIT_ROLES ): raise LeakageError( "required_splits cannot change the code-owned mandatory role order" ) missing_splits = sorted(NORMATIVE_SPLITS - set(split_counts)) if missing_splits: raise LeakageError(f"Required splits are absent: {missing_splits}") findings = leakage_findings( samples, buffer_m, perceptual_threshold, geometry_threshold ) protected_task_coverage, coverage_findings = build_protected_task_coverage( samples, source ) findings.extend(coverage_findings) public_findings = redact_challenge_findings(findings) canonical_source = dict(source) canonical_source["samples"] = sorted( source.get("samples", []), key=lambda item: str(item.get("sample_id", "")) ) source_hash = canonical_hash(canonical_source) development_samples = sorted( (item for item in samples if item["split"] in DEVELOPMENT_SPLITS), key=lambda item: item["sample_id"], ) protected_samples = sorted( ( seal_challenge_sample(item) if item["split"] == "challenge" else item for item in samples if item["split"] in PROTECTED_SPLITS ), key=lambda item: item["sample_id"], ) public_source_trust = { key: value for key, value in source_trust.items() if not key.startswith("_") } common = { "schema_version": SCHEMA_VERSION, "generator_version": GENERATOR_VERSION, "dataset_version": source["dataset_version"], "source_manifest_sha256": source_hash, "source_trust": public_source_trust, "mandatory_split_roles": list(MANDATORY_SPLIT_ROLES), "crs": metric_crs, "independence_buffer_m": buffer_m, "perceptual_hamming_threshold": perceptual_threshold, "label_geometry_hamming_threshold": geometry_threshold, "assignment_mode": canonical_identifier( source.get("assignment_mode") or "preassigned", field="assignment_mode", sample_id="", ), "assignment_algorithm": "group-before-split-deficit-balancer-v1", "assignment_seed": str( (source.get("split_assignment") or {}).get("seed") or "geointel-p4-group-split-v1" ), "claim_boundary": source.get("claim_boundary"), } development = { **common, "manifest_role": "development_and_calibration", "allowed_splits": sorted(DEVELOPMENT_SPLITS), "training_access_allowed_by_split": { "train": True, "val": False, "calibration": False, }, "selection_access_allowed_by_split": { "train": False, "val": True, "calibration": True, }, "samples": development_samples, } development["manifest_sha256"] = canonical_hash(development) protected = { **common, "manifest_role": "protected_release_only", "allowed_splits": sorted(PROTECTED_SPLITS), "implemented_tasks": sorted({item["task"] for item in samples}), "protected_task_coverage": protected_task_coverage, "training_access_allowed": False, "selection_use_allowed": False, "labels_available_by_split": { "test": "frozen_evaluator_only", "background-test": "frozen_evaluator_only", "challenge": "sealed_external", }, "access_policy": ( "Only the frozen release evaluator may consume test/background-test labels; " "challenge labels remain external/sealed." ), "samples": protected_samples, } protected["manifest_sha256"] = canonical_hash(protected) leakage = { "schema_version": SCHEMA_VERSION, "generator_version": GENERATOR_VERSION, "status": "pass" if not findings else "fail", "source_manifest_sha256": source_hash, "development_manifest_sha256": development["manifest_sha256"], "protected_manifest_sha256": protected["manifest_sha256"], "inventory_total": len(samples), "split_counts": dict(sorted(split_counts.items())), "checked_identity_fields": [ "sample_id", "group_id", "source_family", "acquisition_date", "temporal_family", "object_ids", "native_feature_ids", "raw_image_sha256", "processed_image_sha256", "label_sha256", "record_sha256", "perceptual_image_hash", "label_geometry_hash", "label_geometry_fingerprint", "parent_raster_id", "acquisition_id", "bbox_distance", ], "crs_validation": {"status": "pass", "crs": metric_crs, "distance_units": "m"}, "independence_buffer_m": buffer_m, "perceptual_hamming_threshold": perceptual_threshold, "label_geometry_hamming_threshold": geometry_threshold, "finding_count": len(findings), "findings": public_findings, } return development, protected, leakage def validate_protected_manifest( protected_manifest: dict[str, Any], *, trusted_fixture_mode: bool = False, ) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]]] | None: if not isinstance(protected_manifest, dict) or not protected_manifest: raise LeakageError("Protected manifest trust validation failed: empty manifest") required = { "schema_version", "generator_version", "dataset_version", "source_manifest_sha256", "source_trust", "mandatory_split_roles", "manifest_role", "allowed_splits", "implemented_tasks", "protected_task_coverage", "samples", "manifest_sha256", } missing = sorted(required - set(protected_manifest)) if missing: raise LeakageError( f"Protected manifest trust validation failed: missing fields {missing}" ) if protected_manifest["schema_version"] != SCHEMA_VERSION: raise LeakageError("Protected manifest trust validation failed: schema_version") if protected_manifest["generator_version"] != GENERATOR_VERSION: raise LeakageError( "Protected manifest trust validation failed: generator_version" ) if protected_manifest["manifest_role"] != "protected_release_only": raise LeakageError("Protected manifest trust validation failed: manifest_role") if tuple(protected_manifest["mandatory_split_roles"]) != MANDATORY_SPLIT_ROLES: raise LeakageError( "Protected manifest trust validation failed: mandatory roles" ) if set(protected_manifest["allowed_splits"]) != PROTECTED_SPLITS: raise LeakageError("Protected manifest trust validation failed: allowed splits") require_hex( "", "source_manifest_sha256", protected_manifest["source_manifest_sha256"], SHA256_PATTERN, ) expected_manifest_hash = require_hex( "", "manifest_sha256", protected_manifest["manifest_sha256"], SHA256_PATTERN, ) unsigned = { key: value for key, value in protected_manifest.items() if key != "manifest_sha256" } if canonical_hash(unsigned) != expected_manifest_hash: raise LeakageError( "Protected manifest trust validation failed: checksum mismatch" ) trust = protected_manifest["source_trust"] if not isinstance(trust, dict): raise LeakageError("Protected manifest trust validation failed: source trust") evidence_indexes = None if trust.get("mode") == "synthetic_fixture": if not trusted_fixture_mode: raise LeakageError( "Protected synthetic fixture requires explicit trusted_fixture_mode=True" ) if ( protected_manifest["dataset_version"] != TRUSTED_FIXTURE_POLICY["dataset_version"] or protected_manifest.get("claim_boundary") != TRUSTED_FIXTURE_POLICY["claim_boundary"] or trust.get("policy_id") != TRUSTED_FIXTURE_POLICY["policy_id"] or trust.get("production_accuracy_use_allowed") is not False ): raise LeakageError( "Protected manifest trust validation failed: fixture policy binding" ) elif trust.get("mode") == "governed_production": evidence = trust.get("verified_evidence") if ( trust.get("production_accuracy_use_allowed") is not True or not isinstance(evidence, dict) or set(evidence) != { "p3_scan_manifest", "source_provenance_manifest", } ): raise LeakageError( "Protected manifest trust validation failed: provenance binding" ) lean_evidence: dict[str, dict[str, Any]] = {} for key, binding in evidence.items(): if not isinstance(binding, dict): raise LeakageError( "Protected manifest trust validation failed: provenance binding" ) expected_fields = ( {"path", "resolved_path", "sha256", "scan_id", "scanner_version"} if key == "p3_scan_manifest" else { "path", "resolved_path", "sha256", "manifest_type", "records_canonical_json_sha256", } ) if set(binding) != expected_fields: raise LeakageError( "Protected manifest trust validation failed: evidence schema" ) lean_evidence[key] = { field: binding[field] for field in ("path", "resolved_path", "sha256") } verified, p3_items, provenance_records = _validated_governance_bundle( lean_evidence, source_root=None, from_protected_manifest=True ) if ( verified["p3_scan_manifest"]["scan_id"] != evidence["p3_scan_manifest"]["scan_id"] or verified["p3_scan_manifest"]["scanner_version"] != evidence["p3_scan_manifest"]["scanner_version"] or verified["source_provenance_manifest"]["manifest_type"] != evidence["source_provenance_manifest"]["manifest_type"] or verified["source_provenance_manifest"]["records_canonical_json_sha256"] != evidence["source_provenance_manifest"]["records_canonical_json_sha256"] ): raise LeakageError( "Protected manifest trust validation failed: evidence attestation" ) evidence_indexes = (p3_items, provenance_records) else: raise LeakageError( "Protected manifest trust validation failed: unsupported trust mode" ) samples = protected_manifest["samples"] if not isinstance(samples, list) or not samples: raise LeakageError("Protected manifest trust validation failed: empty samples") split_counts = Counter( item.get("split") for item in samples if isinstance(item, dict) ) if set(split_counts) != PROTECTED_SPLITS or any( split_counts[role] < 1 for role in PROTECTED_SPLITS ): raise LeakageError( "Protected manifest trust validation failed: protected coverage" ) for item in samples: if not isinstance(item, dict): raise LeakageError( "Protected manifest trust validation failed: sample type" ) sample_id = canonical_identifier( item.get("sample_id"), field="sample_id", sample_id="" ) if sample_id != item["sample_id"]: raise LeakageError( "Protected manifest trust validation failed: noncanonical sample" ) if item["split"] == "challenge": forbidden = { "label_sha256", "label_geometry_hash", "label_geometry_fingerprint", "object_ids", "native_feature_ids", "record_sha256", "governance_binding", "content_path_bindings", } if forbidden & set(item) or item.get("sealed") is not True: raise LeakageError( "Protected manifest trust validation failed: challenge seal" ) coverage = protected_manifest["protected_task_coverage"] if not isinstance(coverage, list) or { row.get("task") for row in coverage if isinstance(row, dict) } != set(protected_manifest["implemented_tasks"]): raise LeakageError("Protected manifest trust validation failed: task coverage") if any(row.get("status") != "covered" for row in coverage): raise LeakageError( "Protected manifest trust validation failed: unevaluable task" ) return evidence_indexes def protected_identities(protected_manifest: dict[str, Any]) -> set[str]: identities: set[str] = set() scalar_fields = ( "sample_id", "group_id", "record_sha256", "source_family", "temporal_family", "acquisition_date", "raw_image_sha256", "processed_image_sha256", "label_sha256", "label_geometry_hash", "perceptual_image_hash", "label_geometry_fingerprint", "parent_raster_id", "acquisition_id", ) for item in protected_manifest.get("samples", []): identities.update( str(item[field]) for field in scalar_fields if field in item and str(item[field]) ) identities.update(str(value) for value in item.get("object_ids", [])) identities.update(str(value) for value in item.get("native_feature_ids", [])) return identities def assert_training_inputs_safe( input_paths: Iterable[Path], input_records: Iterable[dict[str, Any]], protected_manifest: dict[str, Any], *, trusted_fixture_mode: bool = False, ) -> None: """Refuse untrusted manifests, non-train roles and protected identities.""" evidence_indexes = validate_protected_manifest( protected_manifest, trusted_fixture_mode=trusted_fixture_mode ) production_mode = ( protected_manifest["source_trust"]["mode"] == "governed_production" ) forbidden = protected_identities(protected_manifest) violations: list[str] = [] paths = [Path(path) for path in input_paths] records = list(input_records) if not paths and not records: violations.append("empty_training_inputs") forbidden_content_hashes = { value for value in forbidden if SHA256_PATTERN.fullmatch(value) } for path in paths: lowered = unicodedata.normalize("NFKC", path.as_posix()).casefold() if any( token in lowered for token in ("protected", "holdout", "challenge", "background-test") ): violations.append(f"protected_path:{path}") if not path.is_file(): violations.append(f"unverifiable_training_path:{path}") continue content_hash = sha256_file(path) if content_hash in forbidden_content_hashes: violations.append(f"protected_content_hash:{path}") try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError): payload = None if isinstance(payload, dict): if payload.get("manifest_role") == "protected_release_only": violations.append(f"protected_manifest_content:{path}") def collect_splits(value: Any) -> set[str]: if isinstance(value, dict): found = ( { canonical_identifier( value["split"], field="split", sample_id="", ) } if "split" in value and isinstance(value["split"], str) else set() ) for nested in value.values(): found.update(collect_splits(nested)) return found if isinstance(value, list): found: set[str] = set() for nested in value: found.update(collect_splits(nested)) return found return set() non_train = sorted(collect_splits(payload) - TRAIN_SPLITS) if non_train: violations.append(f"non_train_content:{path}:{','.join(non_train)}") if paths and not records: violations.append("unbound_training_paths") scalar_fields = ( "sample_id", "group_id", "record_sha256", "source_family", "temporal_family", "acquisition_date", "raw_image_sha256", "processed_image_sha256", "label_sha256", "label_geometry_hash", "perceptual_image_hash", "label_geometry_fingerprint", "parent_raster_id", "acquisition_id", ) required_record_fields = { "sample_id", "split", "group_id", "source_family", "temporal_family", "acquisition_date", "raw_image_sha256", "processed_image_sha256", "label_sha256", "label_geometry_hash", "perceptual_image_hash", "label_geometry_fingerprint", "parent_raster_id", "acquisition_id", "object_ids", "native_feature_ids", } canonical_scalar_identifiers = { "sample_id", "group_id", "source_family", "temporal_family", "parent_raster_id", "acquisition_id", } def production_binding_violation(record: dict[str, Any]) -> str | None: if not production_mode: return None if evidence_indexes is None: return "production_evidence_unavailable" p3_items, provenance_records = evidence_indexes sample_id = str(record.get("sample_id", "")) try: governance = record.get("governance_binding") governance_fields = {"source_provenance_record_id", "p3_item_ids"} if not isinstance(governance, dict) or set(governance) != governance_fields: return f"missing_governance_binding:{sample_id}" provenance_id = _required_string( governance["source_provenance_record_id"], field=f"{sample_id}.source_provenance_record_id", ) p3_ids = governance["p3_item_ids"] if not isinstance(p3_ids, dict) or set(p3_ids) != set(ASSET_BINDINGS): return f"incomplete_p3_binding:{sample_id}" provenance = provenance_records.get(provenance_id) if provenance is None or provenance["sample_id"] != sample_id: return f"unavailable_provenance_record:{sample_id}" for field in ("perceptual_image_hash", "label_geometry_fingerprint"): if record.get(field) != provenance[field]: return f"provenance_fingerprint_mismatch:{sample_id}:{field}" content = record.get("content_path_bindings") if not isinstance(content, dict) or set(content) != set(ASSET_BINDINGS): return f"missing_accessible_content_paths:{sample_id}" binding_fields = { "path", "resolved_path", "sha256", "size_bytes", "p3_item_id", "source_provenance_record_id", } for role, (path_field, hash_field) in ASSET_BINDINGS.items(): bound = content[role] if not isinstance(bound, dict) or set(bound) != binding_fields: return f"invalid_content_binding:{sample_id}:{role}" path = _resolve_evidence_path(bound["resolved_path"], None) if not path.is_file(): return f"unavailable_content_path:{sample_id}:{role}" actual = sha256_file(path) size = path.stat().st_size p3_id = require_hex( sample_id, f"content_path_bindings.{role}.p3_item_id", bound["p3_item_id"], P3_ITEM_ID_PATTERN, ) asset = provenance["assets"][role] p3_item = p3_items.get(p3_id) canonical_path = _canonical_bound_path( bound["path"], field=f"{sample_id}.{path_field}" ) if ( record.get(path_field) != bound["path"] or record.get(hash_field) != actual or bound["sha256"] != actual or bound["size_bytes"] != size or bound["source_provenance_record_id"] != provenance_id or p3_ids[role] != p3_id or asset["path"] != canonical_path or asset["sha256"] != actual or asset["size_bytes"] != size or asset["p3_item_id"] != p3_id or p3_item is None ): return f"record_path_hash_binding_mismatch:{sample_id}:{role}" if ( _canonical_bound_path(p3_item["path"], field=f"P3 {p3_id}.path") != canonical_path or p3_item["sha256"] != actual or p3_item["size_bytes"] != size or p3_item["status"] != "examined" or p3_item["read_status"] != "readable" or p3_item["recommended_action"] != "accept" or p3_item["empty_content"] is not False or p3_item["schema_conformity"] not in {"conformant", "not_applicable"} or p3_item["anomalies"] ): return f"unaccepted_p3_binding:{sample_id}:{role}" unsigned_record = { key: value for key, value in record.items() if key not in {"split", "record_sha256"} } if canonical_hash(unsigned_record) != record.get("record_sha256"): return f"record_checksum_mismatch:{sample_id}" except (KeyError, OSError, TypeError, LeakageError) as exc: return f"unverifiable_production_record:{sample_id}:{exc}" return None for record in records: binding_violation = production_binding_violation(record) if binding_violation: violations.append(binding_violation) missing = sorted(required_record_fields - set(record)) if missing: violations.append( f"incomplete_training_record:{record.get('sample_id')}:{','.join(missing)}" ) raw_split = record.get("split") try: split = canonical_identifier( raw_split, field="split", sample_id=str(record.get("sample_id")) ) except LeakageError: split = "" if split not in TRAIN_SPLITS: violations.append(f"non_train_role:{record.get('sample_id')}:{raw_split}") values: set[str] = set() for field in scalar_fields: value = record.get(field) if value in (None, ""): continue if field in canonical_scalar_identifiers: try: value = canonical_identifier( value, field=field, sample_id=str(record.get("sample_id")) ) except LeakageError: violations.append(f"invalid_training_identity:{field}") continue elif field in {"acquisition_date"}: try: value = date.fromisoformat(str(value)).isoformat() except ValueError: violations.append("invalid_training_identity:acquisition_date") continue else: value = str(value).lower() values.add(str(value)) for field in ("object_ids", "native_feature_ids"): try: values.update( normalized_identifier_list( record.get(field, []), field=field, sample_id=str(record.get("sample_id")), ) ) except LeakageError: violations.append(f"invalid_training_identity:{field}") overlap = sorted((values - {""}) & forbidden) if overlap: violations.append(f"protected_identity:{','.join(overlap)}") if violations: raise LeakageError( "Training input firewall blocked: " + "; ".join(sorted(violations)) ) def invalidate_consumable_outputs( output_dir: Path, *, error: str, source_manifest_sha256: str | None, leakage: dict[str, Any] | None = None, ) -> dict[str, Any]: output_dir.mkdir(parents=True, exist_ok=True) failure_id = canonical_hash( { "generator_version": GENERATOR_VERSION, "source_manifest_sha256": source_manifest_sha256, "error": error, } ) status = { "schema_version": SCHEMA_VERSION, "generator_version": GENERATOR_VERSION, "status": "fail", "failure_id": failure_id, "error": error, "source_manifest_sha256": source_manifest_sha256, "development_manifest_sha256": None, "protected_manifest_sha256": None, "consumable_manifests_valid": False, } tombstone = { "schema_version": SCHEMA_VERSION, "generator_version": GENERATOR_VERSION, "status": "invalidated", "failure_id": failure_id, "consumable": False, "reason": "Latest regeneration failed; stale split content is fail-closed.", } failure_report = leakage or { "schema_version": SCHEMA_VERSION, "generator_version": GENERATOR_VERSION, "status": "fail", "source_manifest_sha256": source_manifest_sha256, "finding_count": 1, "findings": [ { "code": "S-GENERATION-ERROR", "message": error, } ], } write_json(output_dir / "generation-status.json", status) write_json(output_dir / "leakage-gate-report.json", failure_report) write_json(output_dir / "development-split-manifest.json", tombstone) write_json(output_dir / "protected-split-manifest.json", tombstone) return status def generate( source_path: Path, output_dir: Path, *, trusted_fixture_mode: bool = False, ) -> dict[str, Any]: output_dir.mkdir(parents=True, exist_ok=True) source_file_hash: str | None = None try: source_file_hash = sha256_file(source_path) source = json.loads(source_path.read_text(encoding="utf-8")) development, protected, leakage = build_manifests( source, source_root=source_path.parent, trusted_fixture_mode=trusted_fixture_mode, ) except (OSError, json.JSONDecodeError, LeakageError) as exc: invalidate_consumable_outputs( output_dir, error=str(exc), source_manifest_sha256=source_file_hash, ) raise generation_status = { "schema_version": SCHEMA_VERSION, "generator_version": GENERATOR_VERSION, "status": leakage["status"], "source_manifest_sha256": leakage["source_manifest_sha256"], "source_file_sha256": source_file_hash, "development_manifest_sha256": ( development["manifest_sha256"] if leakage["status"] == "pass" else None ), "protected_manifest_sha256": ( protected["manifest_sha256"] if leakage["status"] == "pass" else None ), "consumable_manifests_valid": leakage["status"] == "pass", } if leakage["status"] != "pass": error = f"Leakage gate failed with {leakage['finding_count']} findings" invalidate_consumable_outputs( output_dir, error=error, source_manifest_sha256=leakage["source_manifest_sha256"], leakage=leakage, ) raise LeakageError(error) train_samples = [ item for item in development["samples"] if item["split"] == "train" ] try: assert_training_inputs_safe( [], train_samples, protected, trusted_fixture_mode=trusted_fixture_mode ) except LeakageError as exc: invalidate_consumable_outputs( output_dir, error=str(exc), source_manifest_sha256=leakage["source_manifest_sha256"], leakage=leakage, ) raise write_json(output_dir / "generation-status.json", generation_status) write_json(output_dir / "leakage-gate-report.json", leakage) write_json(output_dir / "development-split-manifest.json", development) write_json(output_dir / "protected-split-manifest.json", protected) return { "development": development, "protected": protected, "leakage": leakage, "generation_status": generation_status, } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--source", type=Path, required=True) parser.add_argument("--output-dir", type=Path, required=True) return parser.parse_args() def main() -> int: args = parse_args() try: result = generate(args.source, args.output_dir) except (OSError, json.JSONDecodeError, LeakageError) as exc: print(json.dumps({"status": "fail", "error": str(exc)}, indent=2)) return 2 print( json.dumps( { "status": "pass", "development_manifest_sha256": result["development"]["manifest_sha256"], "protected_manifest_sha256": result["protected"]["manifest_sha256"], "split_counts": result["leakage"]["split_counts"], }, indent=2, ) ) return 0 if __name__ == "__main__": raise SystemExit(main())