build(bench): add offline provider qualification runner

.github/workflows/test.yml

@@ -85,6 +85,9 @@ jobs:
       - name: Test Fly qualification safety contract
         run: python3 scripts/ci/test-fly-filesystem-safety-contract.py
 
+      - name: Test progressive qualification image contract
+        run: python3 scripts/ci/test-progressive-qualification-image.py
+
       - name: Test Repository Policy suite classifier
         run: scripts/ci/test-classify-policy-suites.sh
 
@@ -1169,6 +1172,27 @@ jobs:
         with:
           enable-cache: true
 
+      - name: Test offline progressive provider boundary
+        working-directory: benchmarks
+        run: |
+          PYTHONPATH=harness uv run --locked python -m unittest \
+            tests.test_progressive_provider_plan \
+            tests.test_progressive_provider_run \
+            tests.test_progressive_run \
+            tests.test_qualification_operator
+
+      - name: Build immutable progressive qualification image
+        shell: bash
+        run: |
+          set -euo pipefail
+          source_sha=$(git rev-parse HEAD)
+          image="graphforge-progressive-qualification:${source_sha}"
+          docker build --platform linux/amd64 \
+            --build-arg "GRAPHFORGE_COMMIT=${source_sha}" \
+            --file containers/graphforge-progressive-qualification/Dockerfile \
+            --tag "$image" .
+          test "$(docker image inspect --format '{{ index .Config.Labels "org.opencontainers.image.revision" }}' "$image")" = "$source_sha"
+
       - name: Cargo feature drift check (fail-closed)
         run: |
           python3 scripts/ci/cargo-bazel-drift-check.py

benchmarks/Makefile

@@ -59,6 +59,7 @@ progressive-provider-plan: install
 		plan-progressive -- \
 		--root "$(CURDIR)" --output-dir "$(OUTPUT_DIR)" --commit "$(COMMIT)" \
 		--maximum-scale "$(MAXIMUM_SCALE)" $(if $(PROVIDER_CAPACITY),--provider-capacity "$(PROVIDER_CAPACITY)",) \
+		$(if $(IMAGE_DIGEST),--image-digest "$(IMAGE_DIGEST)",) \
 		--plan-out "$(PLAN_OUT)"
 
 # The registry owns the live command. Pulumi ESC supplies provider credentials;

benchmarks/README.md

@@ -307,14 +307,28 @@ make -C benchmarks progressive-provider-plan \
   COMMIT=$(git rev-parse HEAD) MAXIMUM_SCALE=26 \
   OUTPUT_DIR=/admitted-volume/graphforge-evidence \
   PLAN_OUT=/admitted-volume/graphforge-evidence/provider-plan.json \
+  IMAGE_DIGEST=registry.fly.io/graphforge-bench@sha256:<64-hex-digest> \
   PROVIDER_CAPACITY=/sanitized/provider-capacity.json
 ```
 
-The planner is deliberately not a provider executor. Provider plans are marked
-refused for execution until a dedicated provider image contains `gf`, the
-Graph500 generator/certifier, Python/BenchExec, and a proven cgroups/I/O
-authority. The canonical order remains S18, S19, S20, S22, S24, S25, S26;
-the first failed or missing gate stops the ladder.
+An admitted S20--S26 plan can be consumed by the offline provider runner inside
+the dedicated `containers/graphforge-progressive-qualification` image. The
+runner requires the image's read-only build manifest, matches the immutable
+`registry.fly.io/...@sha256:...` identity supplied by the admitted plan and
+provider transport, and revalidates the canonical profile, projection, source
+tree, native executables, and BenchExec identity before starting BenchExec. It
+does not self-attest its OCI digest: the trusted provider transport must read
+the provider-observed Machine image digest and supply that matching value.
+It accepts only fixed in-image executable paths, an admitted plan and evidence
+directory below `/work`, no pre-existing files for the selected rung, and a
+real `/work` mount. The host
+must separately pass native Linux cgroups-v2 admission. The runner has no
+laptop fallback and never calls a provider API.
+
+A successful rung emits exactly `sN-plan.json`, `sN-benchexec.json`,
+`sN-graphforge.json`, `sN-rung.json`, and `sN-result.json`, all scoped to
+engineering evidence. The canonical order remains S18, S19, S20, S22, S24,
+S25, S26; the first failed or missing gate stops the planner.
 
 Provider credentials belong to Pulumi ESC rather than GitHub workflow inputs or
 the caller's ambient shell. Live operator commands are rendered from
@@ -329,9 +343,11 @@ make -C benchmarks qualification-operator \
 
 The operator uses the shell-free form `pulumi env run <environment> -- <argv>`;
 secret values are never copied into its command line or evidence. The
-`progressive-ladder` command is registered now, but fails before opening ESC
-until the dedicated provider image and scale BenchExec executor are present.
-This is a deliberate capability boundary, not a GitHub-dispatch prerequisite.
+`progressive-ladder` is registered, but still fails before opening ESC. The
+offline image and rung runner do not implement whole-attempt Fly orchestration,
+typed spend authorization, ownership-ledger recovery, or teardown inventory.
+This is a deliberate live capability boundary, not a GitHub-dispatch
+prerequisite.
 
 The controller derives bulk-ingest capability from the same run's bounded
 ordinary `gf import-session commit --json` receipt: its construction evidence

benchmarks/harness/graphforge_bench/progressive_provider_plan.py

@@ -29,8 +29,8 @@
 
 SCALES = (18, 19, 20, 22, 24, 25, 26)
 COMMIT = re.compile(r"^[0-9a-f]{40}$")
+IMAGE_DIGEST = re.compile(r"^registry\.fly\.io/[a-z0-9][a-z0-9._/-]*@sha256:[0-9a-f]{64}$")
 PLAN_SCHEMA = "graphforge-progressive-provider-plan/1"
-EXECUTION_REFUSAL = "provider_executor_unavailable"
 
 
 class ProviderPlanError(ValueError):
@@ -87,14 +87,14 @@ def _validate_rung(root: Path, value: Any, scale: int) -> Mapping[str, Any]:
     error = next(Draft202012Validator(schema).iter_errors(value), None)
     if error is not None:
         raise ProviderPlanError("completed rung evidence is not schema-valid")
-    expected_source = "progressive_profile" if scale in (18, 19) else None
+    expected_source = "progressive_profile" if scale in (18, 19) else "canonical_ladder"
     if (
         value.get("scale") != scale
         or value.get("status") != "passed"
         or value.get("live_edges") != 16 * (1 << scale)
         or value.get("profile_id")
         != f"graph500-s{scale}-{'local' if scale in (18, 19) else 'provider'}"
-        or (expected_source is not None and value.get("source") != expected_source)
+        or value.get("source") != expected_source
     ):
         raise ProviderPlanError("completed rung evidence does not match its canonical profile")
     return value
@@ -135,6 +135,40 @@ def _validate_result_identity(
         raise ProviderPlanError(
             "completed rung result is not bound to the requested commit/profile"
         )
+    if scale not in (18, 19):
+        artifacts = value.get("artifacts")
+        execution_plan = _read_json(
+            output_dir / f"s{scale}-plan.json",
+            "completed provider execution plan is unavailable",
+        )
+        execution_schema = _read_json(
+            root / "schemas" / "progressive-provider-run-plan.json",
+            "provider execution plan schema is unavailable",
+        )
+        execution_error = next(
+            Draft202012Validator(execution_schema).iter_errors(execution_plan), None
+        )
+        if (
+            execution_error is not None
+            or not isinstance(execution_plan, Mapping)
+            or execution_plan.get("rung") != f"S{scale}"
+            or execution_plan.get("identities") != identities
+        ):
+            raise ProviderPlanError("completed provider plan/result identities disagree")
+        try:
+            expected_artifacts = {
+                name: hashlib.sha256(path.read_bytes()).hexdigest()
+                for name, path in {
+                    "plan_sha256": output_dir / f"s{scale}-plan.json",
+                    "benchexec_sha256": output_dir / f"s{scale}-benchexec.json",
+                    "graphforge_sha256": output_dir / f"s{scale}-graphforge.json",
+                    "rung_sha256": output_dir / f"s{scale}-rung.json",
+                }.items()
+            }
+        except OSError as error:
+            raise ProviderPlanError("completed provider artifacts are unavailable") from error
+        if not isinstance(artifacts, Mapping) or dict(artifacts) != expected_artifacts:
+            raise ProviderPlanError("completed provider artifacts do not match their result")
 
 
 def completed_rungs(
@@ -177,6 +211,7 @@ def plan_provider_ladder(
     commit: str,
     maximum_scale: int,
     provider_capacity: Mapping[str, Any] | None = None,
+    image_digest: str | None = None,
 ) -> dict[str, Any]:
     """Return one immutable, sanitized next-rung plan without provider calls."""
     if COMMIT.fullmatch(commit) is None:
@@ -198,6 +233,8 @@ def plan_provider_ladder(
     profile_path = _profile_path(root, selected)
     projection: Mapping[str, Any] | None = None
     if selected.execution == "provider":
+        if image_digest is None or IMAGE_DIGEST.fullmatch(image_digest) is None:
+            raise ProviderPlanError("immutable provider image digest is required")
         try:
             projection = project(selected, completed, provider_capacity)
         except QualificationError as error:
@@ -222,9 +259,10 @@ def plan_provider_ladder(
         "profile_id": selected.id,
         "profile_path": profile_path.relative_to(root).as_posix(),
         "profile_sha256": "sha256:" + _sha256(profile_path),
+        "image_digest": image_digest if selected.execution == "provider" else None,
         "projection": projection,
-        "execution_authorized": selected.execution == "local",
-        "execution_refusal": None if selected.execution == "local" else EXECUTION_REFUSAL,
+        "execution_authorized": True,
+        "execution_refusal": None,
         "claim": "engineering_evidence_only",
     }
     schema_path = root / "schemas" / "progressive-provider-plan.json"
@@ -236,22 +274,20 @@ def plan_provider_ladder(
 
 
 def require_execution_authority(plan: Mapping[str, Any]) -> None:
-    """Refuse provider execution until its image and resource authority exist."""
-    if plan.get("execution") == "provider":
-        raise ProviderPlanError(
-            "provider execution is refused until a dedicated provider image and BenchExec "
-            "authority exist"
-        )
+    """Require the checked-in offline runner before handing a plan to execution."""
+    if plan.get("execution_authorized") is not True or plan.get("execution_refusal") is not None:
+        raise ProviderPlanError("provider execution authority is unavailable")
 
 
 def main(argv: list[str] | None = None) -> int:
     """Write one no-spend provider plan for a protected workflow step."""
-    parser = argparse.ArgumentParser(description=__doc__)
+    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
     parser.add_argument("--root", type=Path, required=True)
     parser.add_argument("--output-dir", type=Path, required=True)
     parser.add_argument("--commit", required=True)
     parser.add_argument("--maximum-scale", type=int, required=True)
     parser.add_argument("--provider-capacity", type=Path)
+    parser.add_argument("--image-digest")
     parser.add_argument("--plan-out", type=Path, required=True)
     args = parser.parse_args(argv)
     try:
@@ -267,6 +303,7 @@ def main(argv: list[str] | None = None) -> int:
             commit=args.commit,
             maximum_scale=args.maximum_scale,
             provider_capacity=capacity,
+            image_digest=args.image_digest,
         )
         args.plan_out.parent.mkdir(parents=True, exist_ok=True)
         args.plan_out.write_text(