fix(bench): await Fly app readiness before remote build

benchmarks/README.md

@@ -158,7 +158,11 @@ PYTHONPATH=benchmarks/harness uv run --project benchmarks python -m \
 Only after that exact plan is accepted, append `--execute
 --confirm-disposable`. Execution owns exactly one app, one image attachment,
 one volume, and one Machine. It persists the local ownership ledger immediately
-after every create, retrieves only the existing closed sanitized evidence, and
+after every create. Before starting the one remote build, bounded
+deadline/backoff polling requires the new app to be visible and empty through
+the same Fly Machines authority used by deploy. A permanently unready app
+returns typed `readiness_timeout` evidence, tears down, and never retries the
+build. The executor retrieves only the existing closed sanitized evidence and
 always runs child-first teardown. Before deleting the app it independently
 requires empty Machine, volume, and secret inventories; afterwards it requires
 the app absent and leaves a cleared ledger. The logged-in `flyctl` credential

benchmarks/harness/graphforge_bench/fly_adapter.py

@@ -48,6 +48,7 @@ class AdapterError(ValueError):
 FAILURE_TYPES = frozenset(
     {
         "authorization_refused",
+        "readiness_timeout",
         "build_failed",
         "provision_failed",
         "lifecycle_failed",

benchmarks/harness/graphforge_bench/fly_tiny_qualification.py

@@ -8,7 +8,7 @@
 from __future__ import annotations
 
 import argparse
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from contextlib import suppress
 from dataclasses import dataclass
 import importlib.util
@@ -46,6 +46,10 @@
 CREATE_TIMEOUT_SECONDS = 180
 RETRIEVAL_TIMEOUT_SECONDS = 1_260
 SFTP_TIMEOUT_SECONDS = 120
+APP_READINESS_TIMEOUT_SECONDS = 60
+APP_READINESS_PROBE_TIMEOUT_SECONDS = 5
+APP_READINESS_INITIAL_BACKOFF_SECONDS = 0.25
+APP_READINESS_MAX_BACKOFF_SECONDS = 2.0
 TEARDOWN_POLL_ATTEMPTS = 6
 TEARDOWN_POLL_INTERVAL_SECONDS = 2
 
@@ -335,6 +339,59 @@ def _machine_id_for_name(value: Any, name: str) -> str:
     return machine_id
 
 
+def wait_for_app_readiness(
+    transport: Transport,
+    invocation: TinyQualificationInvocation,
+    *,
+    timeout_seconds: float = APP_READINESS_TIMEOUT_SECONDS,
+    clock: Callable[[], float] | None = None,
+    sleeper: Callable[[float], None] | None = None,
+) -> None:
+    """Wait until the new app is usable through Fly's Machines authority."""
+    clock = clock or time.monotonic
+    sleeper = sleeper or time.sleep
+    deadline = clock() + timeout_seconds
+    backoff = APP_READINESS_INITIAL_BACKOFF_SECONDS
+    command = ("flyctl", "machine", "list", "--app", invocation.app, "--json")
+    while True:
+        remaining = deadline - clock()
+        if remaining <= 0:
+            raise QualificationError(
+                "readiness_timeout", "created app did not become ready for remote build"
+            )
+        if remaining < 1:
+            raise QualificationError(
+                "readiness_timeout", "created app did not become ready for remote build"
+            )
+        try:
+            machines = _list(
+                transport.json(
+                    command,
+                    timeout=min(APP_READINESS_PROBE_TIMEOUT_SECONDS, int(remaining)),
+                ),
+                "Machine",
+            )
+        except (subprocess.SubprocessError, OSError):
+            machines = None
+        if machines is not None:
+            if clock() >= deadline:
+                raise QualificationError(
+                    "readiness_timeout", "created app did not become ready for remote build"
+                )
+            if machines:
+                raise QualificationError(
+                    "provision_failed", "new app is not empty at readiness admission"
+                )
+            return
+        sleep_for = min(backoff, max(0.0, deadline - clock()))
+        if sleep_for <= 0:
+            raise QualificationError(
+                "readiness_timeout", "created app did not become ready for remote build"
+            )
+        sleeper(sleep_for)
+        backoff = min(backoff * 2, APP_READINESS_MAX_BACKOFF_SECONDS)
+
+
 def verify_machine_state(
     value: Any,
     invocation: TinyQualificationInvocation,
@@ -741,6 +798,12 @@ def execute(
         ledger.app_owned = True
         _save_ledger(ledger_path, ledger)
 
+        wait_for_app_readiness(
+            transport,
+            invocation,
+            timeout_seconds=APP_READINESS_TIMEOUT_SECONDS,
+        )
+
         failure_kind = "build_failed"
         build = remote_build_command(
             app=invocation.app,

benchmarks/tests/test_fly_tiny_qualification.py

@@ -69,6 +69,7 @@ def __init__(
         auto_destroy: bool = True,
         restart_state: object = "valid",
         mount_state: object = "valid",
+        readiness_responses: list[object] | None = None,
     ):
         self.commands: list[tuple[str, ...]] = []
         self.machine_lists = 0
@@ -81,6 +82,7 @@ def __init__(
         self.mount_state = (
             {"volume": "vol_fixture123", "path": "/work"} if mount_state == "valid" else mount_state
         )
+        self.readiness_responses = list(readiness_responses or [])
         self.app_created = False
 
     def run(
@@ -149,6 +151,13 @@ def json(self, argv: tuple[str, ...], *, timeout: int) -> object:
         if argv[1:3] == ("volumes", "create"):
             return {"id": "vol_fixture123"}
         if argv[1:3] == ("machine", "list"):
+            if self.app_created and not self.machine_created and self.readiness_responses:
+                response = self.readiness_responses.pop(0)
+                if isinstance(response, BaseException):
+                    raise response
+                if isinstance(response, list) and response:
+                    self.machine_created = True
+                return response
             if not self.machine_created:
                 return []
             self.machine_lists += 1
@@ -290,6 +299,152 @@ def test_build_failure_is_typed_and_still_destroys_app(self, check_source: objec
                 transport.commands,
             )
 
+    @patch("graphforge_bench.fly_tiny_qualification.check_source")
+    def test_readiness_delayed_convergence_precedes_single_build(
+        self, check_source: object
+    ) -> None:
+        del check_source
+        unavailable = subprocess.CalledProcessError(1, ("flyctl", "machine", "list"))
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            transport = FakeTransport(readiness_responses=[unavailable, unavailable, []])
+            with patch("graphforge_bench.fly_tiny_qualification.time.sleep") as sleep:
+                result = execute(
+                    invocation(),
+                    transport=transport,
+                    root=root,
+                    ledger_path=root / "ledger.json",
+                    evidence_out=root / "evidence.json",
+                    dry_run=False,
+                )
+            self.assertEqual(result["status"], "passed")
+            deploy_index = next(
+                index
+                for index, command in enumerate(transport.commands)
+                if command[:2] == ("flyctl", "deploy")
+            )
+            readiness = (
+                "flyctl",
+                "machine",
+                "list",
+                "--app",
+                "gf-q958-test",
+                "--json",
+            )
+            self.assertEqual(transport.commands[:deploy_index].count(readiness), 3)
+            self.assertEqual(sleep.call_count, 2)
+
+    @patch("graphforge_bench.fly_tiny_qualification.check_source")
+    def test_readiness_permanent_absence_times_out_and_cleans_owned_app(
+        self, check_source: object
+    ) -> None:
+        del check_source
+        unavailable = subprocess.CalledProcessError(1, ("flyctl", "machine", "list"))
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            transport = FakeTransport(readiness_responses=[unavailable])
+            with (
+                patch("graphforge_bench.fly_tiny_qualification.APP_READINESS_TIMEOUT_SECONDS", 1),
+                patch(
+                    "graphforge_bench.fly_tiny_qualification.time.monotonic",
+                    side_effect=(0.0, 0.0, 2.0),
+                ),
+                patch("graphforge_bench.fly_tiny_qualification.time.sleep"),
+            ):
+                result = execute(
+                    invocation(),
+                    transport=transport,
+                    root=root,
+                    ledger_path=root / "ledger.json",
+                    evidence_out=root / "evidence.json",
+                    dry_run=False,
+                )
+            self.assertEqual(result["failure"], "readiness_timeout")
+            self.assertFalse(
+                any(command[:2] == ("flyctl", "deploy") for command in transport.commands)
+            )
+            self.assertEqual(ResourceLedger.load(root / "ledger.json"), ResourceLedger())
+            self.assertFalse((root / "evidence.json").exists())
+            self.assertNotIn("token", json.dumps(result).lower())
+
+    @patch("graphforge_bench.fly_tiny_qualification.check_source")
+    def test_readiness_response_after_deadline_cannot_start_build(
+        self, check_source: object
+    ) -> None:
+        del check_source
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            transport = FakeTransport(readiness_responses=[[]])
+            with (
+                patch("graphforge_bench.fly_tiny_qualification.APP_READINESS_TIMEOUT_SECONDS", 2),
+                patch(
+                    "graphforge_bench.fly_tiny_qualification.time.monotonic",
+                    side_effect=(0.0, 0.0, 3.0),
+                ),
+            ):
+                result = execute(
+                    invocation(),
+                    transport=transport,
+                    root=root,
+                    ledger_path=root / "ledger.json",
+                    evidence_out=root / "evidence.json",
+                    dry_run=False,
+                )
+            self.assertEqual(result["failure"], "readiness_timeout")
+            self.assertFalse(
+                any(command[:2] == ("flyctl", "deploy") for command in transport.commands)
+            )
+            self.assertEqual(ResourceLedger.load(root / "ledger.json"), ResourceLedger())
+            self.assertFalse((root / "evidence.json").exists())
+
+    @patch("graphforge_bench.fly_tiny_qualification.check_source")
+    def test_readiness_malformed_inventory_is_typed_provider_failure(
+        self, check_source: object
+    ) -> None:
+        del check_source
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            transport = FakeTransport(readiness_responses=[{}])
+            result = execute(
+                invocation(),
+                transport=transport,
+                root=root,
+                ledger_path=root / "ledger.json",
+                evidence_out=root / "evidence.json",
+                dry_run=False,
+            )
+            self.assertEqual(result["failure"], "provision_failed")
+            self.assertEqual(ResourceLedger.load(root / "ledger.json"), ResourceLedger())
+            self.assertFalse((root / "evidence.json").exists())
+
+    @patch("graphforge_bench.fly_tiny_qualification.check_source")
+    def test_readiness_rejects_and_adopts_unexpected_child_for_cleanup(
+        self, check_source: object
+    ) -> None:
+        del check_source
+        adopted = [{"id": "abcdef01234567", "name": "unexpected-machine"}]
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            transport = FakeTransport(readiness_responses=[adopted])
+            result = execute(
+                invocation(),
+                transport=transport,
+                root=root,
+                ledger_path=root / "ledger.json",
+                evidence_out=root / "evidence.json",
+                dry_run=False,
+            )
+            self.assertEqual(result["failure"], "provision_failed")
+            self.assertTrue(
+                any(
+                    command[:3] == ("flyctl", "machine", "destroy")
+                    for command in transport.commands
+                )
+            )
+            self.assertEqual(ResourceLedger.load(root / "ledger.json"), ResourceLedger())
+            self.assertFalse((root / "evidence.json").exists())
+            self.assertNotIn("token", json.dumps(result).lower())
+
     @patch("graphforge_bench.fly_tiny_qualification.check_source")
     def test_nonempty_teardown_inventory_fails_closed(self, check_source: object) -> None:
         del check_source