test(scale): prove complete lifecycle ownership growth

.github/workflows/test.yml

@@ -1252,7 +1252,7 @@ jobs:
             //:resource_inputs \
             //:release_bins 2>&1 | tee dist/bazel-representative-build.log
 
-      - name: Real tiny progressive lifecycle producer
+      - name: Real tiny and ownership-growth lifecycle producer
         shell: bash
         run: |
           set -euo pipefail
@@ -1271,6 +1271,13 @@ jobs:
             --certify benchmarks/target/debug/graphforge-benchmark-certify \
             --generator benchmarks/target/debug/graphforge-benchmark-graph500-generator \
             --workspace-root "$RUNNER_TEMP"
+          uv run --project benchmarks --locked python \
+            benchmarks/scripts/test-tiny-lifecycle-certification.py \
+            --growth \
+            --gf "$gf_bin" \
+            --certify benchmarks/target/debug/graphforge-benchmark-certify \
+            --generator benchmarks/target/debug/graphforge-benchmark-graph500-generator \
+            --workspace-root "$RUNNER_TEMP"
 
       - name: Bazel binding cdylibs + packaging handoff
         run: |

benchmarks/scripts/test-tiny-lifecycle-certification.py

@@ -1,10 +1,12 @@
 #!/usr/bin/env python3
-"""Run the real scale-1 lifecycle through the public certification binaries."""
+"""Run tiny and ownership-growth lifecycles through public certification binaries."""
 
 from __future__ import annotations
 
 import argparse
+import hashlib
 import importlib
+import itertools
 import json
 import os
 from pathlib import Path
@@ -35,7 +37,7 @@ def decoded_objects(text: str) -> list[dict[str, object]]:
     return objects
 
 
-def storage_totals(root: Path) -> dict[str, int]:
+def storage_totals(root: Path, selected: set[Path] | None = None) -> dict[str, int]:
     """Independent regular-file oracle, used only after every child has exited.
 
     Directories are not data objects. Hardlinked references count once by the
@@ -52,6 +54,8 @@ def storage_totals(root: Path) -> dict[str, int]:
                 raise SystemExit("allocation oracle encountered a linked directory")
         for name in files:
             path = Path(directory) / name
+            if selected is not None and path not in selected:
+                continue
             before = path.lstat()
             if not stat.S_ISREG(before.st_mode):
                 raise SystemExit("allocation oracle encountered a non-regular file")
@@ -90,14 +94,194 @@ def storage_totals(root: Path) -> dict[str, int]:
     }
 
 
+OWNER_SUFFIXES = ("construction", "import", "transactions", "locks", "admission_lock", "published")
+OWNER_NAMES = {
+    f"{project}-project-{suffix}" for project in ("source", "imported") for suffix in OWNER_SUFFIXES
+} | {"generated-inputs", "query-results", "portable-package"}
+
+
+def owner_inventories(workspace: Path) -> dict[str, set[Path]]:
+    """Classify every quiescent route independently of the producer's owner map."""
+    owners = {name: set() for name in OWNER_NAMES}
+    admissions = {}
+    for project in ("source", "imported"):
+        digest = hashlib.sha256(
+            b"graphforge-project-lifecycle-lock/v1\0"
+            + os.fsencode(workspace)
+            + b"\0"
+            + os.fsencode(project)
+        ).hexdigest()
+        admissions[f".graphforge-admission-{digest}.lock"] = f"{project}-project-admission_lock"
+    for directory, directories, files in os.walk(workspace, followlinks=False):
+        for name in directories:
+            if not stat.S_ISDIR((Path(directory) / name).lstat().st_mode):
+                raise SystemExit("owner oracle encountered linked directory")
+        for name in files:
+            path = Path(directory) / name
+            parts = path.relative_to(workspace).parts
+            if len(parts) == 1 and name in admissions:
+                owner = admissions[name]
+            elif len(parts) == 1 and name in {"nodes.parquet", "edges.parquet"}:
+                owner = "generated-inputs"
+            elif len(parts) == 1 and name in {
+                "node-count.arrow",
+                "edge-count.arrow",
+                "one-hop.arrow",
+                "two-hop.arrow",
+                "imported-node-count.arrow",
+                "imported-edge-count.arrow",
+                "imported-one-hop.arrow",
+                "imported-two-hop.arrow",
+            }:
+                owner = "query-results"
+            elif parts[0] == "portable":
+                owner = "portable-package"
+            elif parts[0] in {"source", "imported"} and len(parts) > 1:
+                suffix = {
+                    ".graphforge-construction": "construction",
+                    "import-sessions": "import",
+                    "transactions": "transactions",
+                    "locks": "locks",
+                }.get(parts[1])
+                if parts[1] in {"FORMAT", "CURRENT", "generations", "graph-objects"}:
+                    suffix = (
+                        "locks"
+                        if len(parts) == 4
+                        and parts[1] == "generations"
+                        and parts[3] == "lease.lock"
+                        else "published"
+                    )
+                if suffix is None:
+                    raise SystemExit("unclassified project owner route")
+                owner = f"{parts[0]}-project-{suffix}"
+            else:
+                raise SystemExit("unclassified workspace owner route")
+            owners[owner].add(path)
+    return owners
+
+
+DATA_OWNERS = {
+    "generated-inputs",
+    "query-results",
+    "portable-package",
+    "source-project-construction",
+    "source-project-import",
+    "source-project-published",
+    "imported-project-published",
+}
+ZERO_OWNERS = {"imported-project-construction", "imported-project-import"}
+BYTE_FIELDS = ("logical_bytes", "physical_logical_bytes", "allocated_bytes")
+
+
+def positive_slopes(name: str, values: list[int], work: list[int]) -> None:
+    deltas = [values[i + 1] - values[i] for i in (0, 1)]
+    if min(deltas) <= 0:
+        raise ValueError(f"{name}: data growth is flat or decreasing")
+    left = deltas[0] * (work[2] - work[1])
+    right = deltas[1] * (work[1] - work[0])
+    if left > 2 * right or right > 2 * left:
+        raise ValueError(f"{name}: adjacent normalized slopes differ by more than factor2")
+
+
+def normalized_ceiling(name: str, values: list[int], work: list[int]) -> None:
+    if values[0] <= 0 or any(b < a for a, b in itertools.pairwise(values)):
+        raise ValueError(f"{name}: positive nondecreasing evidence required")
+    if any(values[i] * work[0] > 2 * values[0] * work[i] for i in (1, 2)):
+        raise ValueError(f"{name}: exceeds existing factor2 normalized ceiling")
+
+
+def validate_growth(observations: list[dict[str, object]]) -> None:
+    """Existing factor2 ceiling, plus strict EOF slopes; no output-fitted constants.
+
+    Allocated owner bytes may plateau due to per-file allocation quanta. They
+    retain the existing quantized normalized ceiling, not an EOF upper bound.
+    Fixed empty locks and absent owners are checked separately. Complete current
+    and peak must actually grow, using the same adjacent slope bound as EOF.
+    """
+    if [o["scale"] for o in observations] != [6, 7, 8]:
+        raise ValueError("growth requires exactly scale6/7/8")
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "harness"))
+    consumer = importlib.import_module("graphforge_bench.progressive_run")
+    work = []
+    for o in observations:
+        try:
+            consumer.validate_lifecycle_storage_receipt(o["receipt"])
+        except consumer.ControllerError as error:
+            raise ValueError(str(error)) from error
+        n, e = o["live_nodes"], o["live_edges"]
+        if n != 1 << o["scale"] or e != 16 * n:
+            raise ValueError("invalid authoritative live denominator")
+        if o["query_rows"] != [n, e] or o["imported_query_rows"] != [n, e]:
+            raise ValueError("growth query output was truncated or duplicated")
+        work.append(n + e)
+        ratios = o["ratios"]
+        for field in ("retained_storage_bytes", "transient_peak_storage_bytes"):
+            if ratios[field] != {
+                "per_live_node": [o["receipt"][field], n],
+                "per_live_edge": [o["receipt"][field], e],
+            }:
+                raise ValueError("fabricated normalized ratio")
+        owners = o["receipt"]["retained_owners"]
+        if set(owners) != OWNER_NAMES:
+            raise ValueError("missing or unknown growth owner")
+    for owner in sorted(OWNER_NAMES):
+        totals = [o["receipt"]["retained_owners"][owner]["totals"] for o in observations]
+        if any(
+            set(t)
+            != {
+                "logical_references",
+                "logical_bytes",
+                "physical_objects",
+                "physical_logical_bytes",
+                "allocated_bytes",
+            }
+            for t in totals
+        ):
+            raise ValueError("missing owner metric")
+        for field in totals[0]:
+            values = [t[field] for t in totals]
+            if any(type(v) is not int or v < 0 for v in values):
+                raise ValueError("invalid owner numerator")
+            if owner in ZERO_OWNERS:
+                if any(values):
+                    raise ValueError(f"{owner}: absent owner became nonzero")
+            elif owner.endswith(("-locks", "-admission_lock")):
+                if field in BYTE_FIELDS and any(values):
+                    raise ValueError(f"{owner}: empty lock gained payload")
+                if len(set(values)) != 1:
+                    raise ValueError(f"{owner}: fixed lock inventory changed")
+            elif owner in DATA_OWNERS:
+                normalized_ceiling(f"{owner}.{field}", values, work)
+                if field in ("logical_bytes", "physical_logical_bytes"):
+                    positive_slopes(f"{owner}.{field}", values, work)
+            # Transaction journals have fixed protocol shape and fixed-width
+            # UUID/digest fields; no graph payload or numeric row inventory.
+            elif len(set(values)) != 1:
+                raise ValueError(f"{owner}: fixed transaction protocol changed")
+    for field in ("retained_storage_bytes", "transient_peak_storage_bytes"):
+        values = [o["receipt"][field] for o in observations]
+        normalized_ceiling(field, values, work)
+        positive_slopes(field, values, work)
+
+
 def main() -> None:
     parser = argparse.ArgumentParser()
     parser.add_argument("--gf", required=True, type=executable)
     parser.add_argument("--certify", required=True, type=executable)
     parser.add_argument("--generator", required=True, type=executable)
     parser.add_argument("--workspace-root", required=True, type=Path)
+    parser.add_argument("--growth", action="store_true")
     args = parser.parse_args()
 
+    if args.growth:
+        observations = [run(args, scale) for scale in (6, 7, 8)]
+        validate_growth(observations)
+        print(json.dumps({"lifecycle_growth": observations}, sort_keys=True))
+    else:
+        run(args, 1)
+
+
+def run(args: argparse.Namespace, scale: int) -> dict[str, object]:
     root = Path(__file__).resolve().parents[1]
     workspace_root = args.workspace_root.resolve(strict=True)
     filesystem = subprocess.run(
@@ -125,11 +309,26 @@ def main() -> None:
         runtime_tmp.mkdir()
         environment["TMPDIR"] = str(runtime_tmp)
         evidence_path = work / "evidence.json"
+        profile = json.loads((root / "fixtures/progressive/tiny-executable.json").read_text())
+        generator_args = profile["phases"][1]["action"]["args"]
+        generator_args[generator_args.index("--scale") + 1] = str(scale)
+        if args.growth:
+            payloads = [
+                "MATCH (n) RETURN n.node_uuid AS id ORDER BY id",
+                "MATCH ()-[r]->() RETURN r.edge_uuid AS id ORDER BY id",
+            ]
+            for phase_name, indices in (("query", (0, 1)), ("reopen_proof", (2, 3))):
+                phase = next(item for item in profile["phases"] if item["phase"] == phase_name)
+                for index, query in zip(indices, payloads, strict=True):
+                    command = phase["action"]["commands"][index]
+                    command[command.index("--cypher") + 1] = query
+        profile_path = work / "profile.json"
+        profile_path.write_text(json.dumps(profile))
         completed = subprocess.run(
             [
                 str(args.certify),
                 "run",
-                str(root / "fixtures/progressive/tiny-executable.json"),
+                str(profile_path),
                 str(evidence_path),
             ],
             cwd=work,
@@ -207,12 +406,12 @@ def main() -> None:
         }
         if [len(queries[name]) for name in ("recount", "query", "reopen_proof")] != [2, 2, 4]:
             raise SystemExit("tiny lifecycle omitted ordinary source/imported query receipts")
-        for index, expected in enumerate((2, 32)):
+        for index, expected in enumerate((1 << scale, (1 << scale) * 16)):
             if (
                 queries["recount"][index].get("scalar_u64") != expected
                 or queries["reopen_proof"][index].get("scalar_u64") != expected
             ):
-                raise SystemExit("tiny stored/imported counts differ from SCALE1 raw input")
+                raise SystemExit("tiny stored/imported counts differ from deterministic raw input")
         for source, imported in zip(
             queries["recount"] + queries["query"], queries["reopen_proof"], strict=True
         ):
@@ -264,17 +463,35 @@ def main() -> None:
                 f"lifecycle retained allocation {retained} "
                 f"differs from independent union {measured}"
             )
-        for name, private in {
-            "source-project-construction": workspace / "source" / ".graphforge-construction",
-            "source-project-import": workspace / "source" / "import-sessions",
-            "source-project-transactions": workspace / "source" / "transactions",
-            "imported-project-transactions": workspace / "imported" / "transactions",
-        }.items():
-            measured_owner = storage_totals(private)
-            if measured_owner["allocated_bytes"] <= 0:
-                raise SystemExit("tiny lifecycle omitted live private/control allocation")
+        for name, paths in owner_inventories(workspace).items():
+            measured_owner = storage_totals(workspace, paths)
             if receipt["retained_owners"][name]["totals"] != measured_owner:
                 raise SystemExit(f"lifecycle owner {name} differs from independent raw facts")
+        for name in (
+            "source-project-construction",
+            "source-project-import",
+            "source-project-transactions",
+            "imported-project-transactions",
+        ):
+            if receipt["retained_owners"][name]["totals"]["allocated_bytes"] <= 0:
+                raise SystemExit("tiny lifecycle omitted live private/control allocation")
+
+        return {
+            "scale": scale,
+            "live_nodes": queries["recount"][0]["scalar_u64"],
+            "live_edges": queries["recount"][1]["scalar_u64"],
+            "query_rows": [q["rows"] for q in queries["query"]],
+            "imported_query_rows": [q["rows"] for q in queries["reopen_proof"][2:]],
+            "receipt": receipt,
+            "allocation_quantum": os.statvfs(workspace).f_frsize,
+            "ratios": {
+                field: {
+                    "per_live_node": [receipt[field], queries["recount"][0]["scalar_u64"]],
+                    "per_live_edge": [receipt[field], queries["recount"][1]["scalar_u64"]],
+                }
+                for field in ("retained_storage_bytes", "transient_peak_storage_bytes")
+            },
+        }
 
 
 if __name__ == "__main__":