diff --git a/experiments/perception/e37_ravnoves00_acceptance_profile.json b/experiments/perception/e37_ravnoves00_acceptance_profile.json new file mode 100644 index 0000000..83042f8 --- /dev/null +++ b/experiments/perception/e37_ravnoves00_acceptance_profile.json @@ -0,0 +1,64 @@ +{ + "schema_version": "missioncore.e37-acceptance-profile/v1", + "profile_id": "e37-ravnoves00-r0-acceptance/v1", + "source": { + "session_id": "20260720T065719Z_viewer_live", + "display_name": "RAVNOVES00", + "materialization_id": "e30-materialization-841af926d8d28ab93538c46d8f31278a2234c4d1c12c7dc4dc296b249d59735a", + "engineering_generation_id": "e30-engineering-generation-62a4fea10dea9b77f69ceac1af5bf0e4928d9c7716083c22258a03670fe5bd4f", + "human_generation_id": "e30-review-generation-7982a882558d0be690b4c7092e328c080bfcbf52478a220452be7e887a588250" + }, + "denominator": { + "expected_items": 486, + "label_provenance": "engineering-reviewed-with-human-exceptions", + "independent_ground_truth": false + }, + "split": { + "strategy": "deterministic-source-stratum-range-holdout", + "seed": "ravnoves00-r0-validation-v1", + "validation_fraction": 0.3, + "mutable_after_publication": false + }, + "ontology": { + "presence": [ + "object-present", + "occupied-environment", + "background-or-noise", + "unknown" + ], + "geometry_association": [ + "object-associated", + "independent-occupied", + "rejected-nonobject", + "insufficient-support", + "unknown" + ], + "freshness": [ + "current", + "held", + "stale", + "unavailable", + "unknown" + ] + }, + "metrics": { + "presence_target": 0.9, + "geometry_association_target": 0.9, + "freshness_target": 0.9, + "accounting_target": 1.0, + "maximum_false_free_claims": 0, + "targets_apply_separately": true + }, + "severity": { + "levels": [ + "standard", + "medium", + "high" + ], + "high_impact_failure_blocks_gate": true + }, + "authority": { + "commands_enabled": false, + "navigation_or_safety_accepted": false + } +} diff --git a/experiments/perception/prepare_e37_worker_package.py b/experiments/perception/prepare_e37_worker_package.py new file mode 100644 index 0000000..0a1cd5e --- /dev/null +++ b/experiments/perception/prepare_e37_worker_package.py @@ -0,0 +1,290 @@ +#!/usr/bin/env python3 +"""Build a minimal immutable E37 package for Worker 006.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import shutil +import uuid +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from k1link.compute.e37_acceptance_contract import ( + E37_PACKAGE_SCHEMA, + E37_PROFILE_SCHEMA, +) + +_RUNTIME_FILES = { + "runtime/k1link/__init__.py": "src/k1link/__init__.py", + "runtime/k1link/compute/__init__.py": None, + "runtime/k1link/compute/e37_acceptance_contract.py": ( + "src/k1link/compute/e37_acceptance_contract.py" + ), + "runtime/run_e37_acceptance_contract.py": ( + "experiments/perception/worker/run_e37_acceptance_contract.py" + ), + "runtime/Invoke-E37AcceptanceContract.ps1": ( + "experiments/perception/worker/Invoke-E37AcceptanceContract.ps1" + ), +} +_GENERATED_COMPUTE_INIT = ( + '"""Minimal E37 worker projection; import the contract module explicitly."""\n' +) +_INPUT_FILES = { + "materialization": ("manifest.json", "materialized-items.jsonl"), + "engineering": ("manifest.json", "engineering-decisions.jsonl"), + "human": ("manifest.json", "review-decisions.jsonl"), +} + + +class E37WorkerPackageError(RuntimeError): + """The E37 package source or immutable package is invalid.""" + + +def build_e37_worker_package( + *, + repository_root: Path, + materialization_root: Path, + engineering_generation_root: Path, + human_generation_root: Path, + profile_path: Path, + output_root: Path, +) -> Path: + """Build or verify one content-addressed E37 worker package.""" + + repository = repository_root.resolve(strict=True) + profile_source = profile_path.resolve(strict=True) + profile = _read_json(profile_source) + if profile.get("schema_version") != E37_PROFILE_SCHEMA: + raise E37WorkerPackageError("E37 package profile is incompatible") + roots = { + "materialization": materialization_root.resolve(strict=True), + "engineering": engineering_generation_root.resolve(strict=True), + "human": human_generation_root.resolve(strict=True), + } + expected_ids = { + "materialization": profile["source"]["materialization_id"], + "engineering": profile["source"]["engineering_generation_id"], + "human": profile["source"]["human_generation_id"], + } + sources: dict[str, Path | None] = {} + for target, relative in _RUNTIME_FILES.items(): + source = None if relative is None else repository / relative + if source is not None and (not source.is_file() or source.is_symlink()): + raise E37WorkerPackageError(f"E37 runtime source is invalid: {relative}") + sources[target] = source + sources["profile.json"] = profile_source + for kind, filenames in _INPUT_FILES.items(): + root = roots[kind] + if root.name != expected_ids[kind]: + raise E37WorkerPackageError(f"E37 {kind} identity changed") + for filename in filenames: + source = root / filename + if not source.is_file() or source.is_symlink(): + raise E37WorkerPackageError(f"E37 {kind} artifact is invalid") + sources[f"input/{kind}/{root.name}/{filename}"] = source + + descriptors = [] + for relative, source in sorted(sources.items()): + payload = ( + _GENERATED_COMPUTE_INIT.encode() + if source is None + else source.read_bytes() + ) + descriptors.append( + { + "path": relative, + "byte_length": len(payload), + "sha256": hashlib.sha256(payload).hexdigest(), + } + ) + identity = { + "schema_version": E37_PACKAGE_SCHEMA, + "classification": "immutable-ravnoves00-r0-worker-input", + "source_ids": expected_ids, + "profile_sha256": _sha256(profile_source), + "artifact_paths": [row["path"] for row in descriptors], + "source_artifacts": descriptors, + "authority": { + "commands_enabled": False, + "navigation_or_safety_accepted": False, + }, + } + identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() + package_id = f"e37-worker-package-{identity_sha256}" + output = output_root.expanduser().absolute() + output.mkdir(mode=0o700, parents=True, exist_ok=True) + destination = output / package_id + if destination.exists(): + validate_e37_worker_package(destination) + return destination + + staging = output / f".{package_id}.{uuid.uuid4().hex}.tmp" + staging.mkdir(mode=0o700, exist_ok=False) + try: + for relative, source in sources.items(): + target = staging / relative + target.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + if source is None: + target.write_text(_GENERATED_COMPUTE_INIT, encoding="utf-8") + else: + shutil.copyfile(source, target) + artifacts = [ + { + "kind": relative, + "path": relative, + "byte_length": (staging / relative).stat().st_size, + "sha256": _sha256(staging / relative), + } + for relative in sorted(sources) + ] + manifest = { + "schema_version": E37_PACKAGE_SCHEMA, + "package_id": package_id, + "identity_sha256": identity_sha256, + "identity": identity, + "created_at_utc": datetime.now(UTC) + .isoformat(timespec="milliseconds") + .replace("+00:00", "Z"), + "artifacts": artifacts, + } + _write_json(staging / "manifest.json", manifest) + validate_e37_worker_package(staging, allow_staging=True) + os.replace(staging, destination) + except BaseException: + shutil.rmtree(staging, ignore_errors=True) + raise + validate_e37_worker_package(destination) + return destination + + +def validate_e37_worker_package( + root: Path, + *, + allow_staging: bool = False, +) -> dict[str, Any]: + """Validate package identity, exact file set, and every member digest.""" + + resolved = root.resolve(strict=True) + manifest = _read_json(resolved / "manifest.json") + identity = manifest.get("identity") + identity_sha256 = manifest.get("identity_sha256") + package_id = manifest.get("package_id") + artifacts = manifest.get("artifacts") + expected_name = ( + isinstance(package_id, str) + and ( + resolved.name == package_id + or ( + allow_staging + and resolved.name.startswith(f".{package_id}.") + and resolved.name.endswith(".tmp") + ) + ) + ) + if ( + manifest.get("schema_version") != E37_PACKAGE_SCHEMA + or not isinstance(identity, dict) + or not isinstance(identity_sha256, str) + or hashlib.sha256(_canonical_json(identity)).hexdigest() != identity_sha256 + or package_id != f"e37-worker-package-{identity_sha256}" + or not expected_name + or not isinstance(artifacts, list) + ): + raise E37WorkerPackageError("E37 worker package identity is invalid") + expected_paths = set(identity.get("artifact_paths", [])) + actual_paths = { + path.relative_to(resolved).as_posix() + for path in resolved.rglob("*") + if path.is_file() + } + if ( + not expected_paths + or actual_paths != expected_paths | {"manifest.json"} + or len(artifacts) != len(expected_paths) + ): + raise E37WorkerPackageError("E37 worker package file set changed") + observed: set[str] = set() + for row in artifacts: + if not isinstance(row, dict): + raise E37WorkerPackageError("E37 worker package artifact is invalid") + relative = row.get("path") + path = resolved / str(relative) + if ( + not isinstance(relative, str) + or relative not in expected_paths + or relative in observed + or Path(relative).is_absolute() + or ".." in Path(relative).parts + or not path.is_file() + or path.is_symlink() + or row.get("kind") != relative + or row.get("byte_length") != path.stat().st_size + or row.get("sha256") != _sha256(path) + ): + raise E37WorkerPackageError("E37 worker package artifact changed") + observed.add(relative) + if observed != expected_paths: + raise E37WorkerPackageError("E37 worker package artifact coverage changed") + return manifest + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode() + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + while chunk := stream.read(1024 * 1024): + digest.update(chunk) + return digest.hexdigest() + + +def _read_json(path: Path) -> dict[str, Any]: + value = json.loads(path.read_text(encoding="utf-8-sig")) + if not isinstance(value, dict): + raise E37WorkerPackageError(f"JSON object expected: {path.name}") + return value + + +def _write_json(path: Path, value: object) -> None: + with path.open("x", encoding="utf-8", newline="\n") as stream: + json.dump(value, stream, indent=2, sort_keys=True) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--repository-root", type=Path, required=True) + parser.add_argument("--materialization", type=Path, required=True) + parser.add_argument("--engineering-generation", type=Path, required=True) + parser.add_argument("--human-generation", type=Path, required=True) + parser.add_argument("--profile", type=Path, required=True) + parser.add_argument("--output-root", type=Path, required=True) + args = parser.parse_args() + package = build_e37_worker_package( + repository_root=args.repository_root, + materialization_root=args.materialization, + engineering_generation_root=args.engineering_generation, + human_generation_root=args.human_generation, + profile_path=args.profile, + output_root=args.output_root, + ) + print(package) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/experiments/perception/run_e37_acceptance_contract.py b/experiments/perception/run_e37_acceptance_contract.py new file mode 100644 index 0000000..8ef93c2 --- /dev/null +++ b/experiments/perception/run_e37_acceptance_contract.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +"""Build the immutable RAVNOVES00 R0 acceptance contract.""" + +from __future__ import annotations + +import argparse +import json +import os +from pathlib import Path + +from k1link.compute.e37_acceptance_contract import ( + build_e37_acceptance_contract, +) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--materialization", type=Path, required=True) + parser.add_argument("--engineering-generation", type=Path, required=True) + parser.add_argument("--human-generation", type=Path, required=True) + parser.add_argument("--profile", type=Path, required=True) + parser.add_argument("--output-root", type=Path, required=True) + parser.add_argument("--worker-node", default=os.environ.get("COMPUTERNAME")) + args = parser.parse_args() + result = build_e37_acceptance_contract( + materialization_root=args.materialization, + engineering_generation_root=args.engineering_generation, + human_generation_root=args.human_generation, + profile_path=args.profile, + output_root=args.output_root, + worker_node=args.worker_node, + ) + print( + json.dumps( + { + "result_id": result.result_id, + "result_root": str(result.result_root), + "accepted": result.accepted, + "metrics": result.report["metrics"], + "rejection_reasons": result.report["acceptance"][ + "rejection_reasons" + ], + }, + ensure_ascii=False, + sort_keys=True, + ) + ) + return 0 if result.accepted else 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/experiments/perception/worker/Invoke-E37AcceptanceContract.ps1 b/experiments/perception/worker/Invoke-E37AcceptanceContract.ps1 new file mode 100644 index 0000000..908995c --- /dev/null +++ b/experiments/perception/worker/Invoke-E37AcceptanceContract.ps1 @@ -0,0 +1,134 @@ +[CmdletBinding()] +param( + [Parameter(Mandatory = $true)] + [string]$PackageRoot, + [string]$OutputRoot = "D:\NDC_MISSIONCORE\runtime\derived\e37-acceptance", + [string]$ContainerImage = "nvcr.io/nvidia/tritonserver:26.06-py3@sha256:58df7489c3f2276f9591d500a012dee03e23d35543ce3c390b4c001e6bf90794", + [ValidateRange(1, 1000)] + [int]$FreeGiBFloor = 300 +) + +$ErrorActionPreference = "Stop" +$ProgressPreference = "SilentlyContinue" + +function Assert-LastExitCode([string]$Operation) { + if ($LASTEXITCODE -ne 0) { + throw "$Operation failed with exit code $LASTEXITCODE" + } +} + +function Resolve-DDirectory([string]$Path, [string]$Label) { + $item = Get-Item -LiteralPath (Resolve-Path -LiteralPath $Path).Path -Force + $root = [IO.Path]::GetPathRoot($item.FullName).TrimEnd("\") + if ( + -not $item.PSIsContainer -or + ($item.Attributes -band [IO.FileAttributes]::ReparsePoint) -or + $root -ine "D:" + ) { + throw "$Label must be a real D: directory" + } + return $item.FullName +} + +function Convert-ToDockerPath([string]$Path) { + return $Path.Replace("\", "/") +} + +function Assert-FreeSpace([string]$Phase) { + $free = [int64](Get-PSDrive -Name D).Free + $floor = [int64]$FreeGiBFloor * 1GB + Write-Host ( + "DISK_GUARD PHASE={0} DRIVE=D FREE_BYTES={1} FREE_GIB={2} FLOOR_GIB={3}" -f + $Phase, $free, [math]::Round($free / 1GB, 3), $FreeGiBFloor + ) + if ($free -lt ($floor + 1GB)) { + throw "D: lacks the guarded E37 reserve during $Phase" + } + return $free +} + +$package = Resolve-DDirectory $PackageRoot "E37 package" +$packageManifestPath = Join-Path $package "manifest.json" +if (-not (Test-Path -LiteralPath $packageManifestPath -PathType Leaf)) { + throw "E37 package manifest is missing" +} +$packageManifest = Get-Content -LiteralPath $packageManifestPath -Raw | + ConvertFrom-Json +if ( + $packageManifest.schema_version -ne "missioncore.e37-worker-package/v1" -or + $packageManifest.package_id -ne (Split-Path $package -Leaf) -or + $packageManifest.package_id -notmatch "^e37-worker-package-[a-f0-9]{64}$" +) { + throw "E37 package manifest is incompatible" +} + +if (-not (Test-Path -LiteralPath $OutputRoot)) { + $null = New-Item -ItemType Directory -Path $OutputRoot +} +$output = Resolve-DDirectory $OutputRoot "E37 output root" +$freeBefore = Assert-FreeSpace "preflight" + +& docker image inspect $ContainerImage *> $null +Assert-LastExitCode "Pinned E37 container image inspection" + +$dockerPackage = Convert-ToDockerPath $package +$dockerOutput = Convert-ToDockerPath $output +$packageName = Split-Path $package -Leaf +$containerPackage = "/opt/e37-input/$packageName" +$command = @( + "run", "--rm", + "--network", "none", + "--read-only", + "--security-opt", "no-new-privileges:true", + "--cap-drop", "ALL", + "--pids-limit", "128", + "--tmpfs", "/tmp:rw,noexec,nosuid,size=64m", + "-e", "PYTHONDONTWRITEBYTECODE=1", + "-e", ("PYTHONPATH={0}/runtime" -f $containerPackage), + "-e", ("E37_WORKER_NODE={0}" -f $env:COMPUTERNAME), + "-v", ("{0}:{1}:ro" -f $dockerPackage, $containerPackage), + "-v", ("{0}:/output:rw" -f $dockerOutput), + "--entrypoint", "python3", + $ContainerImage, + ("{0}/runtime/run_e37_acceptance_contract.py" -f $containerPackage), + "--package", $containerPackage, + "--output-root", "/output" +) + +Write-Output ("PACKAGE_ID={0}" -f $packageManifest.package_id) +Write-Output ("PACKAGE_IDENTITY_SHA256={0}" -f $packageManifest.identity_sha256) +Write-Output ("CONTAINER_IMAGE={0}" -f $ContainerImage) +& docker @command +Assert-LastExitCode "E37 acceptance contract" + +$matches = @( + Get-ChildItem -LiteralPath $output -Directory -Filter "e37-ravnoves-acceptance-*" | + Where-Object { + $manifestPath = Join-Path $_.FullName "manifest.json" + if (-not (Test-Path -LiteralPath $manifestPath -PathType Leaf)) { + return $false + } + $manifest = Get-Content -LiteralPath $manifestPath -Raw | + ConvertFrom-Json + return ( + $manifest.schema_version -eq + "missioncore.e37-acceptance-contract/v1" -and + $manifest.acceptance_state -eq + "accepted-r0-source-scoped-contract" -and + $manifest.identity.execution.worker_node -eq $env:COMPUTERNAME + ) + } +) +if ($matches.Count -ne 1) { + throw "E37 immutable result could not be resolved uniquely" +} +$resultRoot = $matches[0].FullName +$resultManifest = Get-Content -LiteralPath ( + Join-Path $resultRoot "manifest.json" +) -Raw | ConvertFrom-Json +$freeAfter = Assert-FreeSpace "completed" +Write-Output ("RESULT_ROOT={0}" -f $resultRoot) +Write-Output ("RESULT_ID={0}" -f $resultManifest.result_id) +Write-Output ("ACCEPTANCE_STATE={0}" -f $resultManifest.acceptance_state) +Write-Output ("DISK_FREE_BYTES_BEFORE={0}" -f $freeBefore) +Write-Output ("DISK_FREE_BYTES_AFTER={0}" -f $freeAfter) diff --git a/experiments/perception/worker/run_e37_acceptance_contract.py b/experiments/perception/worker/run_e37_acceptance_contract.py new file mode 100644 index 0000000..8b5d004 --- /dev/null +++ b/experiments/perception/worker/run_e37_acceptance_contract.py @@ -0,0 +1,128 @@ +#!/usr/bin/env python3 +"""Execute one immutable E37 package inside the bounded worker container.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +from pathlib import Path +from typing import Any + +from k1link.compute.e37_acceptance_contract import ( + E37_PACKAGE_SCHEMA, + build_e37_acceptance_contract, +) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--package", type=Path, required=True) + parser.add_argument("--output-root", type=Path, required=True) + args = parser.parse_args() + package = args.package.resolve(strict=True) + manifest = _validate_package(package) + source_ids = manifest["identity"]["source_ids"] + result = build_e37_acceptance_contract( + materialization_root=( + package / "input" / "materialization" / source_ids["materialization"] + ), + engineering_generation_root=( + package / "input" / "engineering" / source_ids["engineering"] + ), + human_generation_root=( + package / "input" / "human" / source_ids["human"] + ), + profile_path=package / "profile.json", + output_root=args.output_root, + worker_node=os.environ.get("E37_WORKER_NODE"), + ) + print( + json.dumps( + { + "package_id": manifest["package_id"], + "result_id": result.result_id, + "result_root": str(result.result_root), + "accepted": result.accepted, + "metrics": result.report["metrics"], + }, + ensure_ascii=False, + sort_keys=True, + ) + ) + return 0 if result.accepted else 2 + + +def _validate_package(root: Path) -> dict[str, Any]: + manifest = _read_json(root / "manifest.json") + identity = manifest.get("identity") + identity_sha256 = manifest.get("identity_sha256") + package_id = manifest.get("package_id") + artifacts = manifest.get("artifacts") + if ( + manifest.get("schema_version") != E37_PACKAGE_SCHEMA + or not isinstance(identity, dict) + or not isinstance(identity_sha256, str) + or hashlib.sha256(_canonical_json(identity)).hexdigest() != identity_sha256 + or package_id != f"e37-worker-package-{identity_sha256}" + or root.name != package_id + or not isinstance(artifacts, list) + ): + raise RuntimeError("E37 worker package identity is invalid") + expected = set(identity.get("artifact_paths", [])) + actual = { + path.relative_to(root).as_posix() + for path in root.rglob("*") + if path.is_file() + } + if actual != expected | {"manifest.json"} or len(artifacts) != len(expected): + raise RuntimeError("E37 worker package file set changed") + observed: set[str] = set() + for row in artifacts: + relative = row.get("path") if isinstance(row, dict) else None + path = root / str(relative) + if ( + not isinstance(relative, str) + or relative not in expected + or relative in observed + or Path(relative).is_absolute() + or ".." in Path(relative).parts + or not path.is_file() + or path.is_symlink() + or row.get("byte_length") != path.stat().st_size + or row.get("sha256") != _sha256(path) + ): + raise RuntimeError("E37 worker package artifact changed") + observed.add(relative) + if observed != expected: + raise RuntimeError("E37 worker package artifact coverage changed") + return manifest + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode() + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + while chunk := stream.read(1024 * 1024): + digest.update(chunk) + return digest.hexdigest() + + +def _read_json(path: Path) -> dict[str, Any]: + value = json.loads(path.read_text(encoding="utf-8-sig")) + if not isinstance(value, dict): + raise RuntimeError(f"JSON object expected: {path.name}") + return value + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/k1link/compute/e37_acceptance_contract.py b/src/k1link/compute/e37_acceptance_contract.py new file mode 100644 index 0000000..93625cb --- /dev/null +++ b/src/k1link/compute/e37_acceptance_contract.py @@ -0,0 +1,728 @@ +"""Freeze the RAVNOVES00 R0 acceptance denominator and validation split.""" + +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import uuid +from collections import Counter, defaultdict +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, Final, Literal + +E37_PROFILE_SCHEMA: Final = "missioncore.e37-acceptance-profile/v1" +E37_PACKAGE_SCHEMA: Final = "missioncore.e37-worker-package/v1" +E37_RESULT_SCHEMA: Final = "missioncore.e37-acceptance-contract/v1" +E37_ITEM_SCHEMA: Final = "missioncore.e37-acceptance-item/v1" +E37_REPORT_SCHEMA: Final = "missioncore.e37-acceptance-report/v1" +E37_CONTRACT_SCHEMA: Final = "missioncore.ravnoves00-acceptance-contract/v1" + +E37_ITEMS_NAME: Final = "acceptance-items.jsonl" +E37_CONTRACT_NAME: Final = "acceptance-contract.json" +E37_REPORT_NAME: Final = "run-report.json" +E37_MANIFEST_NAME: Final = "manifest.json" + +_MATERIALIZATION_SCHEMA: Final = "missioncore.e30-evidence-materialization/v2" +_ENGINEERING_SCHEMA: Final = "missioncore.e30-engineering-generation/v1" +_HUMAN_SCHEMA: Final = "missioncore.e30-human-review-generation/v2" +_SOURCE_SESSION_ID: Final = "20260720T065719Z_viewer_live" +_SOURCE_DISPLAY_NAME: Final = "RAVNOVES00" +_AUTHORITY: Final = { + "commands_enabled": False, + "navigation_or_safety_accepted": False, +} + +PresenceLabel = Literal[ + "object-present", + "occupied-environment", + "background-or-noise", + "unknown", +] +GeometryLabel = Literal[ + "object-associated", + "independent-occupied", + "rejected-nonobject", + "insufficient-support", + "unknown", +] +FreshnessLabel = Literal["current", "held", "stale", "unavailable", "unknown"] +SplitName = Literal["development", "validation"] + + +class E37AcceptanceContractError(RuntimeError): + """The R0 profile, source review, or immutable result is invalid.""" + + +@dataclass(frozen=True, slots=True) +class E37AcceptanceContract: + result_id: str + result_root: Path + manifest: dict[str, Any] + contract: dict[str, Any] + report: dict[str, Any] + + @property + def accepted(self) -> bool: + return self.report.get("acceptance", {}).get("accepted") is True + + +def derive_reference_labels( + decision: dict[str, Any], + human_disposition: str | None, +) -> tuple[PresenceLabel, GeometryLabel, FreshnessLabel]: + """Project the reviewed E30 decision into the frozen R0 task ontology.""" + + detector = decision.get("detector_assessment") + ownership = decision.get("point_ownership") + cause = decision.get("cause_code") + effective_stratum = decision.get("effective_stratum") + + if human_disposition == "object-present": + presence: PresenceLabel = "occupied-environment" + geometry: GeometryLabel = "independent-occupied" + elif human_disposition == "background-or-noise": + presence = "background-or-noise" + geometry = "rejected-nonobject" + elif human_disposition == "insufficient-evidence": + presence = "unknown" + geometry = "unknown" + elif detector in {"valid", "class-mismatch", "missed-object"}: + presence = "object-present" + geometry = _geometry_label(ownership) + elif detector == "false-positive": + presence = "background-or-noise" + geometry = _geometry_label(ownership) + elif detector == "not-applicable": + if ownership == "object": + presence = "object-present" + elif ownership == "static-environment": + presence = "occupied-environment" + elif ownership in {"surface-or-background", "self"}: + presence = "background-or-noise" + else: + presence = "unknown" + geometry = _geometry_label(ownership) + else: + presence = "unknown" + geometry = _geometry_label(ownership) + + if cause == "time_mismatch": + freshness: FreshnessLabel = "stale" + elif effective_stratum in {"camera-only", "unknown"} or ownership in { + "insufficient-support", + "insufficient-evidence", + }: + freshness = "unavailable" + else: + freshness = "current" + return presence, geometry, freshness + + +def assign_split( + rows: list[dict[str, Any]], + *, + seed: str, + validation_fraction: float, +) -> dict[str, SplitName]: + """Assign a deterministic source-stratum/range-balanced holdout.""" + + if not seed or not 0.1 <= validation_fraction <= 0.5: + raise E37AcceptanceContractError("E37 split profile is invalid") + grouped: dict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list) + for row in rows: + item_id = row.get("item_id") + stratum = row.get("source_stratum") + range_bucket = row.get("range_bucket") + if not all(isinstance(value, str) and value for value in ( + item_id, + stratum, + range_bucket, + )): + raise E37AcceptanceContractError("E37 split source row is invalid") + grouped[(stratum, range_bucket)].append(row) + + assignments: dict[str, SplitName] = {} + for group_rows in grouped.values(): + ordered = sorted( + group_rows, + key=lambda row: hashlib.sha256( + f"{seed}:{row['item_id']}".encode() + ).hexdigest(), + ) + validation_count = round(len(ordered) * validation_fraction) + if len(ordered) > 1: + validation_count = min(len(ordered) - 1, max(1, validation_count)) + else: + validation_count = 0 + for index, row in enumerate(ordered): + assignments[row["item_id"]] = ( + "validation" if index < validation_count else "development" + ) + if set(assignments) != {row["item_id"] for row in rows}: + raise E37AcceptanceContractError("E37 split accounting is incomplete") + return assignments + + +def build_e37_acceptance_contract( + *, + materialization_root: Path, + engineering_generation_root: Path, + human_generation_root: Path, + profile_path: Path, + output_root: Path, + worker_node: str | None = None, +) -> E37AcceptanceContract: + """Build or verify one immutable source-scoped R0 contract.""" + + profile_file = profile_path.resolve(strict=True) + profile = _read_json(profile_file) + _validate_profile(profile) + materialization = materialization_root.resolve(strict=True) + engineering = engineering_generation_root.resolve(strict=True) + human = human_generation_root.resolve(strict=True) + inputs = _load_inputs(materialization, engineering, human, profile) + + identity = { + "schema_version": E37_RESULT_SCHEMA, + "source": { + "session_id": _SOURCE_SESSION_ID, + "display_name": _SOURCE_DISPLAY_NAME, + "classification": "immutable-private-physical-recording", + }, + "profile": { + "profile_id": profile["profile_id"], + "sha256": _sha256(profile_file), + }, + "reviewed_substrate": inputs["bindings"], + "split": profile["split"], + "ontology": profile["ontology"], + "metrics": profile["metrics"], + "severity": profile["severity"], + "execution": { + "class": "deterministic-offline-contract-build", + "worker_node": worker_node or "unbound-local", + }, + "authority": _AUTHORITY, + } + identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() + result_id = f"e37-ravnoves-acceptance-{identity_sha256}" + destination = output_root.expanduser().absolute() / result_id + if destination.exists(): + return read_e37_acceptance_contract(destination) + + materialization_by_id = { + row["item_id"]: row for row in inputs["materialization_rows"] + } + engineering_rows = inputs["engineering_rows"] + human_by_id = { + row["item_id"]: row for row in inputs["human_rows"] + } + joined_for_split = [] + for decision in engineering_rows: + source = materialization_by_id.get(decision["item_id"]) + if source is None: + raise E37AcceptanceContractError( + "E37 engineering/materialization accounting differs" + ) + joined_for_split.append( + { + "item_id": decision["item_id"], + "source_stratum": decision["source_stratum"], + "range_bucket": source["range_bucket"], + } + ) + assignments = assign_split( + joined_for_split, + seed=profile["split"]["seed"], + validation_fraction=float(profile["split"]["validation_fraction"]), + ) + + items: list[dict[str, Any]] = [] + for decision in engineering_rows: + item_id = decision["item_id"] + source = materialization_by_id[item_id] + human_row = human_by_id.get(item_id) + human_disposition = ( + human_row.get("disposition") if human_row is not None else None + ) + if decision.get("human_exception_required") is True and human_row is None: + raise E37AcceptanceContractError("E37 human exception is unresolved") + if decision.get("human_exception_required") is not True and human_row is not None: + raise E37AcceptanceContractError("E37 human review escaped its exception set") + presence, geometry, freshness = derive_reference_labels( + decision, + human_disposition, + ) + item = { + "schema_version": E37_ITEM_SCHEMA, + "sequence": len(items), + "item_id": item_id, + "review_key": decision["review_key"], + "source_frame_index": source["evidence_binding"]["source_frame_index"], + "session_seconds": source["evidence_binding"]["session_seconds"], + "source_stratum": decision["source_stratum"], + "range_bucket": source["range_bucket"], + "severity": _severity(decision, human_disposition), + "split": assignments[item_id], + "reference": { + "presence": presence, + "geometry_association": geometry, + "freshness": freshness, + }, + "provenance": { + "engineering_verdict": decision["verdict"], + "engineering_confidence": decision["confidence"], + "human_exception": human_row is not None, + "human_disposition": human_disposition, + }, + "authority": _AUTHORITY, + } + items.append(item) + + contract = _contract_document( + result_id=result_id, + identity_sha256=identity_sha256, + profile=profile, + items=items, + ) + report = _report_document( + result_id=result_id, + identity_sha256=identity_sha256, + profile=profile, + items=items, + execution=identity["execution"], + ) + + destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + staging = destination.parent / f".{result_id}.{uuid.uuid4().hex}.tmp" + staging.mkdir(mode=0o700, exist_ok=False) + try: + _write_jsonl(staging / E37_ITEMS_NAME, items) + _write_json(staging / E37_CONTRACT_NAME, contract) + _write_json(staging / E37_REPORT_NAME, report) + artifacts = [ + _artifact(staging / E37_ITEMS_NAME, "reviewed-denominator"), + _artifact(staging / E37_CONTRACT_NAME, "acceptance-contract"), + _artifact(staging / E37_REPORT_NAME, "run-report"), + ] + manifest = { + "schema_version": E37_RESULT_SCHEMA, + "result_id": result_id, + "identity_sha256": identity_sha256, + "identity": identity, + "created_at_utc": _utc_now(), + "acceptance_state": "accepted-r0-source-scoped-contract", + "artifacts": artifacts, + } + _write_json(staging / E37_MANIFEST_NAME, manifest) + os.replace(staging, destination) + except BaseException: + shutil.rmtree(staging, ignore_errors=True) + raise + return read_e37_acceptance_contract(destination) + + +def read_e37_acceptance_contract(root: Path) -> E37AcceptanceContract: + resolved = root.resolve(strict=True) + manifest = _read_json(resolved / E37_MANIFEST_NAME) + identity = _object(manifest.get("identity"), "E37 identity") + identity_sha256 = manifest.get("identity_sha256") + if ( + manifest.get("schema_version") != E37_RESULT_SCHEMA + or not isinstance(identity_sha256, str) + or hashlib.sha256(_canonical_json(identity)).hexdigest() != identity_sha256 + or manifest.get("result_id") != f"e37-ravnoves-acceptance-{identity_sha256}" + or resolved.name != manifest.get("result_id") + or manifest.get("acceptance_state") != "accepted-r0-source-scoped-contract" + or identity.get("authority") != _AUTHORITY + ): + raise E37AcceptanceContractError("E37 result identity is invalid") + artifacts = manifest.get("artifacts") + if not isinstance(artifacts, list) or len(artifacts) != 3: + raise E37AcceptanceContractError("E37 artifact catalog is invalid") + expected = { + E37_ITEMS_NAME: "reviewed-denominator", + E37_CONTRACT_NAME: "acceptance-contract", + E37_REPORT_NAME: "run-report", + } + for row in artifacts: + if not isinstance(row, dict): + raise E37AcceptanceContractError("E37 artifact descriptor is invalid") + name = row.get("path") + if name not in expected or row.get("role") != expected[name]: + raise E37AcceptanceContractError("E37 artifact role changed") + path = resolved / name + if ( + not path.is_file() + or path.is_symlink() + or row.get("byte_length") != path.stat().st_size + or row.get("sha256") != _sha256(path) + ): + raise E37AcceptanceContractError("E37 artifact content changed") + contract = _read_json(resolved / E37_CONTRACT_NAME) + report = _read_json(resolved / E37_REPORT_NAME) + if ( + contract.get("schema_version") != E37_CONTRACT_SCHEMA + or report.get("schema_version") != E37_REPORT_SCHEMA + or contract.get("result_id") != resolved.name + or report.get("result_id") != resolved.name + or report.get("acceptance", {}).get("accepted") is not True + ): + raise E37AcceptanceContractError("E37 contract or report is invalid") + return E37AcceptanceContract( + result_id=resolved.name, + result_root=resolved, + manifest=manifest, + contract=contract, + report=report, + ) + + +def _geometry_label(value: object) -> GeometryLabel: + return { + "object": "object-associated", + "static-environment": "independent-occupied", + "surface-or-background": "rejected-nonobject", + "self": "rejected-nonobject", + "insufficient-support": "insufficient-support", + "insufficient-evidence": "unknown", + }.get(str(value), "unknown") # type: ignore[return-value] + + +def _severity( + decision: dict[str, Any], + human_disposition: str | None, +) -> str: + if human_disposition is not None: + return "high" + if decision.get("cause_code") in {"time_mismatch", "self_points"}: + return "high" + if decision.get("detector_assessment") == "missed-object": + return "high" + if decision.get("verdict") == "corrected": + return "medium" + return "standard" + + +def _load_inputs( + materialization: Path, + engineering: Path, + human: Path, + profile: dict[str, Any], +) -> dict[str, Any]: + materialization_manifest = _read_json(materialization / "manifest.json") + engineering_manifest = _read_json(engineering / "manifest.json") + human_manifest = _read_json(human / "manifest.json") + if ( + materialization_manifest.get("schema_version") != _MATERIALIZATION_SCHEMA + or materialization.name != profile["source"]["materialization_id"] + or materialization_manifest.get("result_id") != materialization.name + or engineering_manifest.get("schema_version") != _ENGINEERING_SCHEMA + or engineering.name != profile["source"]["engineering_generation_id"] + or engineering_manifest.get("result_id") != engineering.name + or human_manifest.get("schema_version") != _HUMAN_SCHEMA + or human.name != profile["source"]["human_generation_id"] + or human_manifest.get("result_id") != human.name + ): + raise E37AcceptanceContractError("E37 input identity is invalid") + materialization_index = materialization / "materialized-items.jsonl" + engineering_decisions = engineering / "engineering-decisions.jsonl" + human_decisions = human / "review-decisions.jsonl" + for path in (materialization_index, engineering_decisions, human_decisions): + if not path.is_file() or path.is_symlink(): + raise E37AcceptanceContractError("E37 input artifact is unavailable") + materialization_rows = _read_jsonl(materialization_index) + engineering_rows = _read_jsonl(engineering_decisions) + human_rows = _read_jsonl(human_decisions) + expected = int(profile["denominator"]["expected_items"]) + if ( + len(materialization_rows) != expected + or len(engineering_rows) != expected + or len({row.get("item_id") for row in materialization_rows}) != expected + or len({row.get("item_id") for row in engineering_rows}) != expected + or {row.get("item_id") for row in materialization_rows} + != {row.get("item_id") for row in engineering_rows} + or {row.get("item_id") for row in human_rows} + != { + row.get("item_id") + for row in engineering_rows + if row.get("human_exception_required") is True + } + ): + raise E37AcceptanceContractError("E37 reviewed denominator is incomplete") + source_binding = materialization_manifest.get("identity", {}).get("source", {}) + if source_binding.get("source_session_id") != _SOURCE_SESSION_ID: + raise E37AcceptanceContractError("E37 source session changed") + return { + "materialization_rows": materialization_rows, + "engineering_rows": engineering_rows, + "human_rows": human_rows, + "bindings": { + "materialization_id": materialization.name, + "materialization_identity_sha256": materialization_manifest.get( + "identity_sha256" + ), + "materialization_manifest_sha256": _sha256( + materialization / "manifest.json" + ), + "materialization_index_sha256": _sha256(materialization_index), + "engineering_generation_id": engineering.name, + "engineering_identity_sha256": engineering_manifest.get( + "identity_sha256" + ), + "engineering_manifest_sha256": _sha256(engineering / "manifest.json"), + "engineering_decisions_sha256": _sha256(engineering_decisions), + "human_generation_id": human.name, + "human_identity_sha256": human_manifest.get("identity_sha256"), + "human_manifest_sha256": _sha256(human / "manifest.json"), + "human_decisions_sha256": _sha256(human_decisions), + }, + } + + +def _validate_profile(profile: dict[str, Any]) -> None: + source = _object(profile.get("source"), "E37 source") + denominator = _object(profile.get("denominator"), "E37 denominator") + split = _object(profile.get("split"), "E37 split") + metrics = _object(profile.get("metrics"), "E37 metrics") + authority = _object(profile.get("authority"), "E37 authority") + ontology = _object(profile.get("ontology"), "E37 ontology") + if ( + profile.get("schema_version") != E37_PROFILE_SCHEMA + or profile.get("profile_id") != "e37-ravnoves00-r0-acceptance/v1" + or source.get("session_id") != _SOURCE_SESSION_ID + or source.get("display_name") != _SOURCE_DISPLAY_NAME + or denominator.get("expected_items") != 486 + or split.get("strategy") + != "deterministic-source-stratum-range-holdout" + or not isinstance(split.get("seed"), str) + or not 0.1 <= float(split.get("validation_fraction", 0)) <= 0.5 + or metrics.get("presence_target") != 0.9 + or metrics.get("geometry_association_target") != 0.9 + or metrics.get("freshness_target") != 0.9 + or metrics.get("accounting_target") != 1.0 + or metrics.get("maximum_false_free_claims") != 0 + or authority != _AUTHORITY + or sorted(ontology) != [ + "freshness", + "geometry_association", + "presence", + ] + ): + raise E37AcceptanceContractError("E37 profile contract changed") + + +def _contract_document( + *, + result_id: str, + identity_sha256: str, + profile: dict[str, Any], + items: list[dict[str, Any]], +) -> dict[str, Any]: + split_counts = Counter(item["split"] for item in items) + dimensions = { + name: Counter(item["reference"][name] for item in items) + for name in ("presence", "geometry_association", "freshness") + } + return { + "schema_version": E37_CONTRACT_SCHEMA, + "result_id": result_id, + "identity_sha256": identity_sha256, + "source_session_id": _SOURCE_SESSION_ID, + "denominator": { + "reviewed_items": len(items), + "development_items": split_counts["development"], + "validation_items": split_counts["validation"], + "terminal_outcomes": len(items), + "accounting_fraction": 1.0, + }, + "split": profile["split"], + "ontology": profile["ontology"], + "dimension_distributions": { + name: dict(sorted(counts.items())) + for name, counts in dimensions.items() + }, + "targets": profile["metrics"], + "severity_distribution": dict( + sorted(Counter(item["severity"] for item in items).items()) + ), + "label_provenance": { + "engineering_items": sum( + not item["provenance"]["human_exception"] for item in items + ), + "human_exception_items": sum( + item["provenance"]["human_exception"] for item in items + ), + "independent_ground_truth": False, + }, + "authority": _AUTHORITY, + } + + +def _report_document( + *, + result_id: str, + identity_sha256: str, + profile: dict[str, Any], + items: list[dict[str, Any]], + execution: dict[str, Any], +) -> dict[str, Any]: + split_counts = Counter(item["split"] for item in items) + false_free_claims = sum( + value == "free" + for item in items + for value in item["reference"].values() + ) + checks = { + "source_identity_frozen": True, + "reviewed_denominator_complete": len(items) == 486, + "development_validation_split_complete": ( + split_counts["development"] + split_counts["validation"] == len(items) + and split_counts["development"] > 0 + and split_counts["validation"] > 0 + ), + "every_dimension_has_terminal_label": all( + len(item["reference"]) == 3 + and all(isinstance(value, str) and value for value in item["reference"].values()) + for item in items + ), + "human_exception_accounting_complete": sum( + item["provenance"]["human_exception"] for item in items + ) + == 2, + "false_free_claims_zero": false_free_claims == 0, + "authority_remains_diagnostic": True, + } + accepted = all(checks.values()) + return { + "schema_version": E37_REPORT_SCHEMA, + "result_id": result_id, + "identity_sha256": identity_sha256, + "status": ( + "accepted-r0-source-scoped-contract" + if accepted + else "rejected-r0-source-scoped-contract" + ), + "source_session_id": _SOURCE_SESSION_ID, + "profile_id": profile["profile_id"], + "execution": execution, + "metrics": { + "reviewed_items": len(items), + "development_items": split_counts["development"], + "validation_items": split_counts["validation"], + "engineering_items": len(items) - 2, + "human_exception_items": 2, + "terminal_outcomes": len(items), + "accounting_fraction": 1.0, + "false_free_claims": false_free_claims, + }, + "acceptance": { + "accepted": accepted, + "checks": checks, + "rejection_reasons": [ + name for name, passed in checks.items() if not passed + ], + }, + "decision": { + "r0_contract_frozen": accepted, + "quality_target_evaluated": False, + "next_gate": "R1 source-scoped perception quality baseline", + }, + "limitations": [ + ( + "labels are an engineering-reviewed source-scoped substrate, " + "not independent ground truth" + ), + ( + "R0 freezes evaluation and does not claim that any 90 percent " + "quality target has passed" + ), + "RAVNOVES00 does not prove another route, camera, rig or mount", + "navigation, command and safety authority remain false", + ], + "authority": _AUTHORITY, + } + + +def _artifact(path: Path, role: str) -> dict[str, Any]: + return { + "role": role, + "path": path.name, + "byte_length": path.stat().st_size, + "sha256": _sha256(path), + } + + +def _object(value: object, label: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise E37AcceptanceContractError(f"{label} must be an object") + return value + + +def _read_json(path: Path) -> dict[str, Any]: + try: + value = json.loads(path.read_text(encoding="utf-8-sig")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + raise E37AcceptanceContractError(f"invalid JSON: {path.name}") from exc + return _object(value, path.name) + + +def _read_jsonl(path: Path) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + try: + with path.open("r", encoding="utf-8-sig") as stream: + for line in stream: + rows.append(_object(json.loads(line), path.name)) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + raise E37AcceptanceContractError(f"invalid JSONL: {path.name}") from exc + return rows + + +def _write_json(path: Path, value: object) -> None: + with path.open("x", encoding="utf-8", newline="\n") as stream: + json.dump(value, stream, ensure_ascii=False, indent=2, sort_keys=True) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + + +def _write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None: + with path.open("x", encoding="utf-8", newline="\n") as stream: + for row in rows: + stream.write( + json.dumps( + row, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ) + ) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode() + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + while chunk := stream.read(1024 * 1024): + digest.update(chunk) + return digest.hexdigest() + + +def _utc_now() -> str: + return datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z") diff --git a/tests/test_e37_acceptance_contract.py b/tests/test_e37_acceptance_contract.py new file mode 100644 index 0000000..e67168c --- /dev/null +++ b/tests/test_e37_acceptance_contract.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +from collections import Counter + +import pytest + +from k1link.compute.e37_acceptance_contract import ( + E37AcceptanceContractError, + assign_split, + derive_reference_labels, +) + + +def _decision( + *, + detector: str, + ownership: str, + cause: str | None = None, + effective_stratum: str | None = "agree", +) -> dict[str, object]: + return { + "detector_assessment": detector, + "point_ownership": ownership, + "cause_code": cause, + "effective_stratum": effective_stratum, + } + + +def test_r0_reference_projection_keeps_dimensions_independent() -> None: + assert derive_reference_labels( + _decision(detector="valid", ownership="object"), + None, + ) == ("object-present", "object-associated", "current") + assert derive_reference_labels( + _decision( + detector="false-positive", + ownership="surface-or-background", + cause="time_mismatch", + ), + None, + ) == ("background-or-noise", "rejected-nonobject", "stale") + assert derive_reference_labels( + _decision( + detector="not-applicable", + ownership="insufficient-support", + effective_stratum="camera-only", + ), + None, + ) == ("unknown", "insufficient-support", "unavailable") + + +def test_human_exception_only_overrides_the_ambiguous_reference() -> None: + ambiguous = _decision( + detector="insufficient-evidence", + ownership="insufficient-evidence", + effective_stratum=None, + ) + assert derive_reference_labels(ambiguous, "object-present") == ( + "occupied-environment", + "independent-occupied", + "unavailable", + ) + assert derive_reference_labels(ambiguous, "background-or-noise") == ( + "background-or-noise", + "rejected-nonobject", + "unavailable", + ) + + +def test_split_is_deterministic_balanced_and_complete() -> None: + rows = [ + { + "item_id": f"item-{index:03d}", + "source_stratum": "agree" if index % 2 else "conflict", + "range_bucket": "near" if index % 3 else "far", + } + for index in range(60) + ] + first = assign_split( + rows, + seed="frozen-seed", + validation_fraction=0.3, + ) + second = assign_split( + list(reversed(rows)), + seed="frozen-seed", + validation_fraction=0.3, + ) + assert first == second + assert set(first) == {row["item_id"] for row in rows} + assert Counter(first.values())["validation"] == 18 + + +def test_split_rejects_mutable_or_invalid_fraction() -> None: + with pytest.raises(E37AcceptanceContractError): + assign_split([], seed="", validation_fraction=0.3) + with pytest.raises(E37AcceptanceContractError): + assign_split([], seed="seed", validation_fraction=0.9) diff --git a/tests/test_e37_worker_package.py b/tests/test_e37_worker_package.py new file mode 100644 index 0000000..54415a8 --- /dev/null +++ b/tests/test_e37_worker_package.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +import importlib.util +import subprocess +import sys +from pathlib import Path + + +def _module() -> object: + path = ( + Path(__file__).resolve().parents[1] + / "experiments" + / "perception" + / "prepare_e37_worker_package.py" + ) + spec = importlib.util.spec_from_file_location("e37_worker_package_test", path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def test_e37_package_is_minimal_content_addressed_projection(tmp_path: Path) -> None: + module = _module() + repository = Path(__file__).resolve().parents[1] + runtime = repository / ".runtime" / "compute-experiments" / "e30" + package = module.build_e37_worker_package( + repository_root=repository, + materialization_root=( + runtime + / "materializations" + / ( + "e30-materialization-" + "841af926d8d28ab93538c46d8f31278a2234c4d1c12c7dc4dc296b249d59735a" + ) + ), + engineering_generation_root=( + runtime + / "engineering-generations" + / ( + "e30-engineering-generation-" + "62a4fea10dea9b77f69ceac1af5bf0e4928d9c7716083c22258a03670fe5bd4f" + ) + ), + human_generation_root=( + runtime + / "human-review-generations" + / ( + "e30-review-generation-" + "7982a882558d0be690b4c7092e328c080bfcbf52478a220452be7e887a588250" + ) + ), + profile_path=( + repository + / "experiments" + / "perception" + / "e37_ravnoves00_acceptance_profile.json" + ), + output_root=tmp_path, + ) + manifest = module.validate_e37_worker_package(package) + assert package.name == f"e37-worker-package-{manifest['identity_sha256']}" + assert manifest["identity"]["classification"] == ( + "immutable-ravnoves00-r0-worker-input" + ) + assert len(manifest["artifacts"]) == 12 + completed = subprocess.run( + [ + sys.executable, + str(package / "runtime" / "run_e37_acceptance_contract.py"), + "--package", + str(package), + "--output-root", + str(tmp_path / "results"), + ], + env={ + "PYTHONPATH": str(package / "runtime"), + "PYTHONDONTWRITEBYTECODE": "1", + "E37_WORKER_NODE": "TEST-WORKER-006", + }, + check=False, + capture_output=True, + text=True, + ) + assert completed.returncode == 0, completed.stderr + assert '"accepted": true' in completed.stdout