diff --git a/benchmarks/README.md b/benchmarks/README.md index c23f980c3..f2adaf1c3 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -356,6 +356,27 @@ Terminal teardown proof: make -C benchmarks progressive-host-ladder-inventory OUTPUT_DIR=$OUTPUT_DIR WORK_ROOT=$WORK_ROOT ``` +This emits `graphforge-host-work-root-inventory/2` from an actual filesystem +inventory and binds it automatically to the existing native result receipts. +It checks the whole work root, including `tmp/`, failed staging directories, +unexpected files, and dangling links. Only the explicitly retained evidence +directory and empty `workspace/` / `tmp/` scaffolding are excluded; the scope is +recorded in the document. Keep evidence outside the work root or in one direct +child directory. An unreadable tree is an error, never an empty inventory. + +Ingest that same output directory with `ingest-ladder-bundle`; native receipts +need no Fly image, provider teardown, or additional manifest. Plans, results, +phase artifacts, admission projections, and cleanup inventory must agree by +identity and digest. Historical provider bundles remain readable as migration +fixtures. Legacy unbound native inventory v1 cannot establish full completion. + +`parity-gate` reports structural retirement, accepted prefix parity, and +`full_ladder_evidence_complete` separately. A valid S18/S19 prefix is useful +engineering evidence but does not establish full harness authority or complete +#959. Full completion requires all seven native rungs through S26 and empty +terminal work-root inventory; it does not claim an official Graph500 submission +or replace independent review of the actual #900 lifecycle/parity evidence. + ## Progressive Graph500 qualification **Graph500-compliant generated input; GraphForge lifecycle measurements; no diff --git a/benchmarks/harness/graphforge_bench/ladder_bundle_ingest.py b/benchmarks/harness/graphforge_bench/ladder_bundle_ingest.py index 79f19a0b5..ba4c2c286 100644 --- a/benchmarks/harness/graphforge_bench/ladder_bundle_ingest.py +++ b/benchmarks/harness/graphforge_bench/ladder_bundle_ingest.py @@ -11,7 +11,7 @@ from jsonschema import Draft202012Validator -from graphforge_bench.parity_gate import ladder_bundle_root +from graphforge_bench.native_ladder_bundle import NativeBundleError, validate_native_bundle from graphforge_bench.scale_parity import compare_ladder_bundle RUNG_NAME = re.compile(r"^s(\d+)-rung\.json$") @@ -57,11 +57,41 @@ def _validate_manifest(document: Any) -> None: raise LadderBundleIngestError("manifest.json commit must be a lowercase Git object ID") +def _has_native_receipts(source: Path) -> bool: + for path in (*source.glob("*-result.json"), *source.glob("*-plan.json")): + document = _read_json(path, f"{path.name} is malformed") + if isinstance(document, dict) and document.get("schema") in { + "graphforge-progressive-host-run-plan/1", + "graphforge-progressive-host-run-result/1", + }: + return True + return False + + def validate_ladder_bundle(source: Path) -> dict[str, Any]: """Validate a completed #900 bundle directory without copying it.""" if not source.is_dir(): raise LadderBundleIngestError("source ladder bundle directory is missing") + if ( + not (source / "manifest.json").exists() + or (source / "work-root-inventory.json").exists() + or _has_native_receipts(source) + ): + try: + native = validate_native_bundle(source) + except (NativeBundleError, OSError) as error: + raise LadderBundleIngestError(str(error)) from error + return { + "schema": INGEST_SCHEMA, + "source": str(source), + "manifest_commit": native["commit"], + "rung_files": [f"s{scale}-rung.json" for scale in native["scales"]], + "rung_scales": native["scales"], + "teardown_status": "empty" if native["empty"] else "failed", + "evidence_files": native["files"], + } + manifest_path = source / "manifest.json" teardown_path = source / "teardown-inventory.json" if not manifest_path.is_file(): @@ -104,12 +134,14 @@ def validate_ladder_bundle(source: Path) -> dict[str, Any]: def ingest_ladder_bundle(source: Path, destination: Path | None = None) -> dict[str, Any]: """Validate a #900 bundle and copy it into the parity fixture tree.""" report = validate_ladder_bundle(source) - target = destination or ladder_bundle_root() + target = destination or Path(__file__).resolve().parents[2] / "fixtures/parity/ladder-bundle" if target.exists() and any(target.glob("*-rung.json")): raise LadderBundleIngestError("destination already contains ingested rung bundles") target.mkdir(parents=True, exist_ok=True) - for name in ("manifest.json", "teardown-inventory.json", *report["rung_files"]): + for name in report.get( + "evidence_files", ("manifest.json", "teardown-inventory.json", *report["rung_files"]) + ): shutil.copy2(source / name, target / name) parity = compare_ladder_bundle(target) diff --git a/benchmarks/harness/graphforge_bench/native_ladder_bundle.py b/benchmarks/harness/graphforge_bench/native_ladder_bundle.py new file mode 100644 index 000000000..eeefbf02f --- /dev/null +++ b/benchmarks/harness/graphforge_bench/native_ladder_bundle.py @@ -0,0 +1,151 @@ +"""Read native ladder receipts and bind cleanup inventory to those exact results.""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path +from typing import Any + +from jsonschema import Draft202012Validator + +from graphforge_bench.native_rung import NativeRungError, read_native_rung +from graphforge_bench.progressive_provider_attempt import CANONICAL_RUNGS + +INVENTORY_SCHEMA = "graphforge-host-work-root-inventory/2" +HOST_PROFILE_ID = "local-linux-cgroups-v2" + + +class NativeBundleError(ValueError): + """Native evidence does not establish a consistent completed prefix.""" + + +def digest(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def read_object(path: Path) -> dict[str, Any]: + try: + if path.is_symlink(): + raise NativeBundleError(f"linked evidence: {path.name}") + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, ValueError) as error: + raise NativeBundleError(f"invalid evidence: {path.name}") from error + if not isinstance(value, dict): + raise NativeBundleError(f"evidence is not an object: {path.name}") + return value + + +def validate_schema(document: dict[str, Any], name: str) -> None: + root = Path(__file__).resolve().parents[2] / "schemas" + error = next(Draft202012Validator(read_object(root / name)).iter_errors(document), None) + if error is not None: + raise NativeBundleError(f"{name}: {error.message}") + + +def native_receipts(source: Path) -> dict[str, Any]: + """Validate existing producer outputs; no second manifest or approval is needed.""" + paths = sorted(source.glob("*-rung.json")) + scales = list(CANONICAL_RUNGS[: len(paths)]) + if ( + not paths + or len(paths) > len(CANONICAL_RUNGS) + or [p.name for p in paths] != [f"s{scale}-rung.json" for scale in scales] + ): + raise NativeBundleError("native rung files are not a canonical prefix") + expected_results = [f"s{scale}-result.json" for scale in scales] + if sorted(p.name for p in source.glob("*-result.json")) != expected_results: + raise NativeBundleError("native result files contradict the completed prefix") + files: list[str] = [] + results: dict[str, str] = {} + common: dict[str, Any] | None = None + for scale in scales: + prefix = f"s{scale}" + result_path = source / f"{prefix}-result.json" + try: + documents = read_native_rung(Path(__file__).resolve().parents[2], source, scale) + except NativeRungError as error: + raise NativeBundleError(str(error)) from error + result = documents["result"] + identities = result["identities"] + shared = { + key: value + for key, value in identities.items() + if key + not in { + "profile_id", + "profile_sha256", + "admitted_projection_sha256", + } + } + if common is not None and common != shared: + raise NativeBundleError("native rung immutable identities differ") + common = shared + files.extend( + f"{prefix}-{kind}.json" for kind in ("plan", "benchexec", "graphforge", "rung") + ) + if scale >= 20: + files.append(f"{prefix}-projection.json") + results[result_path.name] = digest(result_path) + files.append(result_path.name) + assert common is not None + return {"commit": common["commit"], "scales": scales, "results": results, "files": files} + + +def validate_native_bundle(source: Path) -> dict[str, Any]: + receipt = native_receipts(source) + inventory = read_object(source / "work-root-inventory.json") + validate_schema(inventory, "host-work-root-inventory.json") + if inventory["result_sha256"] != receipt["results"]: + raise NativeBundleError("cleanup inventory belongs to different native results") + if inventory["empty"] != (inventory["entries"] == []): + raise NativeBundleError("cleanup inventory contradicts its entries") + receipt["files"].append("work-root-inventory.json") + receipt["empty"] = inventory["empty"] + receipt["complete"] = inventory["empty"] and receipt["scales"] == list(CANONICAL_RUNGS) + return receipt + + +def collect_inventory(work_root: Path, output_dir: Path | None = None) -> dict[str, Any]: + """Inspect the whole work root, retaining only the named evidence directory. + + Empty workspace/tmp scaffold directories are harmless. Every other entry + denotes debris, including an entire remaining subtree. We do not enumerate + dataset contents. No directory links are followed; unreadable scaffolding + fails rather than producing an empty inventory. + """ + work_root = work_root.resolve(strict=True) + evidence = output_dir.resolve(strict=True) if output_dir is not None else None + if evidence is not None and (evidence == work_root or work_root.is_relative_to(evidence)): + raise NativeBundleError("evidence directory must not contain the work root") + if evidence is not None and evidence.is_relative_to(work_root) and evidence.parent != work_root: + raise NativeBundleError("retained evidence must be a direct work-root child") + entries = [] + + def visit(directory: Path) -> None: + for path in sorted(directory.iterdir()): + if path == evidence and not path.is_symlink(): + continue + if path.is_dir() and not path.is_symlink(): + if path.parent == work_root and path.name in {"workspace", "tmp"}: + visit(path) + else: + entries.append(path.relative_to(work_root).as_posix()) + else: + entries.append(path.relative_to(work_root).as_posix()) + + visit(work_root) + receipt = native_receipts(evidence) if evidence is not None else None + document = { + "schema": INVENTORY_SCHEMA, + "host_profile_id": HOST_PROFILE_ID, + "scope": "work_root_except_evidence_and_empty_scaffolding", + "retained_evidence_directory": evidence.name + if evidence is not None and evidence.parent == work_root + else None, + "result_sha256": receipt["results"] if receipt is not None else {}, + "entries": entries, + "empty": entries == [], + } + validate_schema(document, "host-work-root-inventory.json") + return document diff --git a/benchmarks/harness/graphforge_bench/parity_gate.py b/benchmarks/harness/graphforge_bench/parity_gate.py index 10b9f01ef..581348561 100644 --- a/benchmarks/harness/graphforge_bench/parity_gate.py +++ b/benchmarks/harness/graphforge_bench/parity_gate.py @@ -6,6 +6,9 @@ from pathlib import Path from typing import Any +from graphforge_bench.ladder_bundle_ingest import validate_ladder_bundle +from graphforge_bench.native_ladder_bundle import NativeBundleError, validate_native_bundle +from graphforge_bench.progressive_provider_attempt import CANONICAL_RUNGS from graphforge_bench.scale_parity import ( compare_fixture_pair, compare_ladder_bundle, @@ -15,7 +18,7 @@ workspace_root, ) -GATE_SCHEMA = "graphforge-scale-orchestration-parity-gate/1" +GATE_SCHEMA = "graphforge-scale-orchestration-parity-gate-status/1" def _criterion( @@ -63,6 +66,42 @@ def _legacy_orchestration_present(root: Path | None = None) -> bool: return False +def _read_object(path: Path) -> dict[str, Any] | None: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError): + return None + return value if isinstance(value, dict) else None + + +def _canonical_prefix(rung_files: list[Path]) -> tuple[bool, list[int]]: + """Return whether files and embedded scales form an authoritative prefix.""" + if not rung_files or len(rung_files) > len(CANONICAL_RUNGS): + return False, [] + expected = [f"s{scale}-rung.json" for scale in CANONICAL_RUNGS[: len(rung_files)]] + if [path.name for path in rung_files] != expected: + return False, [] + scales: list[int] = [] + for path in rung_files: + document = _read_object(path) + scale = document.get("scale") if document is not None else None + if type(scale) is not int: + return False, scales + scales.append(scale) + return scales == list(CANONICAL_RUNGS[: len(scales)]), scales + + +def _full_ladder_bundle_complete(bundle: Path, *, canonical_prefix: bool) -> tuple[bool, str]: + """Recognize completed native engineering evidence without a cloud ceremony.""" + if not canonical_prefix: + return False, "rung files are not a canonical prefix" + try: + receipt = validate_native_bundle(bundle) + except (NativeBundleError, OSError) as error: + return False, str(error) + return receipt["complete"], f"scales={receipt['scales']}; empty_work_root={receipt['empty']}" + + def parity_gate_status(root: Path | None = None) -> dict[str, Any]: """Report #959 acceptance-criteria readiness from checked-in fixtures only.""" base = root or workspace_root() @@ -87,47 +126,56 @@ def parity_gate_status(root: Path | None = None) -> dict[str, Any]: ladder_ok = False ladder_detail = "no ingested #900 rung bundles" + canonical_prefix, rung_scales = _canonical_prefix(rung_files) if rung_files: - matrices = compare_ladder_bundle(bundle) - ladder_ok = bool(matrices) and all( - matrix["overall"] in {"match", "accepted_difference"} for matrix in matrices - ) - ladder_detail = f"{len(rung_files)} rung file(s), {len(matrices)} comparison(s)" + try: + validate_ladder_bundle(bundle) + matrices = compare_ladder_bundle(bundle) + ladder_ok = bool(matrices) and all( + matrix["overall"] in {"match", "accepted_difference"} for matrix in matrices + ) + ladder_detail = ( + f"{len(rung_files)} rung file(s), {len(matrices)} comparison(s); " + f"canonical_prefix={canonical_prefix}; scales={rung_scales}" + ) + except Exception as error: + ladder_detail = f"{len(rung_files)} rung file(s); comparison failed: {error}" accepted = load_accepted_differences() migration_fixtures = [ path.name for path in (fixtures / "legacy").glob("*.json") if path.name != "tiny-pass.json" ] - harness_authoritative_met = ladder_ok if rung_files else False - parity_matrix_met = tiny_ok and harness_authoritative_met + prefix_comparison_met = bool(rung_files) and ladder_ok and canonical_prefix + prefix_parity_ready = tiny_ok and prefix_comparison_met legacy_retired = not _legacy_orchestration_present(base) + structural_retirement_ready = legacy_retired and historical_ok and bool(migration_fixtures) + full_ladder_complete, full_ladder_detail = _full_ladder_bundle_complete( + bundle, canonical_prefix=canonical_prefix + ) + full_ladder_evidence_complete = prefix_parity_ready and full_ladder_complete + + harness_authoritative_met = full_ladder_evidence_complete criteria = [ _criterion( "parity_matrix_no_unexplained_gaps", - met=parity_matrix_met, + met=prefix_parity_ready, blocked_by="#900 ladder bundles" if tiny_ok and not rung_files else None, evidence=f"tiny overall={tiny_matrix['overall']}; ladder {ladder_detail}", ), _criterion( "harness_authoritative_after_ladder_comparison", met=harness_authoritative_met, - blocked_by="#900" - if not rung_files - else ("ladder parity gaps" if not ladder_ok else None), + blocked_by="#900" if not full_ladder_evidence_complete else None, evidence=ladder_detail, ), _criterion( "legacy_orchestration_retired_with_coverage", - met=legacy_retired and parity_matrix_met and harness_authoritative_met, - blocked_by=( - "parity_matrix_no_unexplained_gaps + harness_authoritative" - if not parity_matrix_met - else ( - "legacy Makefile targets and workflows remain" if not legacy_retired else None - ) - ), + met=legacy_retired, + blocked_by="legacy Makefile targets and workflows remain" + if not legacy_retired + else None, evidence=( f"coverage entries={len(coverage_map())}; legacy_present={not legacy_retired}" ), @@ -147,12 +195,21 @@ def parity_gate_status(root: Path | None = None) -> dict[str, Any]: met=True, evidence="compare_ladder_bundle ingests #900 output read-only only", ), + _criterion( + "full_ladder_evidence_complete", + met=full_ladder_evidence_complete, + blocked_by="complete native S18-S26 evidence and work-root inventory" + if not full_ladder_evidence_complete + else None, + evidence=full_ladder_detail, + ), ] - ready_for_retirement = all(row["met"] for row in criteria) return { "schema": GATE_SCHEMA, - "ready_for_retirement": ready_for_retirement, + "structural_retirement_ready": structural_retirement_ready, + "prefix_parity_ready": prefix_parity_ready, + "full_ladder_evidence_complete": full_ladder_evidence_complete, "accepted_differences_schema": accepted.get("schema"), "criteria": criteria, } diff --git a/benchmarks/harness/graphforge_bench/parity_gate_cli.py b/benchmarks/harness/graphforge_bench/parity_gate_cli.py index b605a6247..2938aff4e 100644 --- a/benchmarks/harness/graphforge_bench/parity_gate_cli.py +++ b/benchmarks/harness/graphforge_bench/parity_gate_cli.py @@ -3,24 +3,15 @@ from __future__ import annotations import json -import sys -from graphforge_bench.parity_gate import assert_tiny_parity_ready, parity_gate_status +from graphforge_bench.parity_gate import parity_gate_status def main() -> int: status = parity_gate_status() print(json.dumps(status, indent=2, sort_keys=True)) - try: - assert_tiny_parity_ready() - except ValueError as error: - print(str(error), file=sys.stderr) + if not (status["structural_retirement_ready"] and status["prefix_parity_ready"]): return 1 - if not status["ready_for_retirement"]: - print( - "parity gate: tiny shadow OK; retirement blocked on #900 ladder bundles", - file=sys.stderr, - ) return 0 diff --git a/benchmarks/harness/graphforge_bench/progressive_host_run.py b/benchmarks/harness/graphforge_bench/progressive_host_run.py index 16ee463f4..a79d1acf5 100644 --- a/benchmarks/harness/graphforge_bench/progressive_host_run.py +++ b/benchmarks/harness/graphforge_bench/progressive_host_run.py @@ -453,20 +453,11 @@ def reclaim_rung_workspace(work_root: Path, scale: int) -> None: shutil.rmtree(target) -def inventory_work_root(work_root: Path) -> dict[str, Any]: - """Return a sanitized inventory proving temporary debris state.""" - workspace = work_root / "workspace" - remaining: list[str] = [] - if workspace.is_dir(): - for path in sorted(workspace.rglob("*")): - if path.is_file() or path.is_dir(): - remaining.append(path.relative_to(work_root).as_posix()) - return { - "schema": "graphforge-host-work-root-inventory/1", - "host_profile_id": HOST_PROFILE_ID, - "workspace_entries": remaining, - "empty": remaining == [], - } +def inventory_work_root(work_root: Path, output_dir: Path | None = None) -> dict[str, Any]: + """Inventory real work-root debris, binding evidence to native results when supplied.""" + from graphforge_bench.native_ladder_bundle import collect_inventory + + return collect_inventory(work_root, output_dir) def run( @@ -700,7 +691,7 @@ def main(argv: Sequence[str] | None = None) -> int: try: work_root = require_work_root(args.work_root) if args.inventory: - document = inventory_work_root(work_root) + document = inventory_work_root(work_root, args.output_dir) publish_json_no_clobber(args.output_dir / "work-root-inventory.json", document) print(json.dumps(document, sort_keys=True)) return 0 if document["empty"] else 2 diff --git a/benchmarks/schemas/host-work-root-inventory.json b/benchmarks/schemas/host-work-root-inventory.json new file mode 100644 index 000000000..687624073 --- /dev/null +++ b/benchmarks/schemas/host-work-root-inventory.json @@ -0,0 +1,55 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "schema": "graphforge-host-work-root-inventory-schema/2", + "type": "object", + "additionalProperties": false, + "properties": { + "schema": { + "const": "graphforge-host-work-root-inventory/2" + }, + "host_profile_id": { + "const": "local-linux-cgroups-v2" + }, + "scope": { + "const": "work_root_except_evidence_and_empty_scaffolding" + }, + "retained_evidence_directory": { + "type": [ + "string", + "null" + ], + "pattern": "^[^/\\\\]+$" + }, + "result_sha256": { + "type": "object", + "patternProperties": { + "^s(18|19|20|22|24|25|26)-result\\.json$": { + "type": "string", + "pattern": "^[0-9a-f]{64}$" + } + }, + "additionalProperties": false + }, + "entries": { + "type": "array", + "uniqueItems": true, + "items": { + "type": "string", + "minLength": 1, + "pattern": "^(?!/)(?!.*(?:^|/)\\.\\.(?:/|$)).+$" + } + }, + "empty": { + "type": "boolean" + } + }, + "required": [ + "schema", + "host_profile_id", + "scope", + "retained_evidence_directory", + "result_sha256", + "entries", + "empty" + ] +} diff --git a/benchmarks/schemas/scale-orchestration-parity-gate-status.json b/benchmarks/schemas/scale-orchestration-parity-gate-status.json new file mode 100644 index 000000000..8010884ac --- /dev/null +++ b/benchmarks/schemas/scale-orchestration-parity-gate-status.json @@ -0,0 +1,53 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "schema": "graphforge-scale-orchestration-parity-gate-status-schema/1", + "type": "object", + "properties": { + "schema": { + "const": "graphforge-scale-orchestration-parity-gate-status/1" + }, + "structural_retirement_ready": { "type": "boolean" }, + "prefix_parity_ready": { "type": "boolean" }, + "full_ladder_evidence_complete": { "type": "boolean" }, + "accepted_differences_schema": { + "const": "graphforge-scale-orchestration-accepted-differences/1" + }, + "criteria": { + "type": "array", + "minItems": 7, + "maxItems": 7, + "items": { "$ref": "#/$defs/criterion" } + } + }, + "required": [ + "schema", + "structural_retirement_ready", + "prefix_parity_ready", + "full_ladder_evidence_complete", + "accepted_differences_schema", + "criteria" + ], + "additionalProperties": false, + "$defs": { + "criterion": { + "type": "object", + "properties": { + "name": { + "type": "string", + "pattern": "^[a-z][a-z0-9_]{0,80}$" + }, + "met": { "type": "boolean" }, + "blocked_by": { + "type": ["string", "null"], + "maxLength": 120 + }, + "evidence": { + "type": ["string", "null"], + "maxLength": 1024 + } + }, + "required": ["name", "met", "blocked_by", "evidence"], + "additionalProperties": false + } + } +} diff --git a/benchmarks/tests/test_ladder_bundle_ingest.py b/benchmarks/tests/test_ladder_bundle_ingest.py index 63194918e..b6e7acde1 100644 --- a/benchmarks/tests/test_ladder_bundle_ingest.py +++ b/benchmarks/tests/test_ladder_bundle_ingest.py @@ -73,6 +73,17 @@ def test_validate_accepts_complete_bundle(self) -> None: self.assertEqual(report["manifest_commit"], COMMIT) self.assertEqual(report["rung_files"], ["s18-rung.json"]) + def test_historical_controller_result_names_do_not_select_native_schema(self) -> None: + for schema in ( + "graphforge-progressive-run-result/1", + "graphforge-progressive-provider-run-result/1", + ): + with self.subTest(schema=schema): + (self.source / "s18-result.json").write_text(json.dumps({"schema": schema})) + self.assertEqual( + validate_ladder_bundle(self.source)["rung_files"], ["s18-rung.json"] + ) + def test_validate_refuses_missing_manifest(self) -> None: (self.source / "manifest.json").unlink() with self.assertRaises(LadderBundleIngestError): @@ -111,8 +122,11 @@ def test_harness_authority_met_after_ingested_bundle(self) -> None: for row in status["criteria"] if row["name"] == "harness_authoritative_after_ladder_comparison" ) - self.assertTrue(harness["met"]) - self.assertIsNone(harness["blocked_by"]) + self.assertFalse(harness["met"]) + self.assertEqual(harness["blocked_by"], "#900") + self.assertTrue(status["prefix_parity_ready"]) + self.assertFalse(status["full_ladder_evidence_complete"]) + self.assertNotIn("ready_for_retirement", status) if __name__ == "__main__": diff --git a/benchmarks/tests/test_native_ladder_bundle.py b/benchmarks/tests/test_native_ladder_bundle.py new file mode 100644 index 000000000..589817c83 --- /dev/null +++ b/benchmarks/tests/test_native_ladder_bundle.py @@ -0,0 +1,269 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +import shutil +import tempfile +import unittest + +from graphforge_bench.ladder_bundle_ingest import ingest_ladder_bundle +from graphforge_bench.native_ladder_bundle import ( + NativeBundleError, + digest, + native_receipts, + validate_native_bundle, +) +from graphforge_bench.progressive_host_run import _result, inventory_work_root +from graphforge_bench.progressive_provider_attempt import CANONICAL_RUNGS +from graphforge_bench.progressive_qualification import load_profiles, project +from graphforge_bench.progressive_run import assemble_rung_evidence +from tests.test_progressive_host_run import host_capacity, host_result +from tests.test_progressive_run import authoritative_receipts, benchexec, graphforge + +ROOT = Path(__file__).resolve().parents[1] + + +def write(path: Path, value: dict) -> None: + path.write_text(json.dumps(value) + "\n", encoding="utf-8") + + +def write_native_bundle(work: Path, scales: tuple[int, ...] = CANONICAL_RUNGS) -> Path: + """Use native result/inventory producers around bounded synthetic phase evidence.""" + work.mkdir(parents=True, exist_ok=True) + source = work / "evidence" + source.mkdir() + profiles = {item.scale: item for item in load_profiles()} + for i, scale in enumerate(scales): + identities = host_result(scale)["identities"] + if scale >= 20: + capacity = host_capacity() + rates = {key: capacity[key] for key in capacity if key.endswith("_per_second")} + preceding = [json.loads((source / f"s{s}-rung.json").read_bytes()) for s in scales[:i]] + projection = project(profiles[scale], preceding, rates) + path = source / f"s{scale}-projection.json" + write(path, projection) + identities["admitted_projection_sha256"] = digest(path) + plan = { + "schema": "graphforge-progressive-host-run-plan/1", + "rung": f"S{scale}", + "execution": "native_linux_benchexec_host", + "identities": identities, + "limits": {"wall_seconds": 14400, "memory_bytes": 4294967296, "cores": 16}, + "outputs": [ + f"s{scale}-{kind}.json" + for kind in ("plan", "benchexec", "graphforge", "rung", "result") + ], + "claim": "engineering_evidence_only", + } + gf = graphforge(scale, authoritative_receipts(scale)) + gf["profile_id"] = identities["profile_id"] + bench = benchexec(gf) + rung = assemble_rung_evidence( + root=ROOT, + scale=scale, + graphforge=gf, + benchexec=bench, + profile_id=identities["profile_id"], + source="progressive_profile" if scale < 20 else "canonical_ladder", + ) + artifacts = {} + for kind, value in ( + ("plan", plan), + ("benchexec", bench), + ("graphforge", gf), + ("rung", rung), + ): + path = source / f"s{scale}-{kind}.json" + write(path, value) + artifacts[f"{kind}_sha256"] = digest(path) + write(source / f"s{scale}-result.json", _result(plan, "passed", None, artifacts)) + write(source / "work-root-inventory.json", inventory_work_root(work, source)) + return source + + +class NativeBundleTests(unittest.TestCase): + def setUp(self) -> None: + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.work = Path(self.temp.name) / "work" + + def test_real_producer_inventory_round_trip_and_ingestion(self) -> None: + source = write_native_bundle(self.work) + receipt = validate_native_bundle(source) + self.assertTrue(receipt["complete"]) + self.assertEqual(receipt["scales"], list(CANONICAL_RUNGS)) + target = Path(self.temp.name) / "ingested" + report = ingest_ladder_bundle(source, target) + self.assertEqual(report["parity_comparisons"], 7) + self.assertTrue(validate_native_bundle(target)["complete"]) + self.assertFalse((target / "manifest.json").exists()) + self.assertFalse((target / "teardown-inventory.json").exists()) + + def test_prefix_is_valid_evidence_without_full_completion(self) -> None: + source = write_native_bundle(self.work, (18, 19)) + self.assertFalse(validate_native_bundle(source)["complete"]) + + def test_inventory_does_not_hide_tmp_failure_trees_links_or_special_files(self) -> None: + source = write_native_bundle(self.work, (18,)) + (self.work / "workspace").mkdir() + (self.work / "tmp").mkdir() + self.assertTrue(inventory_work_root(self.work, source)["empty"]) + for name in ("tmp/payload", ".gf-host-authority-failed/payload", "unexpected"): + path = self.work / name + path.parent.mkdir(exist_ok=True) + path.write_bytes(b"payload") + (self.work / "dangling").symlink_to("absent") + outside = Path(self.temp.name) / "outside" + outside.mkdir() + (outside / "untouched").write_bytes(b"outside") + (self.work / "linked").symlink_to(outside, target_is_directory=True) + if os.name == "posix": + os.mkfifo(self.work / "fifo") + inventory = inventory_work_root(self.work, source) + self.assertFalse(inventory["empty"]) + for name in ( + "tmp/payload", + ".gf-host-authority-failed", + "unexpected", + "dangling", + "linked", + ): + self.assertIn(name, inventory["entries"]) + self.assertNotIn("linked/untouched", inventory["entries"]) + if os.name == "posix": + self.assertIn("fifo", inventory["entries"]) + write(source / "work-root-inventory.json", inventory) + self.assertFalse(validate_native_bundle(source)["complete"]) + + def test_inventory_cannot_exempt_entire_work_root(self) -> None: + self.work.mkdir() + with self.assertRaises(NativeBundleError): + inventory_work_root(self.work, self.work) + + def test_identity_and_digest_mutations_fail_closed(self) -> None: + source = write_native_bundle(self.work, (18, 19)) + for name, mutate in { + "s18-rung.json": lambda d: d.update(live_edges=10), + "s18-result.json": lambda d: d["identities"].update(commit="b" * 40), + "s19-result.json": lambda d: d.update( + status="failed", failure="benchexec_failed", artifacts=None + ), + "work-root-inventory.json": lambda d: d["result_sha256"].update( + {"s18-result.json": "0" * 64} + ), + }.items(): + with self.subTest(name=name): + path = source / name + original = path.read_bytes() + value = json.loads(original) + mutate(value) + write(path, value) + with self.assertRaises(NativeBundleError): + validate_native_bundle(source) + path.write_bytes(original) + + def test_missing_noncanonical_and_malformed_receipts_fail_closed(self) -> None: + source = write_native_bundle(self.work, (18, 19)) + extra = source / "s018-rung.json" + shutil.copy2(source / "s18-rung.json", extra) + with self.assertRaises(NativeBundleError): + native_receipts(source) + extra.unlink() + (source / "s19-result.json").unlink() + with self.assertRaises(NativeBundleError): + native_receipts(source) + (source / "s19-result.json").write_text("[]") + with self.assertRaises(NativeBundleError): + native_receipts(source) + + def test_old_unbound_inventory_cannot_claim_complete_native_evidence(self) -> None: + source = write_native_bundle(self.work) + write( + source / "work-root-inventory.json", + { + "schema": "graphforge-host-work-root-inventory/1", + "host_profile_id": "local-linux-cgroups-v2", + "workspace_entries": [], + "empty": True, + }, + ) + with self.assertRaises(NativeBundleError): + validate_native_bundle(source) + + def test_rehashed_but_contradictory_receipts_still_fail(self) -> None: + source = write_native_bundle(self.work) + mutations = { + "rung": lambda d: d["storage_attribution"]["counts"].update(imported_edges=7), + "plan": lambda d: d["identities"].update(commit="b" * 40), + "graphforge": lambda d: d["phases"][0].update(status="failed"), + "projection": lambda d: d.update(decision="refused"), + } + for kind, mutate in mutations.items(): + with self.subTest(kind=kind): + path = source / f"s26-{kind}.json" + result_path = source / "s26-result.json" + inventory_path = source / "work-root-inventory.json" + backups = {p: p.read_bytes() for p in (path, result_path, inventory_path)} + value = json.loads(path.read_bytes()) + mutate(value) + write(path, value) + result = json.loads(result_path.read_bytes()) + if kind == "projection": + result["identities"]["admitted_projection_sha256"] = digest(path) + else: + result["artifacts"][f"{kind}_sha256"] = digest(path) + write(result_path, result) + inventory = json.loads(inventory_path.read_bytes()) + inventory["result_sha256"][result_path.name] = digest(result_path) + write(inventory_path, inventory) + with self.assertRaises(NativeBundleError): + validate_native_bundle(source) + for p, original in backups.items(): + p.write_bytes(original) + + def test_missing_and_mutated_ordinary_receipts_fail_after_rehashing(self) -> None: + source = write_native_bundle(self.work, (18,)) + paths = [source / f"s18-{kind}.json" for kind in ("graphforge", "benchexec", "result")] + paths.append(source / "work-root-inventory.json") + backups = {path: path.read_bytes() for path in paths} + cases = [ + "all_missing", + "ingest", + "recount", + "query", + "reopen", + "reopen_proof", + "imported_fingerprint", + ] + for case in cases: + with self.subTest(case=case): + for path, original in backups.items(): + path.write_bytes(original) + gf_path, bench_path, result_path, inventory_path = paths + gf = json.loads(gf_path.read_bytes()) + phases = {phase["phase"]: phase for phase in gf["phases"]} + if case == "all_missing": + for phase in phases.values(): + phase.pop("receipts", None) + elif case == "imported_fingerprint": + phases["reopen_proof"]["receipts"][2]["result_sha256"] = "e" * 64 + else: + phases[case].pop("receipts", None) + write(gf_path, gf) + bench = json.loads(bench_path.read_bytes()) + bench["graphforge"] = gf + write(bench_path, bench) + result = json.loads(result_path.read_bytes()) + result["artifacts"]["graphforge_sha256"] = digest(gf_path) + result["artifacts"]["benchexec_sha256"] = digest(bench_path) + write(result_path, result) + inventory = json.loads(inventory_path.read_bytes()) + inventory["result_sha256"][result_path.name] = digest(result_path) + write(inventory_path, inventory) + with self.assertRaisesRegex(NativeBundleError, "ordinary lifecycle receipts"): + validate_native_bundle(source) + + +if __name__ == "__main__": + unittest.main() diff --git a/benchmarks/tests/test_parity_gate.py b/benchmarks/tests/test_parity_gate.py index d9d85c7ac..e05ca3d41 100644 --- a/benchmarks/tests/test_parity_gate.py +++ b/benchmarks/tests/test_parity_gate.py @@ -1,14 +1,46 @@ from __future__ import annotations +from contextlib import redirect_stdout +import io import json +from pathlib import Path +import shutil +import tempfile import unittest +from unittest.mock import patch from graphforge_bench.parity_gate import ( assert_tiny_parity_ready, ladder_bundle_root, parity_gate_status, ) -from graphforge_bench.scale_parity import compare_ladder_bundle +from graphforge_bench.parity_gate_cli import main as parity_gate_main +from graphforge_bench.scale_parity import compare_ladder_bundle, workspace_root +from jsonschema import Draft202012Validator + +EMPTY_OBSERVED = { + "app_exists": False, + "machines": 0, + "volumes": 0, + "secrets": 0, +} + + +def _write_complete_ladder(base: Path) -> Path: + from tests.test_native_ladder_bundle import write_native_bundle + + source = write_native_bundle(base / "native-work") + bundle = base / "fixtures" / "parity" / "ladder-bundle" + shutil.rmtree(bundle) + shutil.copytree(source, bundle) + return bundle + + +def _temporary_fixture_root(temp_name: str) -> Path: + base = Path(temp_name) / "benchmarks" + shutil.copytree(workspace_root() / "fixtures" / "parity", base / "fixtures" / "parity") + (base.parent / "Makefile").write_text("safe-target:\n\t@true\n", encoding="utf-8") + return base class ParityGateTests(unittest.TestCase): @@ -17,14 +49,17 @@ def test_tiny_parity_ready(self) -> None: def test_gate_status_reports_retirement_ready(self) -> None: status = parity_gate_status() - self.assertTrue(status["ready_for_retirement"]) + self.assertNotIn("ready_for_retirement", status) + self.assertTrue(status["structural_retirement_ready"]) + self.assertTrue(status["prefix_parity_ready"]) + self.assertFalse(status["full_ladder_evidence_complete"]) harness = next( row for row in status["criteria"] if row["name"] == "harness_authoritative_after_ladder_comparison" ) - self.assertTrue(harness["met"]) - self.assertIsNone(harness["blocked_by"]) + self.assertFalse(harness["met"]) + self.assertEqual(harness["blocked_by"], "#900") legacy = next( row for row in status["criteria"] @@ -49,6 +84,107 @@ def test_gate_status_json_serializable(self) -> None: payload = parity_gate_status() json.dumps(payload) + def test_gate_status_matches_closed_schema(self) -> None: + schema = json.loads( + ( + workspace_root() / "schemas" / "scale-orchestration-parity-gate-status.json" + ).read_text(encoding="utf-8") + ) + Draft202012Validator.check_schema(schema) + Draft202012Validator(schema).validate(parity_gate_status()) + + def test_exact_complete_ladder_reports_all_states_true(self) -> None: + with tempfile.TemporaryDirectory() as temp_name: + base = _temporary_fixture_root(temp_name) + _write_complete_ladder(base) + status = parity_gate_status(base) + self.assertTrue(status["structural_retirement_ready"]) + self.assertTrue(status["prefix_parity_ready"]) + self.assertTrue(status["full_ladder_evidence_complete"]) + + def test_certification_defects_do_not_conflate_independent_states(self) -> None: + mutations = { + "missing_terminal_rung": lambda bundle: (bundle / "s26-rung.json").unlink(), + "result_identity_mismatch": lambda bundle: self._update_json( + bundle / "s26-result.json", rung="S25" + ), + "incomplete_teardown": lambda bundle: self._update_json( + bundle / "work-root-inventory.json", + empty=False, + entries=["tmp/payload"], + ), + } + for name, mutate in mutations.items(): + with self.subTest(name=name), tempfile.TemporaryDirectory() as temp_name: + base = _temporary_fixture_root(temp_name) + bundle = _write_complete_ladder(base) + mutate(bundle) + status = parity_gate_status(base) + self.assertTrue(status["structural_retirement_ready"]) + self.assertEqual(status["prefix_parity_ready"], name == "incomplete_teardown") + self.assertFalse(status["full_ladder_evidence_complete"]) + + def test_noncanonical_duplicate_and_out_of_order_rungs_fail_prefix(self) -> None: + def noncanonical(bundle: Path) -> None: + source = json.loads((bundle / "s20-rung.json").read_text(encoding="utf-8")) + source["scale"] = 21 + (bundle / "s21-rung.json").write_text(json.dumps(source) + "\n", encoding="utf-8") + + def duplicate(bundle: Path) -> None: + shutil.copy2(bundle / "s18-rung.json", bundle / "s018-rung.json") + + def out_of_order(bundle: Path) -> None: + s18 = json.loads((bundle / "s18-rung.json").read_text(encoding="utf-8")) + s19 = json.loads((bundle / "s19-rung.json").read_text(encoding="utf-8")) + s18["scale"], s19["scale"] = s19["scale"], s18["scale"] + (bundle / "s18-rung.json").write_text(json.dumps(s18) + "\n", encoding="utf-8") + (bundle / "s19-rung.json").write_text(json.dumps(s19) + "\n", encoding="utf-8") + + for name, mutate in { + "noncanonical": noncanonical, + "duplicate": duplicate, + "out_of_order": out_of_order, + }.items(): + with self.subTest(name=name), tempfile.TemporaryDirectory() as temp_name: + base = _temporary_fixture_root(temp_name) + bundle = _write_complete_ladder(base) + mutate(bundle) + status = parity_gate_status(base) + self.assertTrue(status["structural_retirement_ready"]) + self.assertFalse(status["prefix_parity_ready"]) + self.assertFalse(status["full_ladder_evidence_complete"]) + + @staticmethod + def _update_json(path: Path, **updates: object) -> None: + document = json.loads(path.read_text(encoding="utf-8")) + document.update(updates) + path.write_text(json.dumps(document) + "\n", encoding="utf-8") + + def test_cli_ignores_incomplete_full_ladder_evidence(self) -> None: + status = parity_gate_status() + self.assertFalse(status["full_ladder_evidence_complete"]) + with ( + patch("graphforge_bench.parity_gate_cli.parity_gate_status", return_value=status), + redirect_stdout(io.StringIO()) as output, + ): + self.assertEqual(parity_gate_main(), 0) + self.assertNotIn("ready_for_retirement", output.getvalue()) + + def test_cli_fails_only_for_structural_or_prefix_invariant(self) -> None: + baseline = parity_gate_status() + for field in ("structural_retirement_ready", "prefix_parity_ready"): + with self.subTest(field=field): + status = dict(baseline) + status[field] = False + with ( + patch( + "graphforge_bench.parity_gate_cli.parity_gate_status", + return_value=status, + ), + redirect_stdout(io.StringIO()), + ): + self.assertEqual(parity_gate_main(), 1) + if __name__ == "__main__": unittest.main()