feat(benchmarks): add offline progressive Fly transport

.github/workflows/test.yml

@@ -1176,6 +1176,8 @@ jobs:
         working-directory: benchmarks
         run: |
           PYTHONPATH=harness uv run --locked python -m unittest \
+            tests.test_progressive_esc \
+            tests.test_progressive_fly_transport \
             tests.test_progressive_provider_attempt \
             tests.test_progressive_provider_plan \
             tests.test_progressive_provider_run \

benchmarks/Makefile

@@ -1,4 +1,4 @@
-.PHONY: install smoke smoke-python smoke-rust fly-adapter-static progressive-provider-attempt-static local-admission progressive-qualification-list progressive-qualification-plan progressive-qualification-run progressive-qualification-project-s20 progressive-qualification-binaries progressive-provider-plan qualification-operator
+.PHONY: install smoke smoke-python smoke-rust fly-adapter-static progressive-provider-attempt-static progressive-fly-transport-static local-admission progressive-qualification-list progressive-qualification-plan progressive-qualification-run progressive-qualification-project-s20 progressive-qualification-binaries progressive-provider-plan qualification-operator
 
 install:
 	uv sync --locked
@@ -24,6 +24,12 @@ progressive-provider-attempt-static: install
 	PYTHONPATH=harness uv run --locked python -m unittest \
 		tests.test_progressive_provider_attempt
 
+# Provider-free proof of the Fly command boundary and ESC secret capsule.
+progressive-fly-transport-static: install
+	PYTHONPATH=harness uv run --locked python -m unittest \
+		tests.test_progressive_fly_transport \
+		tests.test_progressive_esc
+
 local-admission: install
 	PYTHONPATH=$(CURDIR)/harness uv run --locked reframe -C reframe/settings.py -c reframe/checks -n '^LocalBenchExecAdmission$$' -l | grep -F LocalBenchExecAdmission
 	PYTHONPATH=$(CURDIR)/harness uv run --locked reframe -C reframe/settings.py -c reframe/checks -n '^LocalBenchExecAdmission$$' -r

benchmarks/README.md

@@ -343,6 +343,22 @@ canonical five-file bundle; and persists an fsync-backed ownership ledger for
 cleanup-only recovery. Its transport is injected, so this proof makes no
 provider calls and spends nothing.
 
+The production-shaped Fly boundary and isolated ESC input capsule are exercised
+offline with:
+
+```bash
+make -C benchmarks progressive-fly-transport-static
+```
+
+The transport accepts only an already-published immutable image, emits fixed
+shell-free Fly commands through an injected boundary, applies the attempt
+deadline to every operation, validates provider-observed Machine identity,
+retrieves the result before the remaining canonical artifacts, and returns only
+sanitized teardown counts. The ESC capsule consumes the fixed projected token
+and spend-authorization variables once, removes them from the ambient process,
+and constructs a minimal child environment with fresh credential state. Both
+components remain import-only: these tests perform no provider operation.
+
 Provider credentials belong to Pulumi ESC rather than GitHub workflow inputs or
 the caller's ambient shell. Live operator commands are rendered from
 `config/gate-registry.json` and run through the Python control plane:
@@ -357,10 +373,12 @@ 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` is registered, but still fails before opening ESC. The
-typed whole-attempt state machine, ownership-ledger recovery, and sanitized
-teardown inventory now exist offline; the real Fly transport, ESC environment
-binding, and live recovery gate remain deliberately unavailable. This is a
-live capability boundary, not a GitHub-dispatch prerequisite.
+typed whole-attempt state machine, production-shaped Fly command boundary,
+isolated ESC input capsule, ownership-ledger recovery, and sanitized teardown
+inventory now exist offline. Protected/versioned ESC configuration, immutable
+image publication, an independently scheduled recovery owner or lease, and the
+separately reviewed spend authorization remain prerequisites for live wiring.
+This is a 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_esc.py

@@ -0,0 +1,228 @@
+"""Credential-isolated ESC inputs for the progressive Fly controller.
+
+This module is deliberately import-only.  It does not invoke Pulumi, Fly, or
+the progressive attempt controller and is not wired to an operator command.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator, Mapping, MutableMapping
+from dataclasses import dataclass, field
+import os
+from pathlib import Path
+import tempfile
+from typing import Any
+
+from graphforge_bench.progressive_provider_attempt import (
+    SpendAuthorization,
+    parse_spend_authorization,
+)
+
+FLY_TOKEN_ENV = "FLY_API_TOKEN"
+SPEND_AUTHORIZATION_ENV = "GRAPHFORGE_PROGRESSIVE_SPEND_AUTHORIZATION"
+
+_CREDENTIAL_ALIASES = frozenset({"FLY_ACCESS_TOKEN"})
+_FORBIDDEN_OVERRIDES = frozenset(
+    {
+        "ALL_PROXY",
+        "DEBUG",
+        "FLY_DEBUG",
+        "FLY_LOG_LEVEL",
+        "HTTP_PROXY",
+        "HTTPS_PROXY",
+        "LOG_LEVEL",
+        "NO_PROXY",
+        "PULUMI_LOG_LEVEL",
+        "RUST_LOG",
+    }
+)
+_PROVIDER_PATH = "/usr/local/bin:/usr/bin:/bin"
+
+
+class EscCapsuleError(ValueError):
+    """A sanitized refusal at the protected environment boundary."""
+
+
+class _Secret:
+    """Small non-printing holder for provider credential material."""
+
+    __slots__ = ("_value",)
+
+    def __init__(self, value: str):
+        self._value = value
+
+    def copy(self) -> str:
+        return self._value
+
+    def clear(self) -> None:
+        self._value = ""
+
+    def __repr__(self) -> str:
+        return "<redacted>"
+
+
+class _ProviderEnvironment(Mapping[str, str]):
+    """A subprocess-compatible mapping whose representation stays redacted."""
+
+    __slots__ = ("_fly_token", "_values")
+
+    def __init__(self, fly_token: _Secret, values: Mapping[str, str]):
+        self._fly_token = fly_token
+        self._values = dict(values)
+
+    def __getitem__(self, name: str) -> str:
+        if name == FLY_TOKEN_ENV:
+            return self._fly_token.copy()
+        return self._values[name]
+
+    def __iter__(self) -> Iterator[str]:
+        yield FLY_TOKEN_ENV
+        yield from self._values
+
+    def __len__(self) -> int:
+        return len(self._values) + 1
+
+    def __repr__(self) -> str:
+        return "ProviderEnvironment(FLY_API_TOKEN=<redacted>, isolated_config=True)"
+
+
+@dataclass(repr=False)
+class ProgressiveEscCapsule:
+    """Validated ESC authority and an isolated environment for provider calls."""
+
+    _fly_token: _Secret
+    _authorization: SpendAuthorization | None
+    _temporary: tempfile.TemporaryDirectory[str]
+    home: Path
+    xdg_config_home: Path
+    _authorization_taken: bool = field(default=False, init=False)
+    _closed: bool = field(default=False, init=False)
+    _cleanup_complete: bool = field(default=False, init=False)
+
+    def __repr__(self) -> str:
+        return "ProgressiveEscCapsule(fly_token=<redacted>, authorization=<redacted>)"
+
+    def take_spend_authorization(self) -> SpendAuthorization:
+        """Return the parsed authority once, without retaining its encoded form."""
+        if self._closed or self._authorization_taken or self._authorization is None:
+            raise EscCapsuleError("protected spend authorization is unavailable")
+        authorization = self._authorization
+        self._authorization = None
+        self._authorization_taken = True
+        return authorization
+
+    def subprocess_environment(self) -> Mapping[str, str]:
+        """Build the complete, minimal environment for one Fly subprocess."""
+        if self._closed:
+            raise EscCapsuleError("ESC capsule is closed")
+        return _ProviderEnvironment(
+            self._fly_token,
+            {
+                "HOME": str(self.home),
+                "LANG": "C.UTF-8",
+                "LC_ALL": "C.UTF-8",
+                "PATH": _PROVIDER_PATH,
+                "XDG_CONFIG_HOME": str(self.xdg_config_home),
+            },
+        )
+
+    def close(self) -> None:
+        if self._cleanup_complete:
+            return
+        self._closed = True
+        self._fly_token.clear()
+        self._authorization = None
+        self._temporary.cleanup()
+        self._cleanup_complete = True
+
+    def __enter__(self) -> ProgressiveEscCapsule:
+        if self._closed:
+            raise EscCapsuleError("ESC capsule is closed")
+        return self
+
+    def __exit__(self, *_exc: Any) -> None:
+        self.close()
+
+
+def _pop_projected_inputs(
+    environ: MutableMapping[str, str],
+) -> tuple[str | None, str | None, bool]:
+    token: str | None = None
+    authorization: str | None = None
+    rejected = False
+    protected = {FLY_TOKEN_ENV, SPEND_AUTHORIZATION_ENV}
+    for name in list(environ):
+        normalized = name.upper()
+        if normalized not in protected | _CREDENTIAL_ALIASES:
+            continue
+        value = environ.pop(name)
+        if name == FLY_TOKEN_ENV and token is None:
+            token = value
+        elif name == SPEND_AUTHORIZATION_ENV and authorization is None:
+            authorization = value
+        else:
+            rejected = True
+    return token, authorization, rejected
+
+
+def _reject_ambient_overrides(environ: MutableMapping[str, str]) -> None:
+    if any(name.upper() in _FORBIDDEN_OVERRIDES for name in environ):
+        raise EscCapsuleError("ambient credential or network override is forbidden")
+
+
+def _validate_token(value: str | None) -> str:
+    if (
+        not isinstance(value, str)
+        or not 1 <= len(value) <= 8192
+        or value != value.strip()
+        or any(ord(character) < 0x20 or ord(character) == 0x7F for character in value)
+    ):
+        raise EscCapsuleError("projected Fly credential is unavailable or malformed")
+    return value
+
+
+def _parse_authorization(value: str) -> SpendAuthorization | None:
+    """Keep parser exceptions and their protected input outside the public boundary."""
+    try:
+        return parse_spend_authorization(value)
+    except Exception:
+        return None
+
+
+def load_progressive_esc(
+    environ: MutableMapping[str, str] | None = None,
+) -> ProgressiveEscCapsule:
+    """Consume exactly the two protected projections from the process environment."""
+    source = os.environ if environ is None else environ
+    token_value, authorization_value, rejected_projection = _pop_projected_inputs(source)
+    try:
+        if rejected_projection:
+            raise EscCapsuleError("ambient credential or projected-input override is forbidden")
+        _reject_ambient_overrides(source)
+        token = _Secret(_validate_token(token_value))
+        if not isinstance(authorization_value, str):
+            raise EscCapsuleError("protected spend authorization is unavailable")
+        authorization = _parse_authorization(authorization_value)
+        if authorization is None:
+            token.clear()
+            raise EscCapsuleError("protected spend authorization is invalid")
+    finally:
+        token_value = None
+        authorization_value = None
+
+    try:
+        temporary = tempfile.TemporaryDirectory(prefix="graphforge-progressive-esc-")
+    except Exception:
+        token.clear()
+        raise
+    root = Path(temporary.name)
+    home = root / "home"
+    xdg_config_home = root / "xdg"
+    try:
+        home.mkdir(mode=0o700)
+        xdg_config_home.mkdir(mode=0o700)
+    except Exception:
+        token.clear()
+        temporary.cleanup()
+        raise
+    return ProgressiveEscCapsule(token, authorization, temporary, home, xdg_config_home)