From edd03047882986c216c1c0f4c6e3378885802e5b Mon Sep 17 00:00:00 2001 From: David Spencer <1526975+DecisionNerd@users.noreply.github.com> Date: Sun, 30 Aug 2026 22:10:27 -0600 Subject: [PATCH 1/2] feat(benchmarks): add progressive attempt controller core Refs #1034 --- .github/workflows/test.yml | 1 + benchmarks/Makefile | 7 +- benchmarks/README.md | 21 +- .../progressive_provider_attempt.py | 1039 +++++++++++++++++ .../progressive-provider-attempt-ledger.json | 50 + .../progressive-provider-attempt-result.json | 58 + ...ogressive-provider-teardown-inventory.json | 85 ++ .../progressive-spend-authorization.json | 63 + .../test_progressive_provider_attempt.py | 691 +++++++++++ 9 files changed, 2010 insertions(+), 5 deletions(-) create mode 100644 benchmarks/harness/graphforge_bench/progressive_provider_attempt.py create mode 100644 benchmarks/schemas/progressive-provider-attempt-ledger.json create mode 100644 benchmarks/schemas/progressive-provider-attempt-result.json create mode 100644 benchmarks/schemas/progressive-provider-teardown-inventory.json create mode 100644 benchmarks/schemas/progressive-spend-authorization.json create mode 100644 benchmarks/tests/test_progressive_provider_attempt.py diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 33cd5968b..0ade9efcf 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -1176,6 +1176,7 @@ jobs: working-directory: benchmarks run: | PYTHONPATH=harness uv run --locked python -m unittest \ + tests.test_progressive_provider_attempt \ tests.test_progressive_provider_plan \ tests.test_progressive_provider_run \ tests.test_progressive_run \ diff --git a/benchmarks/Makefile b/benchmarks/Makefile index 3e42442fb..037dc33b1 100644 --- a/benchmarks/Makefile +++ b/benchmarks/Makefile @@ -1,4 +1,4 @@ -.PHONY: install smoke smoke-python smoke-rust fly-adapter-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 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 @@ -19,6 +19,11 @@ fly-adapter-static: install PYTHONPATH=harness uv run --locked python -m unittest \ tests.test_fly_adapter tests.test_fly_tiny_qualification +# Provider-free proof of the whole-attempt state machine and recovery contracts. +progressive-provider-attempt-static: install + PYTHONPATH=harness uv run --locked python -m unittest \ + tests.test_progressive_provider_attempt + 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 diff --git a/benchmarks/README.md b/benchmarks/README.md index b6a1006cd..f35dc8474 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -330,6 +330,19 @@ A successful rung emits exactly `sN-plan.json`, `sN-benchexec.json`, engineering evidence. The canonical order remains S18, S19, S20, S22, S24, S25, S26; the first failed or missing gate stops the planner. +The provider-free whole-attempt controller core is exercised with: + +```bash +make -C benchmarks progressive-provider-attempt-static +``` + +It requires a schema-valid, commit-bound S18/S19 prefix; validates a closed +five-hour, integer-micro-USD spend authorization; binds the first admitted plan +before any provider mutation; advances one rung at a time; accepts only the +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. + 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: @@ -344,10 +357,10 @@ make -C benchmarks qualification-operator \ The operator uses the shell-free form `pulumi env run -- `; secret values are never copied into its command line or evidence. The `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. +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. The controller derives bulk-ingest capability from the same run's bounded ordinary `gf import-session commit --json` receipt: its construction evidence diff --git a/benchmarks/harness/graphforge_bench/progressive_provider_attempt.py b/benchmarks/harness/graphforge_bench/progressive_provider_attempt.py new file mode 100644 index 000000000..8967853fb --- /dev/null +++ b/benchmarks/harness/graphforge_bench/progressive_provider_attempt.py @@ -0,0 +1,1039 @@ +"""Offline whole-attempt control for the progressive provider ladder. + +This module owns ordering, durable local state, evidence admission, and teardown +semantics. Provider I/O is deliberately absent: callers must supply a +``ProviderTransport`` and may inject the no-spend planner and validation hooks. +""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping, Sequence +from dataclasses import asdict, dataclass, field +from datetime import datetime, timedelta, timezone +import hashlib +import json +import os +from pathlib import Path +import re +import tempfile +from typing import Any, Protocol + +from jsonschema import Draft202012Validator + +from graphforge_bench.progressive_provider_plan import ( + ProviderPlanError, + completed_rungs, + plan_provider_ladder, +) + +AUTHORIZATION_SCHEMA = "graphforge-progressive-spend-authorization/1" +ATTEMPT_SCHEMA = "graphforge-progressive-provider-attempt-result/1" +LEDGER_SCHEMA = "graphforge-progressive-provider-attempt-ledger/1" +RESULT_SCHEMA = "graphforge-progressive-provider-run-result/1" +CANONICAL_RUNGS = (18, 19, 20, 22, 24, 25, 26) +PROVIDER_RUNGS = (20, 22, 24, 25, 26) +COMMIT = re.compile(r"^[0-9a-f]{40}$") +SAFE_NAME = re.compile(r"^[a-z0-9][a-z0-9-]{0,62}$") +SAFE_REGION = re.compile(r"^[a-z]{3}$") +IMAGE_DIGEST = re.compile(r"^registry\.fly\.io/[a-z0-9][a-z0-9._/-]*@sha256:[0-9a-f]{64}$") +ATTEMPT_ID = re.compile(r"^[0-9a-f]{32}$") +MACHINE_ID = re.compile(r"^[0-9a-f]{14}$") +VOLUME_ID = re.compile(r"^vol_[a-z0-9]+$") +SCHEMA_ROOT = Path(__file__).resolve().parents[2] / "schemas" + + +class AttemptError(RuntimeError): + """A closed failure suitable for an attempt result or operator diagnostic.""" + + def __init__(self, failure: str, message: str): + super().__init__(message) + self.failure = failure + + +@dataclass(frozen=True) +class SpendAuthorization: + """Typed, bounded authority supplied by the protected control plane.""" + + schema: str + status: str + provider: str + commit: str + admitted_plan_sha256: str + image_digest: str + organization: str + region: str + machine_class: str + volume_gib: int + rung: str + maximum_scale: int + attempt_nonce: str + app: str + issued_at: datetime + expires_at: datetime + teardown_owner: str + maximum_machine_seconds: int + resource_limits: Mapping[str, int] + pricing: Mapping[str, Any] + claim: str + authorization_sha256: str + + +@dataclass(frozen=True) +class AttemptInvocation: + """Local paths and immutable identity for one controller invocation.""" + + root: Path + evidence_dir: Path + ledger_path: Path + commit: str + provider_capacity: Mapping[str, Any] | None = None + + +@dataclass(frozen=True) +class AttemptRequest: + """Operator-facing request kept separate from protected spend authority.""" + + commit: str + organization: str + app: str + region: str + machine_class: str + volume_gib: int + image_digest: str + maximum_scale: int + spend_authorization: str | bytes | Mapping[str, Any] | None + provider_capacity: Mapping[str, Any] | None = None + + +@dataclass(frozen=True) +class ProvisionedAttempt: + """Provider-observed immutable image identity plus opaque cleanup handles.""" + + image_digest: str + resources: Mapping[str, str] = field(default_factory=dict) + + +@dataclass +class AttemptLedger: + """Durable recovery state; authorization and provider output are excluded.""" + + schema: str = LEDGER_SCHEMA + generation: int = 0 + attempt_id: str | None = None + owner_app: str | None = None + commit: str | None = None + authorization_sha256: str | None = None + admitted_plan_sha256: str | None = None + authorized_maximum_scale: int | None = None + authorized_image_digest: str | None = None + expires_at: str | None = None + evidence_dir: str | None = None + phase: str = "new" + image_digest: str | None = None + current_rung: int | None = None + completed_scales: list[int] = field(default_factory=list) + first_failed_rung: int | None = None + failure: str | None = None + cleanup_failure: str | None = None + resources: dict[str, str] = field(default_factory=dict) + teardown_observed: dict[str, Any] | None = None + teardown_checked_at: str | None = None + + +@dataclass(frozen=True) +class AttemptOutcome: + schema: str + status: str + commit: str + authorized_maximum_scale: int + completed_scales: tuple[int, ...] + first_failed_rung: int | None + failure: str | None + cleanup_failure: str | None + authorization_sha256: str + admitted_plan_sha256: str + authorized_image_digest: str + observed_image_digest: str | None + teardown_status: str + teardown_observed: Mapping[str, Any] | None + teardown_checked_at: str | None + + +class ProviderTransport(Protocol): + """The sole boundary at which a future live controller may touch a provider.""" + + def provision( + self, + invocation: AttemptInvocation, + authorization: SpendAuthorization, + *, + deadline: datetime, + ) -> ProvisionedAttempt: ... + + def upload_plan(self, *, rung: int, plan_path: Path, deadline: datetime) -> None: ... + + def execute_rung(self, *, rung: int, image_digest: str, deadline: datetime) -> int: ... + + def retrieve_result( + self, *, rung: int, destination: Path, deadline: datetime + ) -> None: ... + + def retrieve_success_artifacts( + self, + *, + rung: int, + names: Sequence[str], + destination: Path, + deadline: datetime, + ) -> None: ... + + def teardown(self, resources: Mapping[str, str]) -> Mapping[str, Any]: ... + + +class Planner(Protocol): + def __call__( + self, + *, + root: Path, + output_dir: Path, + commit: str, + maximum_scale: int, + provider_capacity: Mapping[str, Any] | None, + image_digest: str | None, + ) -> Mapping[str, Any]: ... + + +ResultValidator = Callable[[Path, int], Mapping[str, Any]] +BundleValidator = Callable[[Path, int, Mapping[str, Any]], None] +PrefixReader = Callable[..., list[Mapping[str, Any]]] + + +def _integer(value: Any) -> bool: + return isinstance(value, int) and not isinstance(value, bool) + + +def _timestamp(value: Any) -> datetime: + if not isinstance(value, str) or not value.endswith("Z"): + raise AttemptError("authorization_refused", "spend authorization expiry is invalid") + try: + parsed = datetime.fromisoformat(value.removesuffix("Z") + "+00:00") + except ValueError as error: + raise AttemptError( + "authorization_refused", "spend authorization expiry is invalid" + ) from error + if parsed.tzinfo is None or parsed.utcoffset() != timezone.utc.utcoffset(parsed): + raise AttemptError("authorization_refused", "spend authorization expiry is invalid") + return parsed + + +def parse_spend_authorization(value: str | bytes | Mapping[str, Any]) -> SpendAuthorization: + """Parse a closed authorization document without retaining its encoded form.""" + if isinstance(value, (str, bytes)): + try: + decoded = json.loads(value) + except (UnicodeDecodeError, json.JSONDecodeError) as error: + raise AttemptError( + "authorization_refused", "spend authorization is malformed" + ) from error + else: + decoded = dict(value) + expected = { + "schema", + "status", + "provider", + "commit", + "admitted_plan_sha256", + "image_digest", + "organization", + "region", + "machine_class", + "volume_gib", + "rung", + "maximum_scale", + "attempt_nonce", + "app", + "issued_at", + "expires_at", + "teardown_owner", + "maximum_machine_seconds", + "resource_limits", + "pricing", + "claim", + } + if not isinstance(decoded, dict) or set(decoded) != expected: + raise AttemptError("authorization_refused", "spend authorization shape is invalid") + _validate_schema( + "progressive-spend-authorization.json", decoded, "authorization_refused" + ) + pricing = decoded.get("pricing") + pricing_fields = { + "currency", + "machine_microusd_per_hour", + "volume_microusd_per_gib_hour", + "transfer_allowance_microusd", + "estimated_total_microusd", + "maximum_total_microusd", + } + if ( + decoded.get("schema") != AUTHORIZATION_SCHEMA + or decoded.get("status") != "authorized" + or decoded.get("provider") != "fly" + or not isinstance(decoded.get("commit"), str) + or COMMIT.fullmatch(decoded["commit"]) is None + or not isinstance(decoded.get("admitted_plan_sha256"), str) + or re.fullmatch(r"[0-9a-f]{64}", decoded["admitted_plan_sha256"]) is None + or not isinstance(decoded.get("image_digest"), str) + or IMAGE_DIGEST.fullmatch(decoded["image_digest"]) is None + or not isinstance(decoded.get("app"), str) + or SAFE_NAME.fullmatch(decoded["app"]) is None + or not isinstance(decoded.get("organization"), str) + or SAFE_NAME.fullmatch(decoded["organization"]) is None + or not isinstance(decoded.get("region"), str) + or SAFE_REGION.fullmatch(decoded["region"]) is None + or not isinstance(decoded.get("machine_class"), str) + or SAFE_NAME.fullmatch(decoded["machine_class"]) is None + or not _integer(decoded.get("volume_gib")) + or not 1 <= decoded["volume_gib"] <= 500 + or not _integer(decoded.get("maximum_scale")) + or decoded["maximum_scale"] not in PROVIDER_RUNGS + or decoded.get("rung") not in {f"S{scale}" for scale in PROVIDER_RUNGS} + or int(str(decoded["rung"])[1:]) > decoded["maximum_scale"] + or not isinstance(decoded.get("attempt_nonce"), str) + or ATTEMPT_ID.fullmatch(decoded["attempt_nonce"]) is None + or decoded.get("app") != f"gf-progressive-{decoded.get('attempt_nonce')}" + or not isinstance(decoded.get("teardown_owner"), str) + or SAFE_NAME.fullmatch(decoded["teardown_owner"]) is None + or not _integer(decoded.get("maximum_machine_seconds")) + or not 1 <= decoded["maximum_machine_seconds"] <= 18_000 + or decoded.get("resource_limits") + != {"apps": 1, "volumes": 1, "machines": 1, "image_builds": 0} + or not isinstance(pricing, dict) + or set(pricing) != pricing_fields + or pricing.get("currency") != "USD" + or any( + not _integer(pricing.get(name)) or not 0 <= pricing[name] <= 1_000_000_000_000 + for name in pricing_fields - {"currency"} + ) + or pricing["estimated_total_microusd"] < 1 + or pricing["maximum_total_microusd"] < 1 + or pricing["estimated_total_microusd"] > pricing["maximum_total_microusd"] + or decoded.get("claim") != "spend_authorization_only" + ): + raise AttemptError("authorization_refused", "spend authorization values are invalid") + issued_at = _timestamp(decoded["issued_at"]) + expires_at = _timestamp(decoded["expires_at"]) + if expires_at <= issued_at or expires_at - issued_at > timedelta(hours=5): + raise AttemptError("authorization_refused", "spend authorization lifetime is invalid") + seconds = decoded["maximum_machine_seconds"] + machine = (pricing["machine_microusd_per_hour"] * seconds + 3599) // 3600 + volume = ( + pricing["volume_microusd_per_gib_hour"] + * decoded["volume_gib"] + * seconds + + 3599 + ) // 3600 + conservative_total = machine + volume + pricing["transfer_allowance_microusd"] + if conservative_total > pricing["estimated_total_microusd"]: + raise AttemptError("authorization_refused", "spend authorization ceiling is insufficient") + canonical = json.dumps(decoded, sort_keys=True, separators=(",", ":")).encode("utf-8") + return SpendAuthorization( + schema=decoded["schema"], + status=decoded["status"], + provider=decoded["provider"], + commit=decoded["commit"], + admitted_plan_sha256=decoded["admitted_plan_sha256"], + image_digest=decoded["image_digest"], + organization=decoded["organization"], + region=decoded["region"], + machine_class=decoded["machine_class"], + volume_gib=decoded["volume_gib"], + rung=decoded["rung"], + maximum_scale=decoded["maximum_scale"], + attempt_nonce=decoded["attempt_nonce"], + app=decoded["app"], + issued_at=issued_at, + expires_at=expires_at, + teardown_owner=decoded["teardown_owner"], + maximum_machine_seconds=decoded["maximum_machine_seconds"], + resource_limits=dict(decoded["resource_limits"]), + pricing=dict(decoded["pricing"]), + claim=decoded["claim"], + authorization_sha256=hashlib.sha256(canonical).hexdigest(), + ) + + +def validate_authorization( + invocation: AttemptInvocation, + authorization: SpendAuthorization, + *, + now: datetime | None = None, +) -> None: + observed_now = now or datetime.now(timezone.utc) + if observed_now.tzinfo is None: + raise AttemptError("authorization_refused", "authorization clock is not timezone-aware") + if not COMMIT.fullmatch(invocation.commit) or invocation.commit != authorization.commit: + raise AttemptError("authorization_refused", "authorization commit mismatch") + if authorization.issued_at > observed_now or observed_now >= authorization.expires_at: + raise AttemptError("authorization_refused", "spend authorization has expired") + if ( + observed_now + timedelta(seconds=authorization.maximum_machine_seconds) + > authorization.expires_at + ): + raise AttemptError( + "authorization_refused", "spend authorization cannot cover its runtime ceiling" + ) + + +def _require_before_deadline(clock: Callable[[], datetime], deadline: datetime) -> None: + observed = clock() + if observed.tzinfo is None or observed >= deadline: + raise AttemptError("authorization_refused", "attempt execution deadline has expired") + + +def _atomic_json(path: Path, value: Mapping[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) + temporary = Path(temporary_name) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as stream: + json.dump(value, stream, indent=2, sort_keys=True) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + temporary.replace(path) + directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory) + finally: + os.close(directory) + finally: + temporary.unlink(missing_ok=True) + + +def _validate_schema(name: str, value: Mapping[str, Any], failure: str) -> None: + try: + schema = json.loads((SCHEMA_ROOT / name).read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error: + raise AttemptError(failure, "attempt schema is unavailable") from error + error = next(Draft202012Validator(schema).iter_errors(value), None) + if error is not None: + raise AttemptError(failure, "attempt document failed its schema") + + +def save_ledger(path: Path, ledger: AttemptLedger) -> None: + document = asdict(ledger) + _validate_schema("progressive-provider-attempt-ledger.json", document, "recovery_refused") + _atomic_json(path, document) + + +def load_ledger(path: Path) -> AttemptLedger: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error: + raise AttemptError( + "recovery_refused", "attempt ledger is unavailable or malformed" + ) from error + if isinstance(value, dict): + _validate_schema("progressive-provider-attempt-ledger.json", value, "recovery_refused") + if value.get("owner_app") != f"gf-progressive-{value.get('attempt_id')}": + raise AttemptError("recovery_refused", "attempt ledger ownership is inconsistent") + if not isinstance(value, dict) or value.pop("schema", None) != LEDGER_SCHEMA: + raise AttemptError("recovery_refused", "attempt ledger schema is invalid") + expected = {name for name in AttemptLedger.__dataclass_fields__ if name != "schema"} + if set(value) != expected: + raise AttemptError("recovery_refused", "attempt ledger shape is invalid") + try: + ledger = AttemptLedger(**value) + except TypeError as error: + raise AttemptError("recovery_refused", "attempt ledger is malformed") from error + if ( + not isinstance(ledger.completed_scales, list) + or any( + not _integer(scale) or scale not in CANONICAL_RUNGS for scale in ledger.completed_scales + ) + or not isinstance(ledger.resources, dict) + or any( + not isinstance(key, str) or not isinstance(item, str) + for key, item in ledger.resources.items() + ) + ): + raise AttemptError("recovery_refused", "attempt ledger values are invalid") + return ledger + + +def _prefix_scales(prefix: Sequence[Mapping[str, Any]]) -> list[int]: + scales = [item.get("scale") for item in prefix] + if any(not _integer(scale) for scale in scales): + raise AttemptError("prerequisite_refused", "completed rung prefix is malformed") + typed = [int(scale) for scale in scales] + if typed != list(CANONICAL_RUNGS[: len(typed)]): + raise AttemptError("prerequisite_refused", "completed rung prefix is not contiguous") + if typed[:2] != [18, 19]: + raise AttemptError("prerequisite_refused", "passed S18 and S19 evidence is required") + return typed + + +def validate_result(path: Path, rung: int) -> Mapping[str, Any]: + """Perform the result-first closed-shape check before retrieving large artifacts.""" + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error: + raise AttemptError( + "retrieval_failed", "provider rung result is unavailable or malformed" + ) from error + if ( + not isinstance(value, dict) + or value.get("schema") != RESULT_SCHEMA + or value.get("rung") != f"S{rung}" + or value.get("status") not in {"passed", "failed"} + or (value.get("status") == "passed" and value.get("failure") is not None) + or (value.get("status") == "failed" and not isinstance(value.get("failure"), str)) + ): + raise AttemptError("evidence_invalid", "provider rung result identity is invalid") + _validate_schema("progressive-provider-run-result.json", value, "evidence_invalid") + return value + + +def validate_staged_bundle(_stage: Path, _rung: int, _result: Mapping[str, Any]) -> None: + """Default hook; authoritative repository validation follows publication.""" + + +def _publish(path: Path, destination: Path) -> None: + if destination.exists() or destination.is_symlink(): + raise AttemptError("retrieval_failed", "attempt evidence already exists") + try: + os.link(path, destination) + directory = os.open(destination.parent, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory) + finally: + os.close(directory) + except OSError as error: + raise AttemptError("retrieval_failed", "attempt evidence publication failed") from error + + +def _artifact_names(rung: int) -> tuple[str, ...]: + return tuple(f"s{rung}-{suffix}.json" for suffix in ("plan", "benchexec", "graphforge", "rung")) + + +def _rollback_rung(evidence_dir: Path, rung: int) -> None: + for name in (*_artifact_names(rung), f"s{rung}-result.json"): + (evidence_dir / name).unlink(missing_ok=True) + directory = os.open(evidence_dir, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory) + finally: + os.close(directory) + + +def _try_rollback_rung(evidence_dir: Path, rung: int) -> str | None: + try: + _rollback_rung(evidence_dir, rung) + except Exception: + return "evidence_cleanup_failed" + return None + + +def _require_fresh_rung(evidence_dir: Path, rung: int) -> None: + if any( + (evidence_dir / name).exists() or (evidence_dir / name).is_symlink() + for name in (*_artifact_names(rung), f"s{rung}-result.json") + ): + raise AttemptError( + "source_mismatch", "incomplete provider evidence requires cleanup-only recovery" + ) + + +def _admitted_plan( + invocation: AttemptInvocation, + authorization: SpendAuthorization, + next_rung: int, + planner: Planner, +) -> Mapping[str, Any]: + try: + plan = planner( + root=invocation.root, + output_dir=invocation.evidence_dir, + commit=invocation.commit, + maximum_scale=authorization.maximum_scale, + provider_capacity=invocation.provider_capacity, + image_digest=authorization.image_digest, + ) + except (OSError, ProviderPlanError, ValueError) as error: + raise AttemptError("progression_refused", "next provider rung was not admitted") from error + if ( + not isinstance(plan, Mapping) + or plan.get("status") != "admitted" + or plan.get("execution_authorized") is not True + or plan.get("execution_refusal") is not None + or plan.get("next_rung") != f"S{next_rung}" + or plan.get("image_digest") != authorization.image_digest + ): + raise AttemptError("progression_refused", "planner violated sequential authority") + return plan + + +def _plan_digest(path: Path, plan: Mapping[str, Any]) -> str: + _atomic_json(path, plan) + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _cleanup_handles(ledger: AttemptLedger) -> dict[str, str]: + handles = dict(ledger.resources) + if ledger.owner_app is not None: + handles["owner_app"] = ledger.owner_app + return handles + + +def _teardown_observation(value: Mapping[str, Any]) -> dict[str, Any]: + expected = {"app_exists", "machines", "volumes", "secrets"} + if not isinstance(value, Mapping) or set(value) != expected: + raise AttemptError("inventory_unavailable", "teardown inventory is malformed") + if ( + type(value["app_exists"]) is not bool + or any(not _integer(value[name]) or value[name] < 0 for name in expected - {"app_exists"}) + ): + raise AttemptError("inventory_unavailable", "teardown inventory is malformed") + return dict(value) + + +def _outcome(ledger: AttemptLedger) -> AttemptOutcome: + failure = ledger.failure + if failure is None and ledger.cleanup_failure is not None: + failure = "cleanup_failed" + return AttemptOutcome( + schema=ATTEMPT_SCHEMA, + status="passed" if failure is None else "failed", + commit=str(ledger.commit), + authorized_maximum_scale=int(ledger.authorized_maximum_scale), + completed_scales=tuple(ledger.completed_scales), + first_failed_rung=ledger.first_failed_rung, + failure=failure, + cleanup_failure=ledger.cleanup_failure, + authorization_sha256=str(ledger.authorization_sha256), + admitted_plan_sha256=str(ledger.admitted_plan_sha256), + authorized_image_digest=str(ledger.authorized_image_digest), + observed_image_digest=ledger.image_digest, + teardown_status="failed" if ledger.cleanup_failure else "empty", + teardown_observed=ledger.teardown_observed, + teardown_checked_at=ledger.teardown_checked_at, + ) + + +def execute( + invocation: AttemptInvocation, + authorization: SpendAuthorization, + *, + transport: ProviderTransport, + planner: Planner = plan_provider_ladder, + prefix_reader: PrefixReader = completed_rungs, + result_validator: ResultValidator = validate_result, + bundle_validator: BundleValidator = validate_staged_bundle, + now: datetime | None = None, + clock: Callable[[], datetime] | None = None, +) -> AttemptOutcome: + """Execute a bounded provider attempt, one admitted rung at a time.""" + clock_fn = clock or (lambda: datetime.now(timezone.utc)) + started_at = now or clock_fn() + validate_authorization(invocation, authorization, now=started_at) + deadline = started_at + timedelta(seconds=authorization.maximum_machine_seconds) + if invocation.ledger_path.exists() or invocation.ledger_path.is_symlink(): + raise AttemptError( + "recovery_refused", "existing attempt ledger requires cleanup-only recovery" + ) + invocation.evidence_dir.mkdir(parents=True, exist_ok=True) + try: + prefix = prefix_reader( + invocation.root, + invocation.evidence_dir, + commit=invocation.commit, + ) + except (OSError, ProviderPlanError, ValueError) as error: + raise AttemptError("prerequisite_refused", "completed rung evidence is invalid") from error + scales = _prefix_scales(prefix) + if any(scale > authorization.maximum_scale for scale in scales): + raise AttemptError("authorization_refused", "completed evidence exceeds authorization") + if scales == list(CANONICAL_RUNGS) or scales[-1] >= authorization.maximum_scale: + raise AttemptError("progression_refused", "authorized provider ladder is already complete") + + next_rung = next(rung for rung in PROVIDER_RUNGS if rung not in scales) + _require_fresh_rung(invocation.evidence_dir, next_rung) + control_plan = invocation.ledger_path.with_name(f".{invocation.ledger_path.name}.plan.json") + first_plan = _admitted_plan(invocation, authorization, next_rung, planner) + first_plan_sha256 = _plan_digest(control_plan, first_plan) + if authorization.rung != f"S{next_rung}": + control_plan.unlink(missing_ok=True) + raise AttemptError("authorization_refused", "authorization selects a different first rung") + if authorization.admitted_plan_sha256 != first_plan_sha256: + control_plan.unlink(missing_ok=True) + raise AttemptError("authorization_refused", "admitted plan contradicts authorization") + + ledger = AttemptLedger( + generation=1, + attempt_id=authorization.attempt_nonce, + owner_app=authorization.app, + commit=invocation.commit, + authorization_sha256=authorization.authorization_sha256, + admitted_plan_sha256=authorization.admitted_plan_sha256, + authorized_maximum_scale=authorization.maximum_scale, + authorized_image_digest=authorization.image_digest, + expires_at=authorization.expires_at.strftime("%Y-%m-%dT%H:%M:%SZ"), + evidence_dir=str(invocation.evidence_dir.resolve()), + phase="authorized", + completed_scales=scales, + ) + save_ledger(invocation.ledger_path, ledger) + provisioned: ProvisionedAttempt | None = None + primary_error: AttemptError | None = None + pending_plan: Mapping[str, Any] | None = first_plan + try: + _require_before_deadline(clock_fn, deadline) + provisioned = transport.provision(invocation, authorization, deadline=deadline) + if IMAGE_DIGEST.fullmatch(provisioned.image_digest) is None: + raise AttemptError( + "machine_identity_mismatch", "provider-observed image digest is invalid" + ) + if provisioned.image_digest != authorization.image_digest: + raise AttemptError( + "machine_identity_mismatch", + "provider-observed image digest contradicts authorization", + ) + if ( + set(provisioned.resources) != {"machine_id", "volume_id"} + or not isinstance(provisioned.resources.get("machine_id"), str) + or MACHINE_ID.fullmatch(provisioned.resources["machine_id"]) is None + or not isinstance(provisioned.resources.get("volume_id"), str) + or VOLUME_ID.fullmatch(provisioned.resources["volume_id"]) is None + ): + raise AttemptError("provision_failed", "provider cleanup handles are malformed") + ledger.image_digest = provisioned.image_digest + ledger.resources = dict(provisioned.resources) + ledger.phase = "provisioned" + save_ledger(invocation.ledger_path, ledger) + + while ledger.completed_scales[-1] < authorization.maximum_scale: + _require_before_deadline(clock_fn, deadline) + next_rung = next(rung for rung in PROVIDER_RUNGS if rung not in ledger.completed_scales) + _require_fresh_rung(invocation.evidence_dir, next_rung) + ledger.current_rung = next_rung + plan = pending_plan or _admitted_plan(invocation, authorization, next_rung, planner) + pending_plan = None + ledger.phase = "planned" + save_ledger(invocation.ledger_path, ledger) + _atomic_json(control_plan, plan) + try: + transport.upload_plan( + rung=next_rung, plan_path=control_plan, deadline=deadline + ) + _require_before_deadline(clock_fn, deadline) + except AttemptError: + raise + except Exception as error: + raise AttemptError("upload_failed", "provider plan upload failed") from error + ledger.phase = "executing" + save_ledger(invocation.ledger_path, ledger) + try: + execution_status = transport.execute_rung( + rung=next_rung, + image_digest=provisioned.image_digest, + deadline=deadline, + ) + _require_before_deadline(clock_fn, deadline) + except AttemptError: + raise + except Exception as error: + raise AttemptError("rung_failed", "provider rung execution failed") from error + + with tempfile.TemporaryDirectory( + prefix=f".s{next_rung}-attempt-", dir=invocation.evidence_dir + ) as temporary: + stage = Path(temporary) + result_name = f"s{next_rung}-result.json" + staged_result = stage / result_name + try: + transport.retrieve_result( + rung=next_rung, destination=staged_result, deadline=deadline + ) + _require_before_deadline(clock_fn, deadline) + except AttemptError: + raise + except Exception as error: + raise AttemptError( + "retrieval_failed", "provider result retrieval failed" + ) from error + result = result_validator(staged_result, next_rung) + identities = result.get("identities") + if ( + not isinstance(identities, Mapping) + or identities.get("commit") != invocation.commit + or identities.get("image_digest") != provisioned.image_digest + ): + raise AttemptError( + "evidence_invalid", "provider rung result identity is invalid" + ) + if result["status"] == "failed": + _publish(staged_result, invocation.evidence_dir / result_name) + ledger.first_failed_rung = next_rung + ledger.failure = "rung_failed" + ledger.phase = "rung_failed" + save_ledger(invocation.ledger_path, ledger) + break + if execution_status != 0: + raise AttemptError( + "rung_failed", + "successful rung result contradicts execution status", + ) + names = _artifact_names(next_rung) + try: + transport.retrieve_success_artifacts( + rung=next_rung, + names=names, + destination=stage, + deadline=deadline, + ) + _require_before_deadline(clock_fn, deadline) + except AttemptError: + raise + except Exception as error: + raise AttemptError( + "retrieval_failed", "provider artifact retrieval failed" + ) from error + if any(not (stage / name).is_file() for name in names): + raise AttemptError("retrieval_failed", "provider rung artifact is unavailable") + bundle_validator(stage, next_rung, result) + try: + for name in (*names, result_name): + _publish(stage / name, invocation.evidence_dir / name) + except AttemptError as error: + if _try_rollback_rung(invocation.evidence_dir, next_rung) is not None: + raise AttemptError( + "evidence_invalid", "partial evidence cleanup failed" + ) from error + raise + _require_before_deadline(clock_fn, deadline) + + try: + accepted = prefix_reader( + invocation.root, + invocation.evidence_dir, + commit=invocation.commit, + ) + except (OSError, ProviderPlanError, ValueError) as error: + _try_rollback_rung(invocation.evidence_dir, next_rung) + raise AttemptError( + "evidence_invalid", "retrieved rung evidence was not accepted" + ) from error + accepted_scales = _prefix_scales(accepted) + if accepted_scales != [*ledger.completed_scales, next_rung]: + _try_rollback_rung(invocation.evidence_dir, next_rung) + raise AttemptError("evidence_invalid", "accepted prefix did not advance once") + ledger.completed_scales = accepted_scales + ledger.current_rung = None + ledger.phase = "rung_accepted" + save_ledger(invocation.ledger_path, ledger) + if ledger.failure is None: + ledger.phase = "completed" + save_ledger(invocation.ledger_path, ledger) + except AttemptError as error: + primary_error = error + ledger.failure = ledger.failure or error.failure + if error.failure in { + "upload_failed", + "rung_failed", + "retrieval_failed", + "evidence_invalid", + "progression_refused", + }: + ledger.first_failed_rung = ledger.current_rung + ledger.phase = "failed" + save_ledger(invocation.ledger_path, ledger) + except Exception as error: + primary_error = AttemptError("provision_failed", "attempt boundary failed") + ledger.failure = ledger.failure or primary_error.failure + ledger.phase = "failed" + save_ledger(invocation.ledger_path, ledger) + primary_error.__cause__ = error + finally: + control_plan.unlink(missing_ok=True) + if ledger.current_rung is not None and ledger.current_rung not in ledger.completed_scales: + ledger.cleanup_failure = _try_rollback_rung( + invocation.evidence_dir, ledger.current_rung + ) + ledger.phase = "teardown" + save_ledger(invocation.ledger_path, ledger) + try: + observed = transport.teardown(_cleanup_handles(ledger)) + ledger.teardown_observed = _teardown_observation(observed) + ledger.teardown_checked_at = clock_fn().astimezone(timezone.utc).strftime( + "%Y-%m-%dT%H:%M:%SZ" + ) + if ledger.teardown_observed != { + "app_exists": False, + "machines": 0, + "volumes": 0, + "secrets": 0, + }: + raise AttemptError("inventory_not_empty", "teardown inventory is not empty") + except AttemptError as error: + ledger.cleanup_failure = ledger.cleanup_failure or error.failure + ledger.phase = "cleanup_failed" + except Exception: + ledger.cleanup_failure = ledger.cleanup_failure or "teardown_failed" + ledger.phase = "cleanup_failed" + else: + ledger.resources.clear() + ledger.phase = "cleanup_failed" if ledger.cleanup_failure else "closed" + save_ledger(invocation.ledger_path, ledger) + + # Expected rung failures are returned as typed outcomes; unexpected controller + # failures are likewise preserved in the durable ledger without raw diagnostics. + _ = primary_error + return _outcome(ledger) + + +def _teardown_document(outcome: AttemptOutcome) -> dict[str, Any]: + return { + "schema": "graphforge-progressive-provider-teardown-inventory/1", + "status": outcome.teardown_status, + "failure": outcome.cleanup_failure, + "commit": outcome.commit, + "authorized_maximum_scale": outcome.authorized_maximum_scale, + "completed_scales": list(outcome.completed_scales), + "authorization_sha256": outcome.authorization_sha256, + "admitted_plan_sha256": outcome.admitted_plan_sha256, + "checked_at": outcome.teardown_checked_at, + "observed": dict(outcome.teardown_observed) if outcome.teardown_observed else None, + "claim": "control_plane_evidence_only", + } + + +def _outcome_document(outcome: AttemptOutcome, teardown_inventory_sha256: str) -> dict[str, Any]: + return { + "schema": outcome.schema, + "status": outcome.status, + "commit": outcome.commit, + "authorized_maximum_scale": outcome.authorized_maximum_scale, + "completed_scales": list(outcome.completed_scales), + "first_failed_rung": ( + f"S{outcome.first_failed_rung}" if outcome.first_failed_rung is not None else None + ), + "failure": outcome.failure, + "cleanup_failure": outcome.cleanup_failure, + "authorization_sha256": outcome.authorization_sha256, + "admitted_plan_sha256": outcome.admitted_plan_sha256, + "authorized_image_digest": outcome.authorized_image_digest, + "observed_image_digest": outcome.observed_image_digest, + "teardown_status": outcome.teardown_status, + "teardown_inventory_sha256": teardown_inventory_sha256, + "claim": "engineering_evidence_only", + } + + +def _write_outcome(result_path: Path, outcome: AttemptOutcome) -> dict[str, Any]: + inventory_path = result_path.with_name(f"{result_path.stem}-teardown-inventory.json") + inventory = _teardown_document(outcome) + _validate_schema( + "progressive-provider-teardown-inventory.json", inventory, "evidence_invalid" + ) + _atomic_json(inventory_path, inventory) + inventory_sha256 = hashlib.sha256(inventory_path.read_bytes()).hexdigest() + document = _outcome_document(outcome, inventory_sha256) + _validate_schema("progressive-provider-attempt-result.json", document, "evidence_invalid") + _atomic_json(result_path, document) + return document + + +def execute_attempt( + request: AttemptRequest, + *, + root: Path, + output_dir: Path, + ledger_path: Path, + result_path: Path, + boundary: ProviderTransport, + planner: Planner = plan_provider_ladder, + prefix_reader: PrefixReader = completed_rungs, + result_validator: ResultValidator = validate_result, + bundle_validator: BundleValidator = validate_staged_bundle, + now: datetime | None = None, + clock: Callable[[], datetime] | None = None, +) -> dict[str, Any]: + """Validate an operator request, run the core, and durably write its result.""" + inventory_path = result_path.with_name(f"{result_path.stem}-teardown-inventory.json") + if any(path.exists() or path.is_symlink() for path in (result_path, inventory_path)): + raise AttemptError("source_mismatch", "attempt result path already exists") + if request.spend_authorization is None: + raise AttemptError("authorization_refused", "spend authorization is required") + authorization = parse_spend_authorization(request.spend_authorization) + if ( + request.commit != authorization.commit + or request.organization != authorization.organization + or request.app != authorization.app + or request.region != authorization.region + or request.machine_class != authorization.machine_class + or request.volume_gib != authorization.volume_gib + or request.image_digest != authorization.image_digest + or request.maximum_scale != authorization.maximum_scale + ): + raise AttemptError("authorization_refused", "request contradicts authorization") + outcome = execute( + AttemptInvocation( + root=root, + evidence_dir=output_dir, + ledger_path=ledger_path, + commit=request.commit, + provider_capacity=request.provider_capacity, + ), + authorization, + transport=boundary, + planner=planner, + prefix_reader=prefix_reader, + result_validator=result_validator, + bundle_validator=bundle_validator, + now=now, + clock=clock, + ) + return _write_outcome(result_path, outcome) + + +def cleanup_only( + ledger_path: Path, + result_path: Path, + *, + transport: ProviderTransport, +) -> dict[str, Any]: + """Retry teardown from durable state without executing or planning a rung.""" + ledger = load_ledger(ledger_path) + evidence_dir = Path(str(ledger.evidence_dir)) + if ( + not evidence_dir.is_absolute() + or evidence_dir.is_symlink() + or not evidence_dir.is_dir() + or evidence_dir.resolve() != evidence_dir + ): + raise AttemptError("recovery_refused", "attempt evidence directory is unsafe") + ledger.cleanup_failure = None + if ledger.current_rung is not None and ledger.current_rung not in ledger.completed_scales: + ledger.cleanup_failure = _try_rollback_rung(evidence_dir, ledger.current_rung) + try: + observed = transport.teardown(_cleanup_handles(ledger)) + ledger.teardown_observed = _teardown_observation(observed) + ledger.teardown_checked_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + if ledger.teardown_observed != { + "app_exists": False, + "machines": 0, + "volumes": 0, + "secrets": 0, + }: + raise AttemptError("inventory_not_empty", "teardown inventory is not empty") + except AttemptError as error: + ledger.cleanup_failure = ledger.cleanup_failure or error.failure + ledger.phase = "cleanup_failed" + except Exception: + ledger.cleanup_failure = ledger.cleanup_failure or "teardown_failed" + ledger.phase = "cleanup_failed" + else: + ledger.resources.clear() + ledger.phase = "cleanup_failed" if ledger.cleanup_failure else "closed" + save_ledger(ledger_path, ledger) + return _write_outcome(result_path, _outcome(ledger)) diff --git a/benchmarks/schemas/progressive-provider-attempt-ledger.json b/benchmarks/schemas/progressive-provider-attempt-ledger.json new file mode 100644 index 000000000..83fa1101f --- /dev/null +++ b/benchmarks/schemas/progressive-provider-attempt-ledger.json @@ -0,0 +1,50 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "schema": "graphforge-progressive-provider-attempt-ledger-schema/1", + "type": "object", + "additionalProperties": false, + "required": ["schema", "generation", "attempt_id", "owner_app", "commit", "authorization_sha256", "admitted_plan_sha256", "authorized_maximum_scale", "authorized_image_digest", "expires_at", "evidence_dir", "phase", "image_digest", "current_rung", "completed_scales", "first_failed_rung", "failure", "cleanup_failure", "resources", "teardown_observed", "teardown_checked_at"], + "properties": { + "schema": {"const": "graphforge-progressive-provider-attempt-ledger/1"}, + "generation": {"type": "integer", "minimum": 1, "maximum": 1000}, + "attempt_id": {"$ref": "#/$defs/nonce"}, + "owner_app": {"type": "string", "pattern": "^gf-progressive-[0-9a-f]{32}$"}, + "commit": {"$ref": "#/$defs/commit"}, + "authorization_sha256": {"$ref": "#/$defs/digest"}, + "admitted_plan_sha256": {"$ref": "#/$defs/digest"}, + "authorized_maximum_scale": {"$ref": "#/$defs/scale"}, + "authorized_image_digest": {"$ref": "#/$defs/image"}, + "expires_at": {"$ref": "#/$defs/timestamp"}, + "evidence_dir": {"type": "string", "pattern": "^/[^\\u0000]+$"}, + "phase": {"enum": ["authorized", "provisioned", "planned", "executing", "rung_failed", "rung_accepted", "completed", "failed", "teardown", "cleanup_failed", "closed"]}, + "image_digest": {"oneOf": [{"type": "null"}, {"$ref": "#/$defs/image"}]}, + "current_rung": {"oneOf": [{"type": "null"}, {"$ref": "#/$defs/scale"}]}, + "completed_scales": {"$ref": "#/$defs/completedScales"}, + "first_failed_rung": {"oneOf": [{"type": "null"}, {"$ref": "#/$defs/scale"}]}, + "failure": {"enum": [null, "authorization_refused", "source_mismatch", "provision_failed", "machine_identity_mismatch", "upload_failed", "rung_failed", "retrieval_failed", "evidence_invalid", "progression_refused"]}, + "cleanup_failure": {"enum": [null, "evidence_cleanup_failed", "teardown_failed", "inventory_unavailable", "inventory_not_empty"]}, + "resources": { + "oneOf": [ + {"type": "object", "maxProperties": 0}, + {"type": "object", "additionalProperties": false, "required": ["machine_id", "volume_id"], "properties": {"machine_id": {"type": "string", "pattern": "^[0-9a-f]{14}$"}, "volume_id": {"type": "string", "pattern": "^vol_[a-z0-9]+$"}}} + ] + }, + "teardown_observed": {"oneOf": [{"type": "null"}, {"$ref": "#/$defs/inventory"}]}, + "teardown_checked_at": {"oneOf": [{"type": "null"}, {"$ref": "#/$defs/timestamp"}]} + }, + "allOf": [ + {"if": {"properties": {"phase": {"const": "authorized"}}, "required": ["phase"]}, "then": {"properties": {"image_digest": {"type": "null"}, "resources": {"maxProperties": 0}}}}, + {"if": {"properties": {"phase": {"const": "closed"}}, "required": ["phase"]}, "then": {"properties": {"cleanup_failure": {"type": "null"}, "resources": {"maxProperties": 0}, "teardown_observed": {"type": "object", "properties": {"app_exists": {"const": false}, "machines": {"const": 0}, "volumes": {"const": 0}, "secrets": {"const": 0}}}, "teardown_checked_at": {"$ref": "#/$defs/timestamp"}}}}, + {"if": {"properties": {"phase": {"const": "cleanup_failed"}}, "required": ["phase"]}, "then": {"properties": {"cleanup_failure": {"type": "string"}}}} + ], + "$defs": { + "commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"}, + "digest": {"type": "string", "pattern": "^[0-9a-f]{64}$"}, + "nonce": {"type": "string", "pattern": "^[0-9a-f]{32}$"}, + "image": {"type": "string", "pattern": "^registry\\.fly\\.io/[a-z0-9][a-z0-9._/-]*@sha256:[0-9a-f]{64}$"}, + "scale": {"enum": [20, 22, 24, 25, 26]}, + "timestamp": {"type": "string", "pattern": "^[0-9]{4}-(0[1-9]|1[0-2])-([0-2][0-9]|3[01])T([01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9]Z$"}, + "completedScales": {"oneOf": [{"const": [18, 19]}, {"const": [18, 19, 20]}, {"const": [18, 19, 20, 22]}, {"const": [18, 19, 20, 22, 24]}, {"const": [18, 19, 20, 22, 24, 25]}, {"const": [18, 19, 20, 22, 24, 25, 26]}]}, + "inventory": {"type": "object", "additionalProperties": false, "required": ["app_exists", "machines", "volumes", "secrets"], "properties": {"app_exists": {"type": "boolean"}, "machines": {"type": "integer", "minimum": 0}, "volumes": {"type": "integer", "minimum": 0}, "secrets": {"type": "integer", "minimum": 0}}} + } +} diff --git a/benchmarks/schemas/progressive-provider-attempt-result.json b/benchmarks/schemas/progressive-provider-attempt-result.json new file mode 100644 index 000000000..1cb650096 --- /dev/null +++ b/benchmarks/schemas/progressive-provider-attempt-result.json @@ -0,0 +1,58 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "schema": "graphforge-progressive-provider-attempt-result-schema/1", + "type": "object", + "additionalProperties": false, + "required": ["schema", "status", "failure", "cleanup_failure", "commit", "authorized_maximum_scale", "completed_scales", "first_failed_rung", "authorization_sha256", "admitted_plan_sha256", "authorized_image_digest", "observed_image_digest", "teardown_status", "teardown_inventory_sha256", "claim"], + "properties": { + "schema": {"const": "graphforge-progressive-provider-attempt-result/1"}, + "status": {"enum": ["passed", "failed"]}, + "failure": {"enum": [null, "authorization_refused", "source_mismatch", "readiness_timeout", "provision_failed", "machine_identity_mismatch", "upload_failed", "rung_failed", "retrieval_failed", "evidence_invalid", "progression_refused", "cleanup_failed"]}, + "cleanup_failure": {"enum": [null, "evidence_cleanup_failed", "teardown_failed", "inventory_unavailable", "inventory_not_empty"]}, + "commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"}, + "authorized_maximum_scale": {"enum": [20, 22, 24, 25, 26]}, + "completed_scales": {"$ref": "#/$defs/completedScales"}, + "first_failed_rung": {"oneOf": [{"type": "null"}, {"$ref": "#/$defs/rung"}]}, + "authorization_sha256": {"$ref": "#/$defs/digest"}, + "admitted_plan_sha256": {"$ref": "#/$defs/digest"}, + "authorized_image_digest": {"$ref": "#/$defs/image"}, + "observed_image_digest": {"oneOf": [{"type": "null"}, {"$ref": "#/$defs/image"}]}, + "teardown_status": {"enum": ["not_required", "empty", "failed"]}, + "teardown_inventory_sha256": {"oneOf": [{"type": "null"}, {"$ref": "#/$defs/digest"}]}, + "claim": {"const": "engineering_evidence_only"} + }, + "allOf": [ + { + "if": {"properties": {"status": {"const": "passed"}}, "required": ["status"]}, + "then": {"properties": {"failure": {"type": "null"}, "cleanup_failure": {"type": "null"}, "first_failed_rung": {"type": "null"}, "observed_image_digest": {"$ref": "#/$defs/image"}, "teardown_status": {"const": "empty"}, "teardown_inventory_sha256": {"$ref": "#/$defs/digest"}}}, + "else": {"properties": {"failure": {"type": "string"}}} + }, + { + "if": {"properties": {"failure": {"enum": ["upload_failed", "rung_failed", "retrieval_failed", "evidence_invalid", "progression_refused"]}}, "required": ["failure"]}, + "then": {"properties": {"first_failed_rung": {"$ref": "#/$defs/rung"}}}, + "else": {"properties": {"first_failed_rung": {"type": "null"}}} + }, + { + "if": {"properties": {"failure": {"const": "cleanup_failed"}}, "required": ["failure"]}, + "then": {"properties": {"cleanup_failure": {"type": "string"}, "teardown_status": {"const": "failed"}, "teardown_inventory_sha256": {"$ref": "#/$defs/digest"}}} + }, + { + "if": {"properties": {"teardown_status": {"const": "not_required"}}, "required": ["teardown_status"]}, + "then": {"properties": {"cleanup_failure": {"type": "null"}, "teardown_inventory_sha256": {"type": "null"}}} + }, + { + "if": {"properties": {"teardown_status": {"const": "empty"}}, "required": ["teardown_status"]}, + "then": {"properties": {"cleanup_failure": {"type": "null"}, "teardown_inventory_sha256": {"$ref": "#/$defs/digest"}}} + }, + { + "if": {"properties": {"teardown_status": {"const": "failed"}}, "required": ["teardown_status"]}, + "then": {"properties": {"cleanup_failure": {"type": "string"}, "teardown_inventory_sha256": {"$ref": "#/$defs/digest"}}} + } + ], + "$defs": { + "completedScales": {"oneOf": [{"const": [18, 19]}, {"const": [18, 19, 20]}, {"const": [18, 19, 20, 22]}, {"const": [18, 19, 20, 22, 24]}, {"const": [18, 19, 20, 22, 24, 25]}, {"const": [18, 19, 20, 22, 24, 25, 26]}]}, + "digest": {"type": "string", "pattern": "^[0-9a-f]{64}$"}, + "image": {"type": "string", "pattern": "^registry\\.fly\\.io/[a-z0-9][a-z0-9._/-]*@sha256:[0-9a-f]{64}$"}, + "rung": {"enum": ["S20", "S22", "S24", "S25", "S26"]} + } +} diff --git a/benchmarks/schemas/progressive-provider-teardown-inventory.json b/benchmarks/schemas/progressive-provider-teardown-inventory.json new file mode 100644 index 000000000..fd2cd3370 --- /dev/null +++ b/benchmarks/schemas/progressive-provider-teardown-inventory.json @@ -0,0 +1,85 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "schema": "graphforge-progressive-provider-teardown-inventory-schema/1", + "type": "object", + "additionalProperties": false, + "required": ["schema", "status", "failure", "commit", "authorized_maximum_scale", "completed_scales", "authorization_sha256", "admitted_plan_sha256", "checked_at", "observed", "claim"], + "properties": { + "schema": {"const": "graphforge-progressive-provider-teardown-inventory/1"}, + "status": {"enum": ["empty", "not_required", "failed"]}, + "failure": {"enum": [null, "evidence_cleanup_failed", "inventory_unavailable", "inventory_not_empty", "teardown_failed"]}, + "commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"}, + "authorized_maximum_scale": {"enum": [20, 22, 24, 25, 26]}, + "completed_scales": {"$ref": "#/$defs/completedScales"}, + "authorization_sha256": {"$ref": "#/$defs/digest"}, + "admitted_plan_sha256": {"$ref": "#/$defs/digest"}, + "checked_at": {"oneOf": [{"type": "null"}, {"$ref": "#/$defs/timestamp"}]}, + "observed": { + "oneOf": [ + {"type": "null"}, + { + "type": "object", + "additionalProperties": false, + "required": ["app_exists", "machines", "volumes", "secrets"], + "properties": { + "app_exists": {"type": "boolean"}, + "machines": {"type": "integer", "minimum": 0}, + "volumes": {"type": "integer", "minimum": 0}, + "secrets": {"type": "integer", "minimum": 0} + } + } + ] + }, + "claim": {"const": "control_plane_evidence_only"} + }, + "allOf": [ + { + "if": {"properties": {"status": {"const": "empty"}}, "required": ["status"]}, + "then": { + "properties": { + "failure": {"type": "null"}, + "checked_at": {"$ref": "#/$defs/timestamp"}, + "observed": { + "type": "object", + "properties": { + "app_exists": {"const": false}, + "machines": {"const": 0}, + "volumes": {"const": 0}, + "secrets": {"const": 0} + } + } + } + } + }, + { + "if": {"properties": {"status": {"const": "not_required"}}, "required": ["status"]}, + "then": {"properties": {"failure": {"type": "null"}, "checked_at": {"type": "null"}, "observed": {"type": "null"}}} + }, + { + "if": {"properties": {"status": {"const": "failed"}}, "required": ["status"]}, + "then": {"properties": {"failure": {"type": "string"}}} + }, + { + "if": {"properties": {"failure": {"const": "inventory_unavailable"}}, "required": ["failure"]}, + "then": {"properties": {"checked_at": {"type": "null"}, "observed": {"type": "null"}}} + }, + { + "if": {"properties": {"failure": {"const": "inventory_not_empty"}}, "required": ["failure"]}, + "then": {"properties": {"checked_at": {"$ref": "#/$defs/timestamp"}, "observed": {"type": "object"}}} + } + ], + "$defs": { + "completedScales": { + "oneOf": [ + {"const": [18, 19]}, + {"const": [18, 19, 20]}, + {"const": [18, 19, 20, 22]}, + {"const": [18, 19, 20, 22, 24]}, + {"const": [18, 19, 20, 22, 24, 25]}, + {"const": [18, 19, 20, 22, 24, 25, 26]} + ] + }, + "digest": {"type": "string", "pattern": "^[0-9a-f]{64}$"}, + "timestamp": {"type": "string", "pattern": "^[0-9]{4}-(0[1-9]|1[0-2])-([0-2][0-9]|3[01])T([01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9]Z$"} + } +} diff --git a/benchmarks/schemas/progressive-spend-authorization.json b/benchmarks/schemas/progressive-spend-authorization.json new file mode 100644 index 000000000..9453bc609 --- /dev/null +++ b/benchmarks/schemas/progressive-spend-authorization.json @@ -0,0 +1,63 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "schema": "graphforge-progressive-spend-authorization-schema/1", + "type": "object", + "additionalProperties": false, + "required": ["schema", "status", "provider", "commit", "admitted_plan_sha256", "image_digest", "organization", "region", "machine_class", "volume_gib", "rung", "maximum_scale", "attempt_nonce", "app", "issued_at", "expires_at", "teardown_owner", "maximum_machine_seconds", "resource_limits", "pricing", "claim"], + "properties": { + "schema": {"const": "graphforge-progressive-spend-authorization/1"}, + "status": {"const": "authorized"}, + "provider": {"const": "fly"}, + "commit": {"$ref": "#/$defs/commit"}, + "admitted_plan_sha256": {"$ref": "#/$defs/digest"}, + "image_digest": {"$ref": "#/$defs/image"}, + "organization": {"$ref": "#/$defs/name"}, + "region": {"type": "string", "pattern": "^[a-z]{3}$"}, + "machine_class": {"$ref": "#/$defs/machineClass"}, + "volume_gib": {"type": "integer", "minimum": 1, "maximum": 500}, + "rung": {"$ref": "#/$defs/rung"}, + "maximum_scale": {"enum": [20, 22, 24, 25, 26]}, + "attempt_nonce": {"$ref": "#/$defs/nonce"}, + "app": {"type": "string", "pattern": "^gf-progressive-[0-9a-f]{32}$"}, + "issued_at": {"$ref": "#/$defs/timestamp"}, + "expires_at": {"$ref": "#/$defs/timestamp"}, + "teardown_owner": {"$ref": "#/$defs/name"}, + "maximum_machine_seconds": {"type": "integer", "minimum": 1, "maximum": 18000}, + "resource_limits": { + "type": "object", + "additionalProperties": false, + "required": ["apps", "volumes", "machines", "image_builds"], + "properties": { + "apps": {"const": 1}, + "volumes": {"const": 1}, + "machines": {"const": 1}, + "image_builds": {"const": 0} + } + }, + "pricing": { + "type": "object", + "additionalProperties": false, + "required": ["currency", "machine_microusd_per_hour", "volume_microusd_per_gib_hour", "transfer_allowance_microusd", "estimated_total_microusd", "maximum_total_microusd"], + "properties": { + "currency": {"const": "USD"}, + "machine_microusd_per_hour": {"type": "integer", "minimum": 1, "maximum": 1000000000000}, + "volume_microusd_per_gib_hour": {"type": "integer", "minimum": 1, "maximum": 1000000000000}, + "transfer_allowance_microusd": {"$ref": "#/$defs/money"}, + "estimated_total_microusd": {"type": "integer", "minimum": 1, "maximum": 1000000000000}, + "maximum_total_microusd": {"type": "integer", "minimum": 1, "maximum": 1000000000000} + } + }, + "claim": {"const": "spend_authorization_only"} + }, + "$defs": { + "commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"}, + "digest": {"type": "string", "pattern": "^[0-9a-f]{64}$"}, + "image": {"type": "string", "pattern": "^registry\\.fly\\.io/[a-z0-9][a-z0-9._/-]*@sha256:[0-9a-f]{64}$"}, + "machineClass": {"enum": ["shared-cpu-1x", "shared-cpu-2x", "shared-cpu-4x", "shared-cpu-6x", "shared-cpu-8x", "performance-1x", "performance-2x", "performance-4x", "performance-6x", "performance-8x", "performance-10x", "performance-12x", "performance-14x", "performance-16x"]}, + "money": {"type": "integer", "minimum": 0, "maximum": 1000000000000}, + "name": {"type": "string", "pattern": "^[a-z0-9][a-z0-9-]{0,62}$"}, + "nonce": {"type": "string", "pattern": "^[0-9a-f]{32}$"}, + "rung": {"enum": ["S20", "S22", "S24", "S25", "S26"]}, + "timestamp": {"type": "string", "pattern": "^[0-9]{4}-(0[1-9]|1[0-2])-([0-2][0-9]|3[01])T([01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9]Z$"} + } +} diff --git a/benchmarks/tests/test_progressive_provider_attempt.py b/benchmarks/tests/test_progressive_provider_attempt.py new file mode 100644 index 000000000..a46e7ea12 --- /dev/null +++ b/benchmarks/tests/test_progressive_provider_attempt.py @@ -0,0 +1,691 @@ +from __future__ import annotations + +from dataclasses import asdict +from datetime import datetime, timezone +import hashlib +import json +from pathlib import Path +import shutil +import subprocess +import tempfile +import unittest +from unittest.mock import patch + +from graphforge_bench.progressive_provider_attempt import ( + AttemptError, + AttemptInvocation, + AttemptRequest, + ProvisionedAttempt, + SpendAuthorization, + _publish, + cleanup_only, + execute, + execute_attempt, + load_ledger, + parse_spend_authorization, +) +from jsonschema import Draft202012Validator +from tests.test_progressive_provider_plan import result as local_result +from tests.test_progressive_provider_plan import rung as rung_evidence +from tests.test_progressive_run import benchexec as benchexec_evidence +from tests.test_progressive_run import graphforge as graphforge_evidence +from tests.test_progressive_run import passed_rung as provider_rung_evidence + +ROOT = Path(__file__).resolve().parents[1] +COMMIT = subprocess.run( + ["git", "-C", str(ROOT.parent), "rev-parse", "HEAD"], + capture_output=True, + check=True, + text=True, +).stdout.strip() +NONCE = "c" * 32 +APP = "gf-progressive-" + NONCE +IMAGE = f"registry.fly.io/{APP}@sha256:" + "1" * 64 +NOW = datetime(2026, 6, 1, tzinfo=timezone.utc) + + +def first_plan() -> dict[str, object]: + return { + "status": "admitted", + "execution_authorized": True, + "execution_refusal": None, + "next_rung": "S20", + "image_digest": IMAGE, + } + + +def authorization_document(maximum_scale: int = 26) -> dict[str, object]: + """Exact benchmarks/schemas/progressive-spend-authorization.json fixture.""" + return { + "schema": "graphforge-progressive-spend-authorization/1", + "status": "authorized", + "provider": "fly", + "commit": COMMIT, + "admitted_plan_sha256": hashlib.sha256( + (json.dumps(first_plan(), indent=2, sort_keys=True) + "\n").encode("utf-8") + ).hexdigest(), + "image_digest": IMAGE, + "organization": "fixture-org", + "region": "dfw", + "machine_class": "performance-4x", + "volume_gib": 500, + "rung": "S20", + "maximum_scale": maximum_scale, + "attempt_nonce": NONCE, + "app": APP, + "issued_at": "2026-06-01T00:00:00Z", + "expires_at": "2026-06-01T05:00:00Z", + "teardown_owner": "qualification-operator", + "maximum_machine_seconds": 18_000, + "resource_limits": {"apps": 1, "volumes": 1, "machines": 1, "image_builds": 0}, + "pricing": { + "currency": "USD", + "machine_microusd_per_hour": 1, + "volume_microusd_per_gib_hour": 1, + "transfer_allowance_microusd": 1, + "estimated_total_microusd": 2506, + "maximum_total_microusd": 3000, + }, + "claim": "spend_authorization_only", + } + + +def authorization(maximum_scale: int = 26) -> SpendAuthorization: + return parse_spend_authorization(authorization_document(maximum_scale)) + + +def write_provider_bundle(directory: Path, scale: int) -> None: + """Write the real five-document provider result contract; never synthesize a rung.""" + directory.mkdir(parents=True, exist_ok=True) + profile = ROOT / "profiles" / "graph500" / f"s{scale}-provider.json" + identities = { + "commit": COMMIT, + "profile_id": f"graph500-s{scale}-provider", + "profile_sha256": hashlib.sha256(profile.read_bytes()).hexdigest(), + "image_digest": IMAGE, + "generator": "sha256:" + "0" * 64, + "generator_executable_sha256": "0" * 64, + "gf_sha256": "0" * 64, + "certify_sha256": "0" * 64, + "benchexec_python_sha256": "0" * 64, + "benchexec_version": "1.0", + "admitted_plan_sha256": "0" * 64, + "source_tree_sha256": "0" * 64, + } + paths = { + kind: directory / f"s{scale}-{kind}.json" + for kind in ("plan", "benchexec", "graphforge", "rung", "result") + } + plan = { + "schema": "graphforge-progressive-provider-execution-plan/1", + "rung": f"S{scale}", + "execution": "provider_native_linux_benchexec", + "identities": identities, + "limits": {"wall_seconds": 14_400, "memory_bytes": 4_294_967_296, "cores": 16}, + "outputs": [path.name for path in paths.values()], + "claim": "engineering_evidence_only", + } + graphforge = graphforge_evidence(scale) + graphforge["profile_id"] = f"graph500-s{scale}-provider" + rung = provider_rung_evidence(scale) + rung.update(profile_id=f"graph500-s{scale}-provider", source="canonical_ladder") + documents = { + "plan": plan, + "benchexec": benchexec_evidence(graphforge), + "graphforge": graphforge, + "rung": rung, + } + for kind, document in documents.items(): + paths[kind].write_text(json.dumps(document), encoding="utf-8") + result = { + "schema": "graphforge-progressive-provider-run-result/1", + "rung": f"S{scale}", + "status": "passed", + "failure": None, + "identities": identities, + "artifacts": { + f"{kind}_sha256": hashlib.sha256(paths[kind].read_bytes()).hexdigest() + for kind in ("plan", "benchexec", "graphforge", "rung") + }, + "claim": "engineering_evidence_only", + } + paths["result"].write_text(json.dumps(result), encoding="utf-8") + + +def planner(**values: object) -> dict[str, object]: + output = Path(values["output_dir"]) # type: ignore[arg-type] + present = {int(path.name[1:].split("-", 1)[0]) for path in output.glob("s*-rung.json")} + scale = next(item for item in (20, 22, 24, 25, 26) if item not in present) + return { + "status": "admitted", + "execution_authorized": True, + "execution_refusal": None, + "next_rung": f"S{scale}", + "image_digest": values["image_digest"], + } + + +class FakeTransport: + def __init__( + self, + remote: Path, + *, + observed_image: str = IMAGE, + fail_rung: int | None = None, + omit_rung: int | None = None, + corrupt_rung: int | None = None, + fail_provision: bool = False, + fail_teardown: bool = False, + teardown_inventory: dict[str, object] | None = None, + resources: dict[str, str] | None = None, + diagnostic: str = "fixture failure", + ) -> None: + self.remote = remote + self.observed_image = observed_image + self.fail_rung = fail_rung + self.omit_rung = omit_rung + self.corrupt_rung = corrupt_rung + self.fail_provision = fail_provision + self.fail_teardown = fail_teardown + self.teardown_inventory = teardown_inventory or { + "app_exists": False, + "machines": 0, + "volumes": 0, + "secrets": 0, + } + self.resources = resources or { + "machine_id": "abcdef01234567", + "volume_id": "vol_fixture123", + } + self.diagnostic = diagnostic + self.calls: list[tuple[object, ...]] = [] + + def provision( + self, + _invocation: AttemptInvocation, + authorization: SpendAuthorization, + *, + deadline: datetime, + ) -> ProvisionedAttempt: + self.calls.append(("provision", authorization.app, deadline.isoformat())) + if self.fail_provision: + raise OSError(self.diagnostic) + return ProvisionedAttempt( + image_digest=self.observed_image, + resources=self.resources, + ) + + def upload_plan(self, *, rung: int, plan_path: Path, deadline: datetime) -> None: + self.calls.append(("upload_plan", rung)) + if not plan_path.is_file(): + raise AssertionError("admitted plan was not persisted") + + def execute_rung(self, *, rung: int, image_digest: str, deadline: datetime) -> int: + self.calls.append(("execute_rung", rung, image_digest)) + write_provider_bundle(self.remote, rung) + if rung == self.fail_rung: + path = self.remote / f"s{rung}-result.json" + value = json.loads(path.read_text(encoding="utf-8")) + value.update(status="failed", failure="benchexec_failed", artifacts=None) + path.write_text(json.dumps(value), encoding="utf-8") + return 1 + if rung == self.omit_rung: + (self.remote / f"s{rung}-graphforge.json").unlink() + if rung == self.corrupt_rung: + with (self.remote / f"s{rung}-benchexec.json").open("a", encoding="utf-8") as stream: + stream.write("\n") + return 0 + + def retrieve_result( + self, *, rung: int, destination: Path, deadline: datetime + ) -> None: + self.calls.append(("retrieve_result", rung)) + shutil.copyfile(self.remote / f"s{rung}-result.json", destination) + + def retrieve_success_artifacts( + self, + *, + rung: int, + names: tuple[str, ...], + destination: Path, + deadline: datetime, + ) -> None: + self.calls.append(("retrieve_success_artifacts", rung)) + for name in names: + source = self.remote / name + if source.is_file(): + shutil.copyfile(source, destination / name) + + def teardown(self, resources: dict[str, str]) -> dict[str, object]: + self.calls.append(("teardown", tuple(sorted(resources)))) + if self.fail_teardown: + raise OSError(self.diagnostic) + return self.teardown_inventory + + +class ProgressiveProviderAttemptTests(unittest.TestCase): + def setUp(self) -> None: + self.temporary = tempfile.TemporaryDirectory() + self.base = Path(self.temporary.name) + self.output = self.base / "evidence" + self.remote = self.base / "remote" + self.remote.mkdir() + self.ledger = self.base / "attempt-ledger.json" + self.invocation = AttemptInvocation(ROOT, self.output, self.ledger, COMMIT) + + def tearDown(self) -> None: + self.temporary.cleanup() + + def write_prefix(self, output: Path, *scales: int) -> None: + output.mkdir(parents=True, exist_ok=True) + for scale in scales: + (output / f"s{scale}-rung.json").write_text( + json.dumps(rung_evidence(scale)), encoding="utf-8" + ) + (output / f"s{scale}-result.json").write_text( + json.dumps(local_result(scale)), encoding="utf-8" + ) + + def execute_attempt( + self, auth: SpendAuthorization, transport: FakeTransport + ) -> dict[str, object]: + return asdict( + execute( + self.invocation, + auth, + transport=transport, + planner=planner, + now=NOW, + clock=lambda: NOW, + ) + ) + + def test_spend_refusal_precedes_every_mutation(self) -> None: + invalid = authorization_document() + invalid["status"] = "refused" + transport = FakeTransport(self.remote) + with self.assertRaises(AttemptError) as raised: + parse_spend_authorization(invalid) + self.assertEqual(raised.exception.failure, "authorization_refused") + expired = authorization() + with self.assertRaisesRegex(AttemptError, "expired"): + execute( + self.invocation, + expired, + transport=transport, + planner=planner, + now=datetime(2028, 1, 1, tzinfo=timezone.utc), + ) + self.assertEqual(transport.calls, []) + self.assertFalse(self.ledger.exists()) + + def test_first_plan_hash_is_bound_before_provisioning(self) -> None: + self.write_prefix(self.output, 18, 19) + document = authorization_document() + document["admitted_plan_sha256"] = "f" * 64 + transport = FakeTransport(self.remote) + with self.assertRaisesRegex(AttemptError, "admitted plan"): + execute( + self.invocation, + parse_spend_authorization(document), + transport=transport, + planner=planner, + now=NOW, + clock=lambda: NOW, + ) + self.assertEqual(transport.calls, []) + self.assertFalse(self.ledger.exists()) + + def test_spend_lifetime_and_integer_ceiling_are_closed(self) -> None: + too_long = authorization_document() + too_long["expires_at"] = "2026-06-02T00:00:00Z" + with self.assertRaisesRegex(AttemptError, "lifetime"): + parse_spend_authorization(too_long) + bool_money = authorization_document() + bool_money["pricing"] = {**bool_money["pricing"], "maximum_total_microusd": True} # type: ignore[dict-item] + with self.assertRaises(AttemptError) as raised: + parse_spend_authorization(bool_money) + self.assertEqual(raised.exception.failure, "authorization_refused") + zero_rate = authorization_document() + zero_rate["pricing"] = { # type: ignore[assignment] + **zero_rate["pricing"], # type: ignore[dict-item] + "machine_microusd_per_hour": 0, + } + with self.assertRaises(AttemptError) as raised: + parse_spend_authorization(zero_rate) + self.assertEqual(raised.exception.failure, "authorization_refused") + + def test_execution_deadline_is_rechecked_between_rungs(self) -> None: + self.write_prefix(self.output, 18, 19) + deadline = datetime(2026, 6, 1, 5, tzinfo=timezone.utc) + observations = iter((NOW,) * 8 + (deadline,) * 2) + transport = FakeTransport(self.remote) + outcome = asdict( + execute( + self.invocation, + authorization(), + transport=transport, + planner=planner, + clock=lambda: next(observations), + ) + ) + executed = [call[1] for call in transport.calls if call[0] == "execute_rung"] + self.assertEqual(executed, [20]) + self.assertEqual(outcome["failure"], "authorization_refused") + + def test_deadline_crossing_during_provider_operations_cannot_pass(self) -> None: + deadline = datetime(2026, 6, 1, 5, tzinfo=timezone.utc) + for stage, live_observations, expected_call in ( + ("upload", 3, "upload_plan"), + ("result", 5, "retrieve_result"), + ("final", 7, "retrieve_success_artifacts"), + ): + with self.subTest(stage=stage), tempfile.TemporaryDirectory() as directory: + base = Path(directory) + output, remote = base / "evidence", base / "remote" + remote.mkdir() + self.write_prefix(output, 18, 19) + observations = iter( + (NOW,) * live_observations + (deadline,) * 3 + ) + transport = FakeTransport(remote) + outcome = asdict( + execute( + AttemptInvocation(ROOT, output, base / "ledger.json", COMMIT), + authorization(20), + transport=transport, + planner=planner, + clock=lambda observations=observations: next(observations), + ) + ) + self.assertTrue(any(call[0] == expected_call for call in transport.calls)) + self.assertEqual(outcome["failure"], "authorization_refused") + self.assertEqual(transport.calls[-1][0], "teardown") + self.assertFalse(any(output.glob("s20-*.json"))) + + def test_s18_and_s19_are_required_before_mutation(self) -> None: + for prefix in ((), (18,)): + with self.subTest(prefix=prefix), tempfile.TemporaryDirectory() as directory: + output = Path(directory) / "evidence" + self.write_prefix(output, *prefix) + invocation = AttemptInvocation( + ROOT, output, Path(directory) / "ledger.json", COMMIT + ) + transport = FakeTransport(Path(directory) / "remote") + transport.remote.mkdir() + with self.assertRaisesRegex(AttemptError, "S18 and S19"): + execute( + invocation, + authorization(), + transport=transport, + planner=planner, + now=NOW, + clock=lambda: NOW, + ) + self.assertEqual(transport.calls, []) + + def test_order_maximum_and_first_failure_stop(self) -> None: + self.write_prefix(self.output, 18, 19) + transport = FakeTransport(self.remote, fail_rung=24) + outcome = self.execute_attempt(authorization(25), transport) + executed = [call[1] for call in transport.calls if call[0] == "execute_rung"] + self.assertEqual(executed, [20, 22, 24]) + self.assertEqual(outcome["completed_scales"], (18, 19, 20, 22)) + self.assertEqual(outcome["first_failed_rung"], 24) + self.assertEqual(transport.calls[-1][0], "teardown") + + def test_maximum_scale_is_a_hard_stop(self) -> None: + self.write_prefix(self.output, 18, 19) + transport = FakeTransport(self.remote) + outcome = self.execute_attempt(authorization(22), transport) + executed = [call[1] for call in transport.calls if call[0] == "execute_rung"] + self.assertEqual(executed, [20, 22]) + self.assertEqual(outcome["status"], "passed") + + def test_missing_or_tampered_bundle_cannot_advance(self) -> None: + for mutation in ("missing", "tampered"): + with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as directory: + base = Path(directory) + output, remote = base / "evidence", base / "remote" + remote.mkdir() + self.write_prefix(output, 18, 19) + invocation = AttemptInvocation(ROOT, output, base / "ledger.json", COMMIT) + transport = FakeTransport( + remote, + omit_rung=20 if mutation == "missing" else None, + corrupt_rung=20 if mutation == "tampered" else None, + ) + outcome = asdict( + execute( + invocation, + authorization(), + transport=transport, + planner=planner, + now=NOW, + clock=lambda: NOW, + ) + ) + executed = [call[1] for call in transport.calls if call[0] == "execute_rung"] + self.assertEqual(executed, [20]) + self.assertEqual(outcome["status"], "failed") + self.assertNotIn(20, outcome["completed_scales"]) + self.assertEqual(transport.calls[-1][0], "teardown") + + def test_partial_publication_is_rolled_back(self) -> None: + self.write_prefix(self.output, 18, 19) + calls = 0 + + def fail_third(source: Path, destination: Path) -> None: + nonlocal calls + calls += 1 + if calls == 3: + raise AttemptError("retrieval_failed", "injected publication failure") + _publish(source, destination) + + with patch( + "graphforge_bench.progressive_provider_attempt._publish", + side_effect=fail_third, + ): + outcome = self.execute_attempt(authorization(20), FakeTransport(self.remote)) + self.assertEqual(outcome["failure"], "retrieval_failed") + self.assertFalse(any(self.output.glob("s20-*.json"))) + + def test_rollback_io_failure_never_skips_teardown(self) -> None: + self.write_prefix(self.output, 18, 19) + transport = FakeTransport(self.remote, fail_rung=20) + with patch( + "graphforge_bench.progressive_provider_attempt._rollback_rung", + side_effect=OSError("injected rollback failure"), + ): + outcome = self.execute_attempt(authorization(20), transport) + self.assertEqual(transport.calls[-1][0], "teardown") + self.assertEqual(outcome["cleanup_failure"], "evidence_cleanup_failed") + self.assertEqual(load_ledger(self.ledger).phase, "cleanup_failed") + + def test_observed_image_mismatch_blocks_every_rung(self) -> None: + self.write_prefix(self.output, 18, 19) + wrong = f"registry.fly.io/{APP}@sha256:" + "2" * 64 + transport = FakeTransport(self.remote, observed_image=wrong) + outcome = self.execute_attempt(authorization(), transport) + self.assertEqual(outcome["failure"], "machine_identity_mismatch") + self.assertFalse(any(call[0] == "execute_rung" for call in transport.calls)) + self.assertEqual(transport.calls[-1][0], "teardown") + + def test_malformed_provider_ids_still_teardown_owned_app(self) -> None: + self.write_prefix(self.output, 18, 19) + transport = FakeTransport( + self.remote, + resources={"machine_id": "bad", "volume_id": "also-bad"}, + ) + outcome = self.execute_attempt(authorization(), transport) + self.assertEqual(outcome["failure"], "provision_failed") + self.assertEqual(sum(call[0] == "teardown" for call in transport.calls), 1) + self.assertIn("owner_app", transport.calls[-1][1]) + + def test_complete_prefix_refuses_without_transport(self) -> None: + transport = FakeTransport(self.remote) + prefix = [rung_evidence(scale) for scale in (18, 19, 20, 22, 24, 25, 26)] + with self.assertRaisesRegex(AttemptError, "already complete"): + execute( + self.invocation, + authorization(), + transport=transport, + planner=planner, + prefix_reader=lambda *_args, **_kwargs: prefix, + now=NOW, + clock=lambda: NOW, + ) + self.assertEqual(transport.calls, []) + + def test_teardown_always_runs_and_incomplete_cleanup_keeps_ledger(self) -> None: + self.write_prefix(self.output, 18, 19) + for fail_provision, fail_rung in ((True, None), (False, 20)): + with tempfile.TemporaryDirectory() as directory: + base = Path(directory) + output, remote = base / "evidence", base / "remote" + remote.mkdir() + self.write_prefix(output, 18, 19) + transport = FakeTransport( + remote, fail_provision=fail_provision, fail_rung=fail_rung + ) + outcome = asdict( + execute( + AttemptInvocation(ROOT, output, base / "ledger.json", COMMIT), + authorization(), + transport=transport, + planner=planner, + now=NOW, + clock=lambda: NOW, + ) + ) + self.assertEqual(sum(call[0] == "teardown" for call in transport.calls), 1) + self.assertEqual(outcome["status"], "failed") + + transport = FakeTransport(self.remote, fail_rung=20, fail_teardown=True) + outcome = self.execute_attempt(authorization(), transport) + self.assertEqual(outcome["cleanup_failure"], "teardown_failed") + persisted = load_ledger(self.ledger) + self.assertEqual(persisted.phase, "cleanup_failed") + self.assertEqual(persisted.resources["machine_id"], "abcdef01234567") + + recovery = FakeTransport(self.remote) + result_path = self.base / "recovery-result.json" + (self.output / "s20-plan.json").write_text("partial") + (self.output / "s20-result.json").write_text("partial") + recovered = cleanup_only(self.ledger, result_path, transport=recovery) + self.assertEqual(recovered["teardown_status"], "empty") + self.assertFalse(any(self.output.glob("s20-*.json"))) + self.assertEqual(load_ledger(self.ledger).phase, "closed") + repeated = cleanup_only(self.ledger, result_path, transport=recovery) + self.assertEqual(repeated["teardown_status"], "empty") + + def test_cleanup_only_tears_down_when_evidence_rollback_fails(self) -> None: + self.write_prefix(self.output, 18, 19) + execute( + self.invocation, + authorization(20), + transport=FakeTransport(self.remote, fail_rung=20, fail_teardown=True), + planner=planner, + now=NOW, + clock=lambda: NOW, + ) + transport = FakeTransport(self.remote) + with patch( + "graphforge_bench.progressive_provider_attempt._rollback_rung", + side_effect=OSError("injected rollback failure"), + ): + outcome = cleanup_only( + self.ledger, self.base / "rollback-recovery.json", transport=transport + ) + self.assertEqual(transport.calls[-1][0], "teardown") + self.assertEqual(outcome["cleanup_failure"], "evidence_cleanup_failed") + ledger = load_ledger(self.ledger) + self.assertEqual(ledger.phase, "cleanup_failed") + self.assertFalse(ledger.resources) + + def test_public_outcome_excludes_sensitive_provider_diagnostics(self) -> None: + self.write_prefix(self.output, 18, 19) + diagnostic = "token=secret Bearer abc@example.com vol_private abcdef01234567 /Users/private" + outcome = self.execute_attempt( + authorization(), + FakeTransport(self.remote, fail_provision=True, diagnostic=diagnostic), + ) + encoded = json.dumps(outcome) + for fragment in diagnostic.split(): + self.assertNotIn(fragment, encoded) + + def test_nonempty_inventory_is_a_cleanup_failure(self) -> None: + self.write_prefix(self.output, 18, 19) + transport = FakeTransport( + self.remote, + teardown_inventory={ + "app_exists": True, + "machines": 2, + "volumes": 0, + "secrets": 1, + }, + ) + outcome = self.execute_attempt(authorization(20), transport) + self.assertEqual(outcome["cleanup_failure"], "inventory_not_empty") + self.assertEqual(outcome["teardown_status"], "failed") + self.assertTrue(load_ledger(self.ledger).resources) + + def test_cleanup_refuses_mismatched_ledger_owner_before_transport(self) -> None: + self.write_prefix(self.output, 18, 19) + execute( + self.invocation, + authorization(20), + transport=FakeTransport(self.remote, fail_teardown=True), + planner=planner, + now=NOW, + clock=lambda: NOW, + ) + document = json.loads(self.ledger.read_text()) + document["attempt_id"] = "d" * 32 + self.ledger.write_text(json.dumps(document)) + transport = FakeTransport(self.remote) + with self.assertRaisesRegex(AttemptError, "ownership"): + cleanup_only(self.ledger, self.base / "recovery.json", transport=transport) + self.assertEqual(transport.calls, []) + + def test_written_ledger_result_and_teardown_inventory_match_schemas(self) -> None: + self.write_prefix(self.output, 18, 19) + result_path = self.base / "attempt-result.json" + document = authorization_document(20) + outcome = execute_attempt( + AttemptRequest( + commit=COMMIT, + organization="fixture-org", + app=APP, + region="dfw", + machine_class="performance-4x", + volume_gib=500, + image_digest=IMAGE, + maximum_scale=20, + spend_authorization=document, + ), + root=ROOT, + output_dir=self.output, + ledger_path=self.ledger, + result_path=result_path, + boundary=FakeTransport(self.remote), + planner=planner, + now=NOW, + clock=lambda: NOW, + ) + inventory_path = self.base / "attempt-result-teardown-inventory.json" + for schema_name, value in ( + ("progressive-spend-authorization.json", document), + ("progressive-provider-attempt-ledger.json", json.loads(self.ledger.read_text())), + ("progressive-provider-attempt-result.json", outcome), + ( + "progressive-provider-teardown-inventory.json", + json.loads(inventory_path.read_text()), + ), + ): + schema = json.loads((ROOT / "schemas" / schema_name).read_text()) + Draft202012Validator(schema).validate(value) + + +if __name__ == "__main__": + unittest.main() From 2c5c4031a0595bcf6613691784fc18f2b24c1871 Mon Sep 17 00:00:00 2001 From: David Spencer <1526975+DecisionNerd@users.noreply.github.com> Date: Sun, 30 Aug 2026 22:12:52 -0600 Subject: [PATCH 2/2] style(benchmarks): format attempt controller --- .../progressive_provider_attempt.py | 30 ++++++------------- .../test_progressive_provider_attempt.py | 8 ++--- 2 files changed, 11 insertions(+), 27 deletions(-) diff --git a/benchmarks/harness/graphforge_bench/progressive_provider_attempt.py b/benchmarks/harness/graphforge_bench/progressive_provider_attempt.py index 8967853fb..49e55bc36 100644 --- a/benchmarks/harness/graphforge_bench/progressive_provider_attempt.py +++ b/benchmarks/harness/graphforge_bench/progressive_provider_attempt.py @@ -174,9 +174,7 @@ def upload_plan(self, *, rung: int, plan_path: Path, deadline: datetime) -> None def execute_rung(self, *, rung: int, image_digest: str, deadline: datetime) -> int: ... - def retrieve_result( - self, *, rung: int, destination: Path, deadline: datetime - ) -> None: ... + def retrieve_result(self, *, rung: int, destination: Path, deadline: datetime) -> None: ... def retrieve_success_artifacts( self, @@ -262,9 +260,7 @@ def parse_spend_authorization(value: str | bytes | Mapping[str, Any]) -> SpendAu } if not isinstance(decoded, dict) or set(decoded) != expected: raise AttemptError("authorization_refused", "spend authorization shape is invalid") - _validate_schema( - "progressive-spend-authorization.json", decoded, "authorization_refused" - ) + _validate_schema("progressive-spend-authorization.json", decoded, "authorization_refused") pricing = decoded.get("pricing") pricing_fields = { "currency", @@ -327,10 +323,7 @@ def parse_spend_authorization(value: str | bytes | Mapping[str, Any]) -> SpendAu seconds = decoded["maximum_machine_seconds"] machine = (pricing["machine_microusd_per_hour"] * seconds + 3599) // 3600 volume = ( - pricing["volume_microusd_per_gib_hour"] - * decoded["volume_gib"] - * seconds - + 3599 + pricing["volume_microusd_per_gib_hour"] * decoded["volume_gib"] * seconds + 3599 ) // 3600 conservative_total = machine + volume + pricing["transfer_allowance_microusd"] if conservative_total > pricing["estimated_total_microusd"]: @@ -589,9 +582,8 @@ def _teardown_observation(value: Mapping[str, Any]) -> dict[str, Any]: expected = {"app_exists", "machines", "volumes", "secrets"} if not isinstance(value, Mapping) or set(value) != expected: raise AttemptError("inventory_unavailable", "teardown inventory is malformed") - if ( - type(value["app_exists"]) is not bool - or any(not _integer(value[name]) or value[name] < 0 for name in expected - {"app_exists"}) + if type(value["app_exists"]) is not bool or any( + not _integer(value[name]) or value[name] < 0 for name in expected - {"app_exists"} ): raise AttemptError("inventory_unavailable", "teardown inventory is malformed") return dict(value) @@ -722,9 +714,7 @@ def execute( save_ledger(invocation.ledger_path, ledger) _atomic_json(control_plan, plan) try: - transport.upload_plan( - rung=next_rung, plan_path=control_plan, deadline=deadline - ) + transport.upload_plan(rung=next_rung, plan_path=control_plan, deadline=deadline) _require_before_deadline(clock_fn, deadline) except AttemptError: raise @@ -864,8 +854,8 @@ def execute( try: observed = transport.teardown(_cleanup_handles(ledger)) ledger.teardown_observed = _teardown_observation(observed) - ledger.teardown_checked_at = clock_fn().astimezone(timezone.utc).strftime( - "%Y-%m-%dT%H:%M:%SZ" + ledger.teardown_checked_at = ( + clock_fn().astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") ) if ledger.teardown_observed != { "app_exists": False, @@ -932,9 +922,7 @@ def _outcome_document(outcome: AttemptOutcome, teardown_inventory_sha256: str) - def _write_outcome(result_path: Path, outcome: AttemptOutcome) -> dict[str, Any]: inventory_path = result_path.with_name(f"{result_path.stem}-teardown-inventory.json") inventory = _teardown_document(outcome) - _validate_schema( - "progressive-provider-teardown-inventory.json", inventory, "evidence_invalid" - ) + _validate_schema("progressive-provider-teardown-inventory.json", inventory, "evidence_invalid") _atomic_json(inventory_path, inventory) inventory_sha256 = hashlib.sha256(inventory_path.read_bytes()).hexdigest() document = _outcome_document(outcome, inventory_sha256) diff --git a/benchmarks/tests/test_progressive_provider_attempt.py b/benchmarks/tests/test_progressive_provider_attempt.py index a46e7ea12..a276ca980 100644 --- a/benchmarks/tests/test_progressive_provider_attempt.py +++ b/benchmarks/tests/test_progressive_provider_attempt.py @@ -236,9 +236,7 @@ def execute_rung(self, *, rung: int, image_digest: str, deadline: datetime) -> i stream.write("\n") return 0 - def retrieve_result( - self, *, rung: int, destination: Path, deadline: datetime - ) -> None: + def retrieve_result(self, *, rung: int, destination: Path, deadline: datetime) -> None: self.calls.append(("retrieve_result", rung)) shutil.copyfile(self.remote / f"s{rung}-result.json", destination) @@ -385,9 +383,7 @@ def test_deadline_crossing_during_provider_operations_cannot_pass(self) -> None: output, remote = base / "evidence", base / "remote" remote.mkdir() self.write_prefix(output, 18, 19) - observations = iter( - (NOW,) * live_observations + (deadline,) * 3 - ) + observations = iter((NOW,) * live_observations + (deadline,) * 3) transport = FakeTransport(remote) outcome = asdict( execute(