fix(scale): reconcile measured lifecycle accounting and admission

benchmarks/harness/graphforge_bench/parity_gate.py

@@ -9,12 +9,13 @@
 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.progressive_qualification import PHASES
 from graphforge_bench.scale_parity import (
     compare_fixture_pair,
     compare_ladder_bundle,
     coverage_map,
     load_accepted_differences,
-    validate_historical_legacy_cert,
+    read_historical_legacy_cert,
     workspace_root,
 )
 
@@ -119,8 +120,11 @@ def parity_gate_status(root: Path | None = None) -> dict[str, Any]:
     historical_ok = False
     if historical_path.is_file():
         try:
-            validate_historical_legacy_cert(historical_path, expected_sha="a" * 40)
-            historical_ok = True
+            historical = read_historical_legacy_cert(historical_path, expected_sha="a" * 40)
+            historical_ok = (
+                historical.status == "passed"
+                and tuple(phase.phase for phase in historical.phases) == PHASES
+            )
         except Exception:
             historical_ok = False
 

benchmarks/harness/graphforge_bench/scale_parity.py

@@ -12,6 +12,7 @@
 from enum import StrEnum
 import json
 from pathlib import Path
+import re
 from typing import Any
 
 from jsonschema import Draft202012Validator
@@ -301,8 +302,58 @@ def coverage_map() -> dict[str, str]:
     }
 
 
-def validate_historical_legacy_cert(evidence_path: Path, *, expected_sha: str) -> None:
-    """Ensure preserved legacy certification evidence still passes the historical validator."""
+def read_historical_legacy_cert(evidence_path: Path, *, expected_sha: str) -> NormalizedEvidence:
+    """Read a preserved lifecycle fixture; this does not authenticate certification."""
+    try:
+        document = json.loads(evidence_path.read_text(encoding="utf-8"))
+    except (OSError, ValueError) as error:
+        raise ParityError("historical lifecycle fixture is unreadable") from error
+    if not isinstance(document, dict) or document.get("schema") != LEGACY_CERT_SCHEMA:
+        raise ParityError("historical lifecycle fixture requires the legacy schema")
+    if (
+        re.fullmatch(r"[0-9a-f]{40}", expected_sha) is None
+        or document.get("git_sha") != expected_sha
+    ):
+        raise ParityError("historical lifecycle fixture git_sha mismatch")
+    run = document.get("run")
+    if not isinstance(run, dict) or type(run.get("scale")) is not int or run["scale"] <= 0:
+        raise ParityError("historical lifecycle fixture requires a positive scale")
+    if document.get("result") not in ("pass", "fail"):
+        raise ParityError("historical lifecycle fixture requires pass/fail result")
+    phases = document.get("phases")
+    if not isinstance(phases, list) or any(not isinstance(phase, dict) for phase in phases):
+        raise ParityError("historical lifecycle fixture requires phase objects")
+    expected_ids = set(_legacy_cert_phase_mapping(load_accepted_differences()))
+    ids = [phase.get("id") for phase in phases]
+    if (
+        any(not isinstance(phase_id, str) for phase_id in ids)
+        or len(ids) != len(expected_ids)
+        or set(ids) != expected_ids
+    ):
+        raise ParityError("historical lifecycle fixture requires each legacy phase exactly once")
+    for phase in phases:
+        if phase.get("status") not in ("pass", "fail"):
+            raise ParityError("historical lifecycle fixture requires pass/fail phase status")
+        for metric in ("elapsed_ms", "rss_peak_bytes"):
+            if type(phase.get(metric)) is not int or phase[metric] < 0:
+                raise ParityError(f"historical lifecycle fixture has invalid {metric}")
+    if document["result"] == "pass" and any(phase["status"] != "pass" for phase in phases):
+        raise ParityError("historical lifecycle result contradicts a failed phase")
+    return normalize_legacy_cert_lifecycle(document)
+
+
+def validate_historical_legacy_cert(
+    evidence_path: Path,
+    *,
+    expected_sha: str,
+    provider_result_path: Path | None = None,
+    provider_result_sha256: str | None = None,
+) -> None:
+    """Validate legacy evidence only when an external provider result authenticates it."""
+    if provider_result_path is None or provider_result_sha256 is None:
+        raise ParityError(
+            "historical certification evidence requires an external provider-result anchor"
+        )
     import importlib.util
 
     script = workspace_root().parent / "scripts" / "ci" / "validate-g500-certification.py"
@@ -311,10 +362,17 @@ def validate_historical_legacy_cert(evidence_path: Path, *, expected_sha: str) -
         raise ParityError("unable to load validate-g500-certification.py")
     module = importlib.util.module_from_spec(spec)
     spec.loader.exec_module(module)
-    document = json.loads(evidence_path.read_text(encoding="utf-8"))
-    if not isinstance(document, dict):
-        raise ParityError("legacy certification fixture must be a JSON object")
-    module.validate(document, expected_sha)
+    try:
+        module.validate(
+            evidence_path.read_bytes(),
+            expected_sha,
+            provider_result_path.read_bytes(),
+            provider_result_sha256,
+        )
+    except (OSError, ValueError) as error:
+        raise ParityError(
+            "legacy certification evidence is not externally authenticated"
+        ) from error
 
 
 def compare_ladder_bundle(bundle_root: Path) -> list[dict[str, Any]]:

benchmarks/scale-parity-index.md

@@ -2,7 +2,7 @@
 
 Maps retired legacy Graph500 scale orchestration to the isolated `benchmarks/` harness.
 Retirement is complete in-tree; the parity gate remains blocked on ingested #900 ladder
-bundles until Fly execution evidence is checked in (see `fixtures/parity/ladder-bundle/README.md`).
+bundles until native host execution evidence is ingested (see `fixtures/parity/ladder-bundle/README.md`).
 
 ## Coverage map
 
@@ -12,8 +12,8 @@ bundles until Fly execution evidence is checked in (see `fixtures/parity/ladder-
 | `make bench-g500-scale20` | `profiles/graph500/s20-*.json` + progressive qualification | tiny shadow OK; ladder bundle pending #900 |
 | `make g500-ladder-qualification` | progressive qualification schemas + controller | retired; harness authoritative |
 | `cargo test -p graphforge-api --test scale_g500_ladder` (S10 CI) | `benchmarks/scripts/test-tiny-lifecycle-certification.py` | bounded correctness retained in product CI |
-| `cargo test … certification_target_live…` (retired) | `qualification-operator GATE=progressive-ladder` | blocked on #900 Fly execution |
-| `scripts/ci/validate-g500-certification.py` | `graphforge_bench.scale_parity` + progressive schemas | historical fixture validates |
+| `cargo test … certification_target_live…` (retired) | `qualification-operator GATE=progressive-ladder` | blocked on #900 native host execution |
+| `scripts/ci/validate-g500-certification.py` | `graphforge_bench.scale_parity` + progressive schemas | historical lifecycle fixture remains readable |
 | `docs/development/perf-g500-ladder.md` | `benchmarks/README.md` | historical reference retained |
 | `.github/workflows/g500-certification.yml` | `.github/workflows/progressive-ladder.yml` | retired; progressive-ladder handoff wired |
 
@@ -38,13 +38,19 @@ PYTHONPATH=harness uv run --locked python -m unittest tests.test_scale_parity te
 # Historical legacy certification fixture readability
 PYTHONPATH=harness uv run --locked python -c "
 from pathlib import Path
-from graphforge_bench.scale_parity import validate_historical_legacy_cert, workspace_root
+from graphforge_bench.scale_parity import read_historical_legacy_cert, workspace_root
 fixture = workspace_root() / 'fixtures/parity/legacy/cert-s20-minimal.json'
-validate_historical_legacy_cert(fixture, expected_sha='a' * 40)
+historical = read_historical_legacy_cert(fixture, expected_sha='a' * 40)
+assert historical.status == 'passed' and len(historical.phases) == 10
 print('legacy cert fixture readable')
 "
 ```
 
+The historical reader checks the preserved fixture and lifecycle mapping only.
+It does not authenticate a benchmark result or satisfy full-ladder certification.
+`validate_historical_legacy_cert` separately requires an externally authenticated
+provider-result anchor before validating certification evidence.
+
 ## Retirement gate (#959 acceptance)
 
 Legacy orchestration is retired in-tree. Full gate green requires:

benchmarks/tests/test_parity_gate.py

@@ -75,6 +75,15 @@ def test_historical_evidence_criterion_met(self) -> None:
         )
         self.assertTrue(historical["met"])
 
+    def test_unreadable_history_blocks_structure_without_changing_ladder_parity(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_name:
+            base = _temporary_fixture_root(temp_name)
+            (base / "fixtures/parity/legacy/cert-s20-minimal.json").unlink()
+            status = parity_gate_status(base)
+        self.assertFalse(status["structural_retirement_ready"])
+        self.assertTrue(status["prefix_parity_ready"])
+        self.assertFalse(status["full_ladder_evidence_complete"])
+
     def test_ingested_ladder_bundle_runs_comparisons(self) -> None:
         comparisons = compare_ladder_bundle(ladder_bundle_root())
         self.assertEqual(len(comparisons), 2)

benchmarks/tests/test_scale_parity.py

@@ -1,10 +1,13 @@
 from __future__ import annotations
 
 import json
+from pathlib import Path
+import tempfile
 import unittest
 
 from graphforge_bench.scale_parity import (
     Outcome,
+    ParityError,
     assert_no_unexplained_gaps,
     compare_evidence,
     compare_fixture_pair,
@@ -14,6 +17,7 @@
     normalize_legacy_evidence,
     normalize_new_evidence,
     normalize_rung_evidence,
+    read_historical_legacy_cert,
     validate_historical_legacy_cert,
     workspace_root,
 )
@@ -110,9 +114,52 @@ def test_rung_and_legacy_cert_lifecycle_align(self) -> None:
         self.assertIn(matrix["overall"], {Outcome.MATCH.value, Outcome.ACCEPTED_DIFFERENCE.value})
         assert_no_unexplained_gaps(matrix)
 
-    def test_historical_legacy_cert_fixture_validates(self) -> None:
+    def test_historical_legacy_cert_fixture_fails_closed_without_provider_anchor(self) -> None:
         fixture = FIXTURES / "legacy" / "cert-s20-minimal.json"
-        validate_historical_legacy_cert(fixture, expected_sha="a" * 40)
+        with self.assertRaisesRegex(ParityError, "external provider-result anchor"):
+            validate_historical_legacy_cert(fixture, expected_sha="a" * 40)
+
+    def test_historical_fixture_readability_does_not_authenticate_certification(self) -> None:
+        fixture = FIXTURES / "legacy" / "cert-s20-minimal.json"
+        historical = read_historical_legacy_cert(fixture, expected_sha="a" * 40)
+        self.assertEqual(historical.status, "passed")
+        self.assertEqual(len(historical.phases), 10)
+        with self.assertRaisesRegex(ParityError, "external provider-result anchor"):
+            validate_historical_legacy_cert(fixture, expected_sha="a" * 40)
+
+    def test_historical_reader_rejects_malformed_or_incomplete_lifecycles(self) -> None:
+        original = (FIXTURES / "legacy" / "cert-s20-minimal.json").read_text()
+        mutations = {
+            "wrong_schema": lambda d: d.update(schema="unknown"),
+            "wrong_sha": lambda d: d.update(git_sha="b" * 40),
+            "invalid_result": lambda d: d.update(result=[]),
+            "invalid_scale": lambda d: d["run"].update(scale=True),
+            "missing_phase": lambda d: d["phases"].pop(),
+            "duplicate_phase": lambda d: d["phases"].__setitem__(0, d["phases"][1]),
+            "invalid_phase": lambda d: d["phases"].__setitem__(0, None),
+            "invalid_id": lambda d: d["phases"][0].update(id=[]),
+            "invalid_status": lambda d: d["phases"][0].update(status=[]),
+            "invalid_elapsed": lambda d: d["phases"][0].update(elapsed_ms=True),
+            "negative_rss": lambda d: d["phases"][0].update(rss_peak_bytes=-1),
+            "failed_unmapped_drill": lambda d: d["phases"][-1].update(status="fail"),
+        }
+        with tempfile.TemporaryDirectory() as directory:
+            fixture = Path(directory) / "historical.json"
+            for name, mutate in mutations.items():
+                with self.subTest(name=name):
+                    document = json.loads(original)
+                    mutate(document)
+                    fixture.write_text(json.dumps(document))
+                    with self.assertRaises(ParityError):
+                        read_historical_legacy_cert(fixture, expected_sha="a" * 40)
+            for encoded in ("{", "[]"):
+                with self.subTest(encoded=encoded):
+                    fixture.write_text(encoded)
+                    with self.assertRaises(ParityError):
+                        read_historical_legacy_cert(fixture, expected_sha="a" * 40)
+            fixture.unlink()
+            with self.assertRaises(ParityError):
+                read_historical_legacy_cert(fixture, expected_sha="a" * 40)
 
     def test_coverage_map_lists_legacy_entrypoints(self) -> None:
         mapping = coverage_map()

crates/graphforge-api/src/import_session.rs

@@ -800,8 +800,9 @@ impl GraphImportSession {
         &mut self,
         progress: &crate::GraphConstructionProgress,
     ) -> Result<(), GfError> {
-        let application_io =
-            graphforge_storage::ConstructionPhaseAttribution::from_construction(&progress.evidence);
+        let application_io = graphforge_storage::ConstructionPhaseAttribution::from_construction(
+            &progress.evidence,
+        )?;
         application_io.validate_for_qualification()?;
         let publication_work = PublicationWorkComponents::from_application_io(&application_io)?;
         let construction_staging = progress

crates/graphforge-api/src/lib.rs

@@ -4102,28 +4102,47 @@ fn materialize_compact_graph_target(
         strategy: graphforge_storage::GraphFilesOpenStrategy::PrivateMaterialize,
         files_validated: reused
             .files_validated
-            .saturating_add(copied.files_validated),
+            .checked_add(copied.files_validated)
+            .ok_or_else(|| GfError::Storage("hydration validated-file count overflows".into()))?,
         bytes_validated: reused
             .bytes_validated
-            .saturating_add(copied.bytes_validated),
+            .checked_add(copied.bytes_validated)
+            .ok_or_else(|| GfError::Storage("hydration validated-byte count overflows".into()))?,
         files_copied: copied.files_copied,
         bytes_copied: copied.bytes_copied,
         files_opened_in_place: 0,
         files_reused: reused.files_reused,
         bytes_reused: reused.bytes_reused,
         application_read_bytes: reused
             .application_read_bytes
-            .saturating_add(copied.application_read_bytes),
+            .checked_add(copied.application_read_bytes)
+            .ok_or_else(|| GfError::Storage("hydration read byte count overflows".into()))?,
         application_read_calls: reused
             .application_read_calls
-            .saturating_add(copied.application_read_calls),
+            .checked_add(copied.application_read_calls)
+            .ok_or_else(|| GfError::Storage("hydration read call count overflows".into()))?,
         application_write_bytes: reused
             .application_write_bytes
-            .saturating_add(copied.application_write_bytes),
+            .checked_add(copied.application_write_bytes)
+            .ok_or_else(|| GfError::Storage("hydration write byte count overflows".into()))?,
         application_write_calls: reused
             .application_write_calls
-            .saturating_add(copied.application_write_calls),
-        fsync_calls: reused.fsync_calls.saturating_add(copied.fsync_calls),
+            .checked_add(copied.application_write_calls)
+            .ok_or_else(|| GfError::Storage("hydration write call count overflows".into()))?,
+        fsync_calls: reused
+            .fsync_calls
+            .checked_add(copied.fsync_calls)
+            .ok_or_else(|| GfError::Storage("hydration fsync count overflows".into()))?,
+        file_fsync_calls: reused
+            .file_fsync_calls
+            .checked_add(copied.file_fsync_calls)
+            .ok_or_else(|| GfError::Storage("hydration file barrier count overflows".into()))?,
+        directory_fsync_calls: reused
+            .directory_fsync_calls
+            .checked_add(copied.directory_fsync_calls)
+            .ok_or_else(|| {
+                GfError::Storage("hydration directory barrier count overflows".into())
+            })?,
     };
     Ok(evidence)
 }

crates/graphforge-api/src/resumable_construction.rs

@@ -749,14 +749,18 @@ mod tests {
             assert!(evidence.publication_application_read_bytes > 0);
             assert!(evidence.cas_application_read_bytes > 0);
             assert!(evidence.hydration_application_read_bytes > 0);
-            let reconciled = evidence
-                .seal_application_read_bytes
-                .saturating_add(evidence.shape_application_read_bytes)
-                .saturating_add(evidence.encode_application_read_bytes)
-                .saturating_add(evidence.publication_application_read_bytes)
-                .saturating_add(evidence.cas_application_read_bytes)
-                .saturating_add(evidence.hydration_application_read_bytes);
-            assert_eq!(evidence.total_application_read_bytes(), reconciled);
+            let reconciled = [
+                evidence.seal_application_read_bytes,
+                evidence.shape_application_read_bytes,
+                evidence.encode_application_read_bytes,
+                evidence.publication_application_read_bytes,
+                evidence.cas_application_read_bytes,
+                evidence.hydration_application_read_bytes,
+            ]
+            .into_iter()
+            .try_fold(0_u64, u64::checked_add)
+            .unwrap();
+            assert_eq!(evidence.total_application_read_bytes().unwrap(), reconciled);
             let payload = evidence.write_bytes;
             assert!(
                 payload > 100_000,
@@ -775,10 +779,19 @@ mod tests {
                     "phase exceeded its fixed application bytes-per-staged-payload ceiling"
                 );
             }
-            assert_eq!(
-                evidence.cas_application_read_bytes,
-                evidence.canonical_output_bytes
-            );
+            let cas = &evidence.cas_publication_io;
+            assert_eq!(cas.payload.read_bytes, evidence.canonical_output_bytes);
+            assert!(cas.manifest_reads.read_bytes > 0);
+            assert!(cas.manifest_reads.read_calls > 0);
+            let measured_cas_reads = [
+                cas.payload.read_bytes,
+                cas.manifest.read_bytes,
+                cas.manifest_reads.read_bytes,
+            ]
+            .into_iter()
+            .try_fold(0_u64, u64::checked_add)
+            .unwrap();
+            assert_eq!(evidence.cas_application_read_bytes, measured_cas_reads);
             assert!(evidence.staged_and_retained_disk_bytes >= payload);
             observations.push((
                 payload,