feat(azure): safely purge retained validation cells

bin/fm-azure-validation.sh

@@ -48,6 +48,9 @@
 #     --confirm-subscription <exact-id> --confirm-head <exact-sha>
 #   fm-azure-validation.sh retain-failure --cell <azv-id> --confirm-retain \
 #     --confirm-subscription <exact-id>
+#   fm-azure-validation.sh purge-retained --cell <azv-id> --confirm-purge \
+#     --confirm-subscription <exact-id> --confirm-cell <azv-id> \
+#     --confirm-request-digest <sha256:digest>
 #   fm-azure-validation.sh queue
 #   fm-azure-validation.sh auth-seed [--codex <profile>] [--claude <profile>]
 #     [--apply --confirm-seed --confirm-subscription <exact-id>]
@@ -63,14 +66,14 @@ set -euo pipefail
 SCRIPT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd -P)
 
 usage() {
-  sed -n '2,59p' "$0" | sed 's/^# \{0,1\}//'
+  sed -n '2,62p' "$0" | sed 's/^# \{0,1\}//'
 }
 
 case "${1:-}" in
   help|-h|--help|"")
     usage
     ;;
-  build-runtime-bundle|submit|dispatch|drive|observe|collect|status|respond|replace|close|retain-failure|queue|auth-seed)
+  build-runtime-bundle|submit|dispatch|drive|observe|collect|status|respond|replace|close|retain-failure|purge-retained|queue|auth-seed)
     exec python3 "$SCRIPT_DIR/fm-azure-validation.py" "$@"
     ;;
   *)

bin/fm-worker-lifecycle.py

@@ -38,7 +38,8 @@
 # account home is exactly how a credential stager and its remover once resolved
 # different directories and leaked a credential.
 PI_ACCOUNT_HOME_TOOL = ROOT / "bin" / "fm-pi-account-home.py"
-STATE_SCHEMA = "fm.worker-lifecycle/v1"
+LEGACY_STATE_SCHEMA = "fm.worker-lifecycle/v1"
+STATE_SCHEMA = "fm.worker-lifecycle/v2"
 # The scalar pending_action slot this schema carried is superseded by the
 # per-slot pending_actions map. The sentinel is deliberately a string an OLD
 # binary's verify_state refuses ("pending provider action is malformed"), so a
@@ -52,6 +53,7 @@
 RELEASE_SCHEMA = "fm.worker-release/v2"
 AUTHORITY_SCHEMA = "fm.worker-authority/v1"
 CAPACITY_RESERVATION_SCHEMA = "fm.capacity-reservation/v1"
+CAPACITY_FENCE_RETIREMENT_SCHEMA = "fm.capacity-fence-retirement/v1"
 SPECIALIZED_WORKLOAD_ROLES = ("validation", "review", "browser", "networkless-verifier", "crosscheck")
 PROVIDER_REQUEST_SCHEMA = "fm.worker-provider-request/v1"
 PROVIDER_RESPONSE_SCHEMA = "fm.worker-provider-response/v1"
@@ -504,6 +506,7 @@ def empty_state(env):
         "queue": {},
         "workers": {},
         "capacity_reservations": {},
+        "retired_capacity_fences": {},
         "completed_worker_seconds": 0.0,
         "pending_action": LEGACY_PENDING_SENTINEL,
         "pending_actions": {},
@@ -525,6 +528,7 @@ def verify_state(env, state):
         not isinstance(state.get("queue"), dict)
         or not isinstance(state.get("workers"), dict)
         or not isinstance(state.get("capacity_reservations"), dict)
+        or not isinstance(state.get("retired_capacity_fences"), dict)
         or not isinstance(state.get("executions"), dict)
     ):
         raise LifecycleError("lifecycle queue, worker, or shared capacity inventory is malformed")
@@ -560,6 +564,26 @@ def verify_state(env, state):
         require_binding("capacity reservation fence", reservation.get("fence_binding"))
         if "shape_id" in reservation:
             require_id("capacity shape id", reservation.get("shape_id"))
+    for fence, retirement in state["retired_capacity_fences"].items():
+        if (
+            not isinstance(retirement, dict)
+            or retirement.get("schema") != CAPACITY_FENCE_RETIREMENT_SCHEMA
+            or retirement.get("fence_binding") != fence
+            or not isinstance(retirement.get("reservation_ids"), list)
+            or not retirement.get("reservation_ids")
+            or retirement.get("reservation_ids")
+            != sorted(set(retirement.get("reservation_ids") or []))
+            or len(retirement.get("reservation_ids") or []) > 256
+            or not isinstance(retirement.get("retired_at"), str)
+            or not retirement.get("retired_at")
+        ):
+            raise LifecycleError("durable specialized capacity fence retirement is malformed")
+        require_binding("retired capacity fence", fence)
+        require_binding(
+            "capacity fence retirement receipt", retirement.get("retirement_receipt")
+        )
+        for reservation_id in retirement["reservation_ids"]:
+            require_id("retired capacity reservation id", reservation_id)
     legacy = state.get("pending_action")
     if legacy is not None and legacy != LEGACY_PENDING_SENTINEL:
         # A dict here means load_state's migration did not run; anything else
@@ -599,7 +623,14 @@ def load_state(env):
         if "is absent" not in str(exc):
             raise
         state = empty_state(env)
+    if state.get("schema") == LEGACY_STATE_SCHEMA:
+        # A v1 document has no fence-retirement authority to preserve. Upgrade
+        # it in memory; the next locked save makes the v2 rollback fence
+        # durable, after which a v1 binary refuses instead of reopening a
+        # retired fence it does not understand.
+        state["schema"] = STATE_SCHEMA
     state.setdefault("capacity_reservations", {})
+    state.setdefault("retired_capacity_fences", {})
     state.setdefault("executions", {})
     state.setdefault("pending_actions", {})
     state.setdefault("revision", 0)
@@ -2763,6 +2794,15 @@ def parser():
     capacity_release.add_argument("--cleanup-receipt", required=True)
     capacity_release.add_argument("--confirm-subscription", required=True)
 
+    capacity_retire = sub.add_parser(
+        "capacity-retire-fence",
+        help="permanently close one exact specialized capacity fence after release",
+    )
+    capacity_retire.add_argument("--fence-binding", required=True)
+    capacity_retire.add_argument("--reservation-id", action="append", required=True)
+    capacity_retire.add_argument("--retirement-receipt", required=True)
+    capacity_retire.add_argument("--confirm-subscription", required=True)
+
     execute = sub.add_parser("execute", help="run one exact private task command and collect its bound result")
     execute.add_argument("--task", required=True)
     execute.add_argument("--task-generation", required=True)
@@ -3339,13 +3379,19 @@ def specialized_reservation_from_args(args):
     )
 
 
+def refuse_retired_capacity_fence(state, fence):
+    if fence in state["retired_capacity_fences"]:
+        raise LifecycleError("retired capacity fence cannot admit another reservation")
+
+
 def command_capacity_reserve(env, args):
     if args.confirm_subscription != env["subscription"]:
         raise LifecycleError("--confirm-subscription must exactly match FM_AZURE_SUBSCRIPTION_ID")
     candidate = specialized_reservation_from_args(args)
     reservation_id = candidate["reservation_id"]
     with controller_lock(env):
         state = load_state(env)
+        refuse_retired_capacity_fence(state, candidate["fence_binding"])
         existing = state["capacity_reservations"].get(reservation_id)
         readmission_id = None
         identity_fields = (
@@ -3477,6 +3523,7 @@ def command_capacity_reserve_shape(env, args):
     )
     with controller_lock(env):
         state = load_state(env)
+        refuse_retired_capacity_fence(state, args.fence_binding)
         entries = []
         for candidate in constituents:
             existing = state["capacity_reservations"].get(candidate["reservation_id"])
@@ -3606,6 +3653,103 @@ def command_capacity_release(env, args):
     print("specialized capacity reservation released after exact zero-compute proof")
 
 
+def exact_provider_capacity_identity(reservation, provider):
+    return (
+        isinstance(reservation, dict)
+        and reservation.get("schema") == CAPACITY_RESERVATION_SCHEMA
+        and reservation.get("reservation_id") == provider.get("reservation_id")
+        and reservation.get("role") == provider.get("role")
+        and reservation.get("sku") == provider.get("sku")
+        and str(reservation.get("sku_family", "")).lower()
+        == str(provider.get("sku_family", "")).lower()
+        and reservation.get("vcpus") == provider.get("vcpus")
+        and not isinstance(reservation.get("amount_usd"), bool)
+        and isinstance(reservation.get("amount_usd"), (int, float))
+        and not isinstance(provider.get("amount_usd"), bool)
+        and isinstance(provider.get("amount_usd"), (int, float))
+        and math.isclose(
+            float(reservation["amount_usd"]), float(provider["amount_usd"]),
+            rel_tol=0.0, abs_tol=1e-6,
+        )
+    )
+
+
+def command_capacity_retire_fence(env, args):
+    """Atomically close a released fence against every future admission.
+
+    Provider inventory and the complete same-fence ledger census occur while
+    the shared controller lock excludes both reservation entry points. The v2
+    retirement tombstone is committed before that lock opens, so successful
+    return is the irreversible admission barrier an artifact purge can rely on.
+    """
+    if args.confirm_subscription != env["subscription"]:
+        raise LifecycleError("--confirm-subscription must exactly match FM_AZURE_SUBSCRIPTION_ID")
+    fence = require_binding("capacity reservation fence", args.fence_binding)
+    receipt = require_binding("capacity fence retirement receipt", args.retirement_receipt)
+    reservation_ids = sorted(set(
+        require_id("retired capacity reservation id", value)
+        for value in args.reservation_id
+    ))
+    if len(reservation_ids) != len(args.reservation_id) or len(reservation_ids) > 256:
+        raise LifecycleError("capacity fence retirement reservation ids are not exact and distinct")
+    expected = {
+        "schema": CAPACITY_FENCE_RETIREMENT_SCHEMA,
+        "fence_binding": fence,
+        "reservation_ids": reservation_ids,
+        "retirement_receipt": receipt,
+    }
+    with controller_lock(env):
+        state = load_state(env)
+        prior = state["retired_capacity_fences"].get(fence)
+        if prior is not None and any(prior.get(key) != value for key, value in expected.items()):
+            raise LifecycleError("capacity fence already has a different retirement identity")
+        allowed = set(reservation_ids)
+        same_fence = {
+            reservation_id: reservation
+            for reservation_id, reservation in state["capacity_reservations"].items()
+            if isinstance(reservation, dict) and reservation.get("fence_binding") == fence
+        }
+        outside = sorted(set(same_fence) - allowed)
+        if outside:
+            raise LifecycleError(
+                "capacity fence retirement census found an unplanned reservation: {}".format(
+                    outside[0]
+                )
+            )
+        for reservation_id, reservation in same_fence.items():
+            if (
+                reservation.get("reservation_id") != reservation_id
+                or reservation.get("status") != "released"
+                or not HEX_BINDING.match(str(reservation.get("cleanup_receipt", "")).split(":")[-1])
+            ):
+                raise LifecycleError(
+                    "capacity fence retirement requires every exact reservation released"
+                )
+        inventory = provider_call(env, "inventory")["inventory"]
+        provider_reservations = inventory.get("capacity_reservations")
+        if not isinstance(provider_reservations, list):
+            raise LifecycleError("provider capacity inventory is malformed")
+        for provider in provider_reservations:
+            if not isinstance(provider, dict) or provider.get("active") is not True:
+                continue
+            reservation_id = provider.get("reservation_id")
+            controller = state["capacity_reservations"].get(reservation_id)
+            if not exact_provider_capacity_identity(controller, provider):
+                raise LifecycleError(
+                    "provider-active capacity lacks exact controller identity during fence retirement"
+                )
+            if reservation_id in allowed or controller.get("fence_binding") == fence:
+                raise LifecycleError(
+                    "provider still observes active capacity on the retiring fence"
+                )
+        if prior is None:
+            state["retired_capacity_fences"][fence] = dict(expected, retired_at=iso_utc())
+            save_state(env, state)
+            print("specialized capacity fence retired after exact release census")
+        else:
+            print("specialized capacity fence already retired with exact identity")
+
+
 # What an ORDINARY crewmate payload may contain: the repository as a
 # credential-free bundle, plus the one task file its entrypoint reads. This set
 # is deliberately NOT widened for the compartment lane; see below.
@@ -4663,6 +4807,8 @@ def main(argv=None):
         command_capacity_reserve_shape(env, args)
     elif args.command == "capacity-release":
         command_capacity_release(env, args)
+    elif args.command == "capacity-retire-fence":
+        command_capacity_retire_fence(env, args)
     elif args.command == "execute":
         command_execute(env, args)
     elif args.command == "authority-receipt":

bin/fm-worker-lifecycle.sh

@@ -45,6 +45,7 @@
 #   fm-worker-lifecycle.sh message-put <exact assignment flags> --file <json> | --attach <bundle>
 #   fm-worker-lifecycle.sh message-collect <exact assignment flags> --output-dir <dir>
 #   fm-worker-lifecycle.sh compartment-chain-tip <exact assignment flags> --sequence <n> --chain-digest <sha256>
+#   fm-worker-lifecycle.sh capacity-retire-fence <exact released fence and reservation ids>
 #   fm-worker-lifecycle.sh status [--live] [--json]
 #   fm-worker-lifecycle.sh acceptance-plan
 set -euo pipefail
@@ -137,7 +138,7 @@ fm_worker_receipt_credential_remains() {  # <task home file> <task id> <controll
 }
 
 case "${1:-}" in
-  request|release|resume|steer|execute|authority-receipt|capacity-reserve|capacity-reserve-shape|capacity-release|abandon-claim|message-put|message-collect|compartment-chain-tip)
+  request|release|resume|steer|execute|authority-receipt|capacity-reserve|capacity-reserve-shape|capacity-release|capacity-retire-fence|abandon-claim|message-put|message-collect|compartment-chain-tip)
     fm_refuse_if_gate_agent
     exec python3 "$SCRIPT_DIR/fm-worker-lifecycle.py" "$@"
     ;;

docs/azure-validation.md

@@ -56,6 +56,7 @@ A heavy validation request reserves its complete 40-vCPU peak atomically through
 Shard constituents never use the selected control family, so the complete shape fits the exact 10-vCPU families.
 If any constituent fails regional, exact-family, specialized-envelope, or budget admission, no constituent is reserved and the complete request stays durably queued with the exact refusal.
 Child runner VMs re-admit their exact pre-reserved constituent ids idempotently through the same allocator, so their live processors are covered once without double-counting, and each child's exact first-day cost bound may re-admit at or below its cushioned constituent amount.
+An explicit purge of a failed-retained cell first releases its sealed remaining constituents, then atomically retires the exact validation fence in the shared allocator. Retirement holds the allocator admission lock across a fresh entire-ledger and provider-inventory zero proof and the durable tombstone save; every later single or shape reserve on that fence refuses. Retained disk, private container, RBAC, and identity deletion begins only after that permanent barrier succeeds, including on retries after a crash.
 The one-shot runner independently refuses any child beyond the parent's reserved shard slots.
 An author worker must match the foundation's exact general-worker resource class before its processor count is trusted.
 Quota is only a live upper bound and never turns into a warm allocation.
@@ -451,6 +452,41 @@ bin/fm-azure-validation.sh retain-failure \
 No failure path deletes a worktree disk, credential disk, private container, or result object.
 No command names or deletes a resource group, subnet, foundation identity, another cell, another shard, another task prefix, or a local daemon.
 
+### Explicit retained-failure purge
+
+`purge-retained` is the deliberate, irreversible exception for a failed cell whose retained diagnosis is no longer needed.
+It accepts only `failed-retained`, its own resumable `purging` phase, or the terminal `purged` tombstone.
+The operator must repeat the exact subscription, cell, and request digest in addition to the destructive confirmation:
+
+```sh
+bin/fm-azure-validation.sh purge-retained \
+  --cell '<azv-id>' \
+  --confirm-purge \
+  --confirm-subscription "$FM_AZURE_SUBSCRIPTION_ID" \
+  --confirm-cell '<same-azv-id>' \
+  --confirm-request-digest '<sha256:exact-request-digest>'
+```
+
+The command acquires the cell's shard-driver lock before its cell-state lock, so it cannot purge while the same cell is driving child invocations.
+Before any destructive call, it proves the control VM, NIC, OS disk, Run Commands, and shutdown schedule absent.
+It also reads every runner state owned by the cell, requires every retry lineage to be terminal with Azure compute absent, and requires every dispatched reservation to carry a durable released receipt.
+Every control and shard constituent must still have the exact durable allocator id, fence, shape, SKU, family, vCPU, and cost identity recorded at admission.
+The allocator's entire exact-fence ledger must be inside the sealed control, planned-root, and runner-state census; any outside row refuses before the purge plan is sealed, regardless of release status, provider inactivity, or recorded workload role.
+The pinned shared-provider inventory must show every exact censused reservation inactive or absent with matching SKU, family, vCPU, and cost identity; a provider-active exact id, including one hidden behind a stale released allocator row, refuses.
+Because provider inventory is not fence-bound, any provider-active id without an exact controller reservation also refuses; an unrelated active reservation is ignored only when its controller record proves a different fence.
+An already-released exact constituent is accepted with its cleanup receipt, while any admitted constituent with no dispatch lineage and a queued or reserved status is listed separately for exact release.
+
+It then proves the worktree disk is detached with its recorded stable identity and current ETag, proves the private container and its complete two-role inventory, and proves the cell identity has only its exact container and auth-share grants.
+Those identities, every verified shard lineage, and only the remaining capacity constituents are sealed into an immutable purge plan.
+The state durably enters non-replaceable `purging` with the plan digest before the first deletion.
+Any remaining exact capacity constituent is released first and then read back as durably released and provider-inactive or absent before the retained disk, RBAC, container, or identity is touched.
+Every retry repeats the complete fresh allocator/provider census before attempting a remaining release, and the census is repeated after release before artifact deletion, so a new same-fence reservation can never fall outside the sealed plan.
+
+Retries never rebuild or widen that plan.
+They accept only remaining subsets of its disk, role, container, identity, and capacity identities, use the stored ETags for conditional deletion, and persist progress after every boundary.
+A partial or ambiguous attempt remains `purging`; `replace` and `retain-failure` cannot reclaim it.
+Completion retains the local state as a `purged` tombstone with the immutable plan and digest, and repeating the exact command is a no-op.
+
 ## Cleanup order
 
 Successful close occurs only after complete result/report/evidence collection, CI-green proof, exact remote-current head proof, and explicit head confirmation.