diff --git a/docs/cookbook/config.md b/docs/cookbook/config.md
index 1ede588f..111f6296 100644
--- a/docs/cookbook/config.md
+++ b/docs/cookbook/config.md
@@ -226,6 +226,8 @@ keys you need to change.
 | `disk_bw_max_load` | `.disk_bw_max_load` | `.disk_bw_max_load` | `0.8`     | [$\beta$ (soft) budget](./architecture.md#budgets): disk-bandwidth load threshold, a fraction `0.0`–`1.0` of the cap. |
 | `net_bw_cap`       | `.net_bw_cap`       | `.net_bw_cap`       | `1 GiBps` | [$\beta$ (soft) budget](./architecture.md#budgets): soft network-bandwidth capacity ([unit](#unit)). |
 | `net_bw_max_load`  | `.net_bw_max_load`  | `.net_bw_max_load`  | `0.8`     | [$\beta$ (soft) budget](./architecture.md#budgets): network-bandwidth load threshold, a fraction `0.0`–`1.0` of the cap. |
+| `boot_lease_ttl`   | `.boot_lease_ttl`   | `.boot_lease_ttl`   | `300 s`   | How long a placement may hold admitted capacity before its VM has booted. A backstop for a wedged boot; must exceed the slowest legitimate boot, including a cold image pull. |
+| `restart_grace`    | `.restart_grace`    | `.restart_grace`    | `5 s`     | How long a crashed VM's capacity is held while its supervisor restarts it, so a competing placement cannot take capacity the VM is about to reclaim. |
 
 <!-- tabs open -->
 ### `config.exs`
@@ -239,7 +241,9 @@ config :hyper, Hyper.Cfg.Budget,
   disk_bw_cap: Unit.Bandwidth.gibps(1),
   disk_bw_max_load: 0.8,
   net_bw_cap: Unit.Bandwidth.gibps(1),
-  net_bw_max_load: 0.8
+  net_bw_max_load: 0.8,
+  boot_lease_ttl: Unit.Time.s(300),
+  restart_grace: Unit.Time.s(5)
 ```
 
 ### `config.toml`
@@ -254,6 +258,8 @@ disk_bw_cap = "1GiBps"
 disk_bw_max_load = 0.8
 net_bw_cap = "1GiBps"
 net_bw_max_load = 0.8
+boot_lease_ttl = "300s"
+restart_grace = "5s"
 ```
 <!-- tabs close -->
 
diff --git a/lib/hyper.ex b/lib/hyper.ex
index 4cb03cbc..1f018d56 100644
--- a/lib/hyper.ex
+++ b/lib/hyper.ex
@@ -23,14 +23,13 @@ defmodule Hyper do
       instance_spec = Hyper.Vm.Instance.spec(spec.type)
 
       start_fun = fn -> Hyper.Node.start_image_vm(vm_id, spec) end
-      stop_fun = fn pid -> Hyper.Node.stop_image_vm(pid) end
 
       # Rank candidates by how many of the image's layer bytes they already
       # have mounted; a fork's freshly published delta is resident nowhere yet,
       # so its parent's base layers dominate the score — which is the point.
       layers = Hyper.Img.Db.Image.chain_sizes(spec.img_id)
 
-      case Hyper.Cluster.Scheduler.run(instance_spec, layers, start_fun, stop_fun) do
+      case Hyper.Cluster.Scheduler.run(vm_id, instance_spec, layers, start_fun) do
         {:ok, {_node, pid}} -> {:ok, pid}
         {:error, _} = err -> err
       end
diff --git a/lib/hyper/cfg/budget.ex b/lib/hyper/cfg/budget.ex
index 91dcb812..9be962a5 100644
--- a/lib/hyper/cfg/budget.ex
+++ b/lib/hyper/cfg/budget.ex
@@ -18,7 +18,9 @@ defmodule Hyper.Cfg.Budget do
           disk_bw_cap: Unit.Bandwidth.t(),
           disk_bw_max_load: float(),
           net_bw_cap: Unit.Bandwidth.t(),
-          net_bw_max_load: float()
+          net_bw_max_load: float(),
+          boot_lease_ttl: Unit.Time.t(),
+          restart_grace: Unit.Time.t()
         }
   defstruct [
     :mem_max,
@@ -28,7 +30,9 @@ defmodule Hyper.Cfg.Budget do
     :disk_bw_cap,
     :disk_bw_max_load,
     :net_bw_cap,
-    :net_bw_max_load
+    :net_bw_max_load,
+    :boot_lease_ttl,
+    :restart_grace
   ]
 
   @default_mem_max Unit.Information.gib(4)
@@ -40,6 +44,15 @@ defmodule Hyper.Cfg.Budget do
   @default_net_bw_cap Unit.Bandwidth.gibps(1)
   @default_net_bw_max_load 0.8
 
+  # A boot lease is a backstop for a caller that is alive but wedged, not the
+  # primary release mechanism (the monitor is). It must comfortably exceed the
+  # slowest legitimate boot — a cold image pull — because expiring a lease under
+  # a still-booting VM leaves it running unaccounted.
+  @default_boot_lease_ttl Unit.Time.s(300)
+  # Long enough to cover a `:transient` FireVMM restart, short enough that a VM
+  # that is never coming back frees its capacity promptly.
+  @default_restart_grace Unit.Time.s(5)
+
   @spec load :: {:ok, t()} | {:error, term()}
   def load do
     with {:ok, mem_max} <- information(:mem_max, "budget.mem_max", @default_mem_max),
@@ -52,7 +65,11 @@ defmodule Hyper.Cfg.Budget do
            number(:disk_bw_max_load, "budget.disk_bw_max_load", @default_disk_bw_max_load),
          {:ok, net_bw_cap} <- bandwidth(:net_bw_cap, "budget.net_bw_cap", @default_net_bw_cap),
          {:ok, net_bw_max_load} <-
-           number(:net_bw_max_load, "budget.net_bw_max_load", @default_net_bw_max_load) do
+           number(:net_bw_max_load, "budget.net_bw_max_load", @default_net_bw_max_load),
+         {:ok, boot_lease_ttl} <-
+           duration(:boot_lease_ttl, "budget.boot_lease_ttl", @default_boot_lease_ttl),
+         {:ok, restart_grace} <-
+           duration(:restart_grace, "budget.restart_grace", @default_restart_grace) do
       config = %__MODULE__{
         mem_max: mem_max,
         disk_max: disk_max,
@@ -61,7 +78,9 @@ defmodule Hyper.Cfg.Budget do
         disk_bw_cap: disk_bw_cap,
         disk_bw_max_load: disk_bw_max_load,
         net_bw_cap: net_bw_cap,
-        net_bw_max_load: net_bw_max_load
+        net_bw_max_load: net_bw_max_load,
+        boot_lease_ttl: boot_lease_ttl,
+        restart_grace: restart_grace
       }
 
       :persistent_term.put(__MODULE__, config)
@@ -88,6 +107,14 @@ defmodule Hyper.Cfg.Budget do
     end
   end
 
+  @spec duration(atom(), String.t(), Unit.Time.t()) ::
+          {:ok, Unit.Time.t()} | {:error, term()}
+  defp duration(key, toml, default) do
+    with {:ok, v} <- required(key, toml, default) do
+      coerce(v, &Unit.Time.parse/1, Unit.Time, key)
+    end
+  end
+
   @spec number(atom(), String.t(), number()) :: {:ok, number()} | {:error, term()}
   defp number(key, toml, default) do
     case required(key, toml, default) do
diff --git a/lib/hyper/cluster/scheduler.ex b/lib/hyper/cluster/scheduler.ex
index 4877658c..b1bf20e6 100644
--- a/lib/hyper/cluster/scheduler.ex
+++ b/lib/hyper/cluster/scheduler.ex
@@ -5,7 +5,7 @@ defmodule Hyper.Cluster.Scheduler do
   nodes that cannot fit the spec, and ranks survivors by how many bytes of the
   VM's image layers they already have mounted (`colo(N, VM) = sum of |L|` over
   shared mounted layers). The result is an ordered candidate list; the chosen
-  node confirms authoritatively via `Hyper.Node.Budget.admit/2` (see `place/3`).
+  node confirms authoritatively via `Hyper.Node.Budget.lease/2` (see `place/3`).
 
   All filtering is best-effort on a possibly-stale snapshot: a node that no
   longer fits simply refuses at confirmation time.
@@ -63,20 +63,20 @@ defmodule Hyper.Cluster.Scheduler do
   @doc """
   Place and boot `spec` somewhere in the cluster.
 
-  Confirms each candidate by RPC-ing `Hyper.Node.try_run/3` on it; the first node
-  to boot the VM and reserve its budget wins. `start_fun`/`stop_fun` describe how
-  to boot/tear down the VM on the target node.
+  Confirms each candidate by RPC-ing `Hyper.Node.try_run/3` on it; the first
+  node to admit `spec` and boot it wins. A candidate that refuses has built
+  nothing, so walking the list is cheap.
   """
   @spec run(
+          Hyper.Vm.Id.t(),
           Spec.t(),
           layer_sizes(),
-          (-> {:ok, pid()} | {:error, term()}),
-          (pid() -> :ok)
+          (-> {:ok, pid()} | {:error, term()})
         ) :: {:ok, {node(), pid()}} | {:error, :no_capacity}
-  @decorate with_span("Hyper.Cluster.Scheduler.run", include: [:spec])
-  def run(spec, layers, start_fun, stop_fun) do
+  @decorate with_span("Hyper.Cluster.Scheduler.run", include: [:vm_id, :spec])
+  def run(vm_id, spec, layers, start_fun) do
     attempt = fn target ->
-      :erpc.call(target, Hyper.Node, :try_run, [spec, start_fun, stop_fun])
+      :erpc.call(target, Hyper.Node, :try_run, [vm_id, spec, start_fun])
     end
 
     place(spec, layers, attempt)
diff --git a/lib/hyper/node.ex b/lib/hyper/node.ex
index 01c7d13f..3ab22945 100644
--- a/lib/hyper/node.ex
+++ b/lib/hyper/node.ex
@@ -186,7 +186,7 @@ defmodule Hyper.Node do
       child_vm_id = Hyper.Vm.Id.generate()
       spec = Hyper.Vm.Instance.spec(parent.type)
 
-      try_run(spec, fn -> start_forked_vm(child_vm_id, parent) end, &stop_image_vm/1)
+      try_run(child_vm_id, spec, fn -> start_forked_vm(child_vm_id, parent) end)
     end
   end
 
@@ -269,35 +269,40 @@ defmodule Hyper.Node do
   end
 
   @doc """
-  Start a VM here and confirm its budget.
-
-  `start_fun` boots the VM and returns `{:ok, vm_pid}`; the reservation is held
-  against `vm_pid` and released when it dies. If the reserve loses a race (the
-  node filled up since the scheduler's snapshot) the just-started VM is torn down
-  via `stop_fun` and `{:error, reason}` is returned.
+  Admit `spec` on this node, then boot it.
+
+  Capacity is leased BEFORE `start_fun` runs, so a refusal costs nothing
+  physical and a concurrent herd cannot boot past the node's budget. The VM
+  turns that lease into its own reservation from `Hyper.Node.FireVMM.init/1`,
+  which `DynamicSupervisor.start_child` has already run by the time `start_fun`
+  returns — so this function drops its lease unconditionally and never owns the
+  reservation itself.
   """
-  @spec try_run(
-          Hyper.Vm.Instance.Spec.t(),
-          (-> {:ok, pid()} | {:error, term()}),
-          (pid() -> :ok)
-        ) :: {:ok, pid()} | {:error, term()}
-  def try_run(spec, start_fun, stop_fun) do
-    case start_fun.() do
-      {:ok, pid} ->
-        case Hyper.Node.Budget.admit(spec, pid) do
-          :ok ->
-            {:ok, pid}
-
-          {:error, reason} ->
-            :ok = stop_fun.(pid)
-            {:error, reason}
-        end
-
-      {:error, reason} ->
-        {:error, reason}
+  @spec try_run(Hyper.Vm.Id.t(), Hyper.Vm.Instance.Spec.t(), (-> {:ok, pid()} | {:error, term()})) ::
+          {:ok, pid()} | {:error, term()}
+  @decorate with_span("Hyper.Node.try_run", include: [:vm_id, :spec])
+  def try_run(vm_id, spec, start_fun) do
+    with {:ok, token} <- Hyper.Node.Budget.lease(vm_id, spec) do
+      try do
+        start_fun.()
+      after
+        release(vm_id, token)
+      end
     end
   end
 
+  # Dropping the lease must not turn a successful boot into a raise: `Hard` is a
+  # single GenServer, so a restart or a queue timeout here would otherwise
+  # propagate out of the `after` over a VM that is already live and claimed.
+  # Swallowing is safe because the lease is monitored on this process too, and
+  # this process exits as soon as `try_run/3` returns.
+  @spec release(Hyper.Vm.Id.t(), reference()) :: :ok
+  defp release(vm_id, token) do
+    Hyper.Node.Budget.drop(vm_id, token)
+  catch
+    :exit, _ -> :ok
+  end
+
   @spec test_system :: :ok | {:error, term()}
   def test_system do
     with {:ok, _} <- Hyper.Cfg.Budget.load(),
@@ -357,11 +362,29 @@ defmodule Hyper.Node do
     end
   end
 
-  defp start_vm_or_release(opts, uid, mutable) do
+  @doc false
+  # Public (not private) so the `:ignore` handling below is reachable from a
+  # hermetic test without a real jail — `boot_with_mutable/4`'s only caller.
+  @spec start_vm_or_release(FireVMM.Opts.t(), Users.id(), pid()) ::
+          {:ok, pid()} | {:error, term()}
+  def start_vm_or_release(opts, uid, mutable) do
     case start_vm(opts) do
       {:ok, pid} ->
         {:ok, pid}
 
+      # `FireVMM.init/1` declines with `:ignore` when either step of its `with`
+      # refuses: `Hyper.Cluster.Routing.register_self/1` finds the vm_id
+      # already registered (a stale dead incarnation), or
+      # `Hyper.Node.Budget.claim/2` has no lease left to claim (the lease
+      # expired or its holder died mid-boot). Either way it is a refusal, not
+      # a fault, but it must still hand back the uid and the mutable layer:
+      # nothing else will, and a leaked uid is unrecoverable short of
+      # restarting the node.
+      :ignore ->
+        Img.Mutable.release(mutable)
+        Users.release(uid)
+        {:error, :not_admitted}
+
       {:error, reason} ->
         Img.Mutable.release(mutable)
         Users.release(uid)
diff --git a/lib/hyper/node/budget.ex b/lib/hyper/node/budget.ex
index c93a5224..1ce7b386 100644
--- a/lib/hyper/node/budget.ex
+++ b/lib/hyper/node/budget.ex
@@ -1,8 +1,8 @@
 defmodule Hyper.Node.Budget do
   @moduledoc """
-  Public entry point for this node's resource budget. Thin facade over
-  `Hyper.Node.Budget.Hard`, the per-node accounting GenServer supervised by
-  `Hyper.Node.Budget.Supervisor`.
+  Public entry point for this node's resource budget, fronting leases. Thin
+  facade over `Hyper.Node.Budget.Hard`, the per-node accounting GenServer
+  supervised by `Hyper.Node.Budget.Supervisor`.
   """
 
   alias Hyper.Node.Budget.Hard
@@ -10,24 +10,25 @@ defmodule Hyper.Node.Budget do
 
   use OpenTelemetryDecorator
 
-  @doc "Can this node run the given vm spec? `:ok` if yes, `{:error, reason}` otherwise."
-  @spec can_run(Hyper.Vm.Instance.Spec.t()) :: :ok | {:error, term()}
-  defdelegate can_run(vm_spec), to: Hard
-
-  @doc "Reserve the spec's budget, run `callable`, and release the budget afterwards."
-  @spec with_budget(Hyper.Vm.Instance.Spec.t(), (-> result)) :: result | {:error, term()}
-        when result: var
-  defdelegate with_budget(vm_spec, callable), to: Hard
-
   @doc """
-  Authoritatively confirm this node can run `spec`, reserving its budget for the
-  lifetime of `owner`. Live soft-load check first, then an atomic hard reserve.
+  Provisionally admit `spec` for `vm_id` on this node, before anything is built.
+
+  Live soft-load check first, then an atomic hard lease. Returns a token for
+  `drop/2`; the VM turns the lease into its own reservation with `claim/2`.
   """
-  @spec admit(Spec.t(), pid()) :: :ok | {:error, term()}
-  @decorate with_span("Hyper.Node.Budget.admit", include: [:spec])
-  def admit(spec, owner) do
+  @spec lease(Hyper.Vm.Id.t(), Spec.t()) :: {:ok, reference()} | {:error, term()}
+  @decorate with_span("Hyper.Node.Budget.lease", include: [:vm_id, :spec])
+  def lease(vm_id, spec) do
     with :ok <- Hyper.Node.Budget.Soft.can_run(spec) do
-      Hard.reserve(spec, owner)
+      Hard.lease(vm_id, spec)
     end
   end
+
+  @doc "Bind `vm_id`'s leased capacity to `owner` for that process's lifetime."
+  @spec claim(Hyper.Vm.Id.t(), pid()) :: :ok | {:error, :no_lease}
+  defdelegate claim(vm_id, owner), to: Hard
+
+  @doc "Release the lease `token` identifies. A no-op once the VM has claimed it."
+  @spec drop(Hyper.Vm.Id.t(), reference()) :: :ok
+  defdelegate drop(vm_id, token), to: Hard
 end
diff --git a/lib/hyper/node/budget/hard.ex b/lib/hyper/node/budget/hard.ex
index fcd325cb..201f51ab 100644
--- a/lib/hyper/node/budget/hard.ex
+++ b/lib/hyper/node/budget/hard.ex
@@ -2,13 +2,26 @@ defmodule Hyper.Node.Budget.Hard do
   @moduledoc """
   Hard per-node resource accounting. One `Hard` runs per BEAM node (named
   `__MODULE__`, started under `Hyper.Node.Budget.Supervisor`) and tracks how much
-  memory and disk the VMs scheduled onto this machine have reserved.
-
-  "Hard" means the limits are inviolable: a reservation that would push the
-  running total past the node's configured `mem_max`/`disk_max`
-  (`Hyper.Cfg.Budget`) is refused outright. Callers reserve through
-  `with_budget/2`, which holds the budget for the duration of the callback and
-  releases it afterwards.
+  memory and disk this machine's VMs hold.
+
+  Capacity is granted *before* a VM boots. `lease/2` reserves against the placing
+  caller and returns a token; the VM then converts that lease into a reservation
+  of its own via `claim/2` from inside `Hyper.Node.FireVMM.init/1`, and the
+  caller drops its token. A lease counts against the caps exactly like a
+  reservation, so a concurrent herd cannot all pass admission and then boot into
+  memory nothing has taken.
+
+  A lease is released by three independent mechanisms, because no one of them
+  covers every failure:
+
+    * the **monitor** on the leasing process — a crash, in microseconds;
+    * the **`boot_lease_ttl`** — a caller that is alive but wedged, which no
+      monitor will ever fire for;
+    * `drop/2` — the normal path, once the boot has finished either way.
+
+  When a claimed owner dies its reservation becomes a lease again for
+  `restart_grace`, so a `:transient` VM restart re-claims the same capacity
+  rather than racing a competing placement for it.
   """
 
   use GenServer
@@ -19,56 +32,163 @@ defmodule Hyper.Node.Budget.Hard do
   alias Hyper.Vm.Instance
 
   defmodule State do
-    @moduledoc false
+    @moduledoc """
+    Pure ledger for one node's hard budget. One entry per vm_id, in one of two
+    kinds:
 
-    @type t :: %__MODULE__{
-            mem_allocated: Unit.Information.t(),
-            disk_allocated: Unit.Information.t(),
-            reservations: %{reference() => Hyper.Vm.Instance.Spec.t()}
-          }
+      * `{:leased, spec, token, expires_at}` — capacity granted to a boot that
+        has not happened yet. Revocable by `drop/3` with the matching token, or
+        by `expire/2` once `expires_at` passes.
+      * `{:claimed, spec, owner_ref}` — capacity held by a live VM. Immune to
+        `drop/3` and to `expire/2`; released only via `release/3`, which turns
+        it back into a short-lived lease so a supervisor restart leaves no gap.
 
-    defstruct [:mem_allocated, :disk_allocated, reservations: %{}]
+    Both kinds occupy capacity. That is the invariant admission rests on: a VM
+    that is about to exist must be as visible to `lease/5` as one that already
+    is.
+
+    Pure — no processes and no clock. `expires_at`, and the `now` passed to
+    `expire/2`, are monotonic milliseconds supplied by the caller.
+    """
 
     use Unit.Operators
 
-    @spec zero() :: t()
-    def zero do
-      %__MODULE__{
-        mem_allocated: Unit.Information.zero(),
-        disk_allocated: Unit.Information.zero()
-      }
+    alias Hyper.Vm.Instance.Spec
+    alias Unit.Information
+
+    @type token :: reference()
+    @type caps :: %{mem: Information.t(), disk: Information.t()}
+    @type entry ::
+            {:leased, Spec.t(), token(), integer()}
+            | {:claimed, Spec.t(), reference()}
+    @type t :: %__MODULE__{entries: %{Hyper.Vm.Id.t() => entry()}}
+
+    defstruct entries: %{}
+
+    @spec new() :: t()
+    def new, do: %__MODULE__{}
+
+    @doc "Memory and disk held by every entry, leased and claimed alike."
+    @spec allocated(t()) :: caps()
+    def allocated(%__MODULE__{entries: entries}) do
+      Enum.reduce(entries, %{mem: Information.zero(), disk: Information.zero()}, fn
+        {_vm_id, entry}, acc ->
+          spec = spec_of(entry)
+          %{mem: acc.mem + spec.mem, disk: acc.disk + spec.disk}
+      end)
     end
 
-    @doc "Add `spec`'s reservation to the running total."
-    @spec bump(t(), Instance.Spec.t()) :: t()
-    def bump(s, spec) do
-      %{
-        s
-        | mem_allocated: s.mem_allocated + spec.mem,
-          disk_allocated: s.disk_allocated + spec.disk
-      }
+    @doc """
+    Grant `spec`'s capacity to `vm_id` provisionally, expiring at `expires_at`.
+
+    Refuses `:already_held` rather than overwriting: a vm_id is unique per VM,
+    so a second lease is a bug, and silently replacing a claimed entry would
+    leave a live VM unaccounted.
+    """
+    @spec lease(t(), Hyper.Vm.Id.t(), Spec.t(), caps(), integer()) ::
+            {:ok, token(), t()} | {:error, :mem_exhausted | :disk_exhausted | :already_held}
+    def lease(%__MODULE__{entries: entries} = state, vm_id, spec, caps, expires_at) do
+      if Map.has_key?(entries, vm_id) do
+        {:error, :already_held}
+      else
+        with :ok <- fits(state, spec, caps) do
+          token = make_ref()
+          entry = {:leased, spec, token, expires_at}
+          {:ok, token, %{state | entries: Map.put(entries, vm_id, entry)}}
+        end
+      end
     end
 
-    @doc "Release `spec`'s reservation from the running total."
-    @spec cut(t(), Instance.Spec.t()) :: t()
-    def cut(s, spec) do
-      %{
-        s
-        | mem_allocated: s.mem_allocated - spec.mem,
-          disk_allocated: s.disk_allocated - spec.disk
-      }
+    @doc """
+    Convert `vm_id`'s lease into a reservation owned by `owner_ref`.
+
+    Never refuses on capacity — that was granted at lease time, and refusing
+    here would leave a booted VM with no accounting. Re-claiming an already
+    claimed vm_id rebinds it to the new owner (a `:transient` VM restart).
+    """
+    @spec claim(t(), Hyper.Vm.Id.t(), reference()) :: {:ok, t()} | {:error, :no_lease}
+    def claim(%__MODULE__{entries: entries} = state, vm_id, owner_ref) do
+      case Map.fetch(entries, vm_id) do
+        {:ok, entry} ->
+          entry = {:claimed, spec_of(entry), owner_ref}
+          {:ok, %{state | entries: Map.put(entries, vm_id, entry)}}
+
+        :error ->
+          {:error, :no_lease}
+      end
+    end
+
+    @doc """
+    Release `vm_id`'s lease, if `token` still matches it.
+
+    A no-op on a claimed entry and on a lease that has since been re-issued with
+    a fresh token, so a placing caller's late `drop` can never take capacity out
+    from under a live VM.
+    """
+    @spec drop(t(), Hyper.Vm.Id.t(), token()) :: t()
+    def drop(%__MODULE__{entries: entries} = state, vm_id, token) do
+      case Map.fetch(entries, vm_id) do
+        {:ok, {:leased, _spec, ^token, _expires_at}} ->
+          %{state | entries: Map.delete(entries, vm_id)}
+
+        _other ->
+          state
+      end
+    end
+
+    @doc """
+    Turn `vm_id`'s reservation back into a lease expiring at `expires_at`.
+
+    Called when a claimed owner dies. The capacity stays held for the grace
+    window so a `:transient` restart can re-claim it, and the fresh token
+    invalidates any `drop` still in flight from the original placement.
+    """
+    @spec release(t(), Hyper.Vm.Id.t(), integer()) :: t()
+    def release(%__MODULE__{entries: entries} = state, vm_id, expires_at) do
+      case Map.fetch(entries, vm_id) do
+        {:ok, {:claimed, spec, _owner_ref}} ->
+          entry = {:leased, spec, make_ref(), expires_at}
+          %{state | entries: Map.put(entries, vm_id, entry)}
+
+        _other ->
+          state
+      end
     end
 
-    @doc "Record that monitor `ref` owns `spec`'s reservation."
-    @spec track(t(), reference(), Hyper.Vm.Instance.Spec.t()) :: t()
-    def track(s, ref, spec), do: %{s | reservations: Map.put(s.reservations, ref, spec)}
+    @doc "Drop every lease whose deadline has passed. Reservations are untouched."
+    @spec expire(t(), integer()) :: {[Hyper.Vm.Id.t()], t()}
+    def expire(%__MODULE__{entries: entries} = state, now) do
+      expired =
+        for {vm_id, {:leased, _spec, _token, expires_at}} <- entries,
+            expires_at <= now,
+            do: vm_id
+
+      {Enum.sort(expired), %{state | entries: Map.drop(entries, expired)}}
+    end
 
-    @doc "Drop monitor `ref`, returning the spec it owned (or nil) and the new state."
-    @spec untrack(t(), reference()) :: {Hyper.Vm.Instance.Spec.t() | nil, t()}
-    def untrack(s, ref) do
-      {spec, rest} = Map.pop(s.reservations, ref)
-      {spec, %{s | reservations: rest}}
+    @doc "The vm_id whose reservation `owner_ref` owns, or nil."
+    @spec vm_id_for_owner(t(), reference()) :: Hyper.Vm.Id.t() | nil
+    def vm_id_for_owner(%__MODULE__{entries: entries}, owner_ref) do
+      Enum.find_value(entries, fn
+        {vm_id, {:claimed, _spec, ^owner_ref}} -> vm_id
+        _other -> nil
+      end)
+    end
+
+    @spec fits(t(), Spec.t(), caps()) :: :ok | {:error, :mem_exhausted | :disk_exhausted}
+    defp fits(state, spec, caps) do
+      %{mem: mem, disk: disk} = allocated(state)
+
+      cond do
+        mem + spec.mem > caps.mem -> {:error, :mem_exhausted}
+        disk + spec.disk > caps.disk -> {:error, :disk_exhausted}
+        true -> :ok
+      end
     end
+
+    @spec spec_of(entry()) :: Spec.t()
+    defp spec_of({:leased, spec, _token, _expires_at}), do: spec
+    defp spec_of({:claimed, spec, _owner_ref}), do: spec
   end
 
   # Client API
@@ -78,113 +198,168 @@ defmodule Hyper.Node.Budget.Hard do
     GenServer.start_link(__MODULE__, opts, name: __MODULE__)
   end
 
-  @doc "Can this node run the given vm spec? `:ok` if yes, `{:error, reason}` otherwise."
-  @spec can_run(Instance.Spec.t()) :: :ok | {:error, term()}
-  @decorate with_span("Hyper.Node.Budget.Hard.can_run", include: [:vm_spec])
-  def can_run(vm_spec) do
-    GenServer.call(__MODULE__, {:can_run, vm_spec})
-  end
-
   @doc """
-  Reserve `vm_spec`'s budget, run `callable`, and release the budget afterwards.
+  Grant `spec`'s capacity to `vm_id` provisionally, held against the calling
+  process and expiring after `boot_lease_ttl`.
 
-  Returns `callable`'s value if the reservation succeeds, or `{:error, reason}`
-  if the node cannot fit the spec. The budget is released even if `callable`
-  raises.
+  Returns a token for `drop/2`. Refuses if `spec` does not fit what is left.
   """
-  @spec with_budget(Instance.Spec.t(), (-> result)) :: result | {:error, term()}
-        when result: var
-  @decorate with_span("Hyper.Node.Budget.Hard.with_budget", include: [:vm_spec])
-  def with_budget(vm_spec, callable) do
-    with :ok <- GenServer.call(__MODULE__, {:ingest, vm_spec}) do
-      try do
-        callable.()
-      after
-        GenServer.call(__MODULE__, {:egress, vm_spec})
-      end
-    end
-  end
+  @spec lease(Hyper.Vm.Id.t(), Instance.Spec.t()) :: {:ok, State.token()} | {:error, term()}
+  @decorate with_span("Hyper.Node.Budget.Hard.lease", include: [:vm_id, :spec])
+  def lease(vm_id, spec), do: GenServer.call(__MODULE__, {:lease, vm_id, spec, self()})
 
   @doc """
-  Reserve `spec`'s budget for the lifetime of `owner`.
-
-  Atomic: refuses (`{:error, reason}`) if `spec` does not fit remaining headroom.
-  On success the reservation releases automatically when `owner` dies.
+  Convert `vm_id`'s lease into a reservation owned by `owner`, released when
+  `owner` dies. Never refuses on capacity.
   """
-  @spec reserve(Instance.Spec.t(), pid()) :: :ok | {:error, term()}
-  @decorate with_span("Hyper.Node.Budget.Hard.reserve", include: [:spec])
-  def reserve(spec, owner), do: GenServer.call(__MODULE__, {:reserve, spec, owner})
+  @spec claim(Hyper.Vm.Id.t(), pid()) :: :ok | {:error, :no_lease}
+  @decorate with_span("Hyper.Node.Budget.Hard.claim", include: [:vm_id])
+  def claim(vm_id, owner), do: GenServer.call(__MODULE__, {:claim, vm_id, owner})
 
-  @doc "Configured caps minus what is currently reserved."
+  @doc "Release the lease `token` identifies. A no-op once the VM has claimed it."
+  @spec drop(Hyper.Vm.Id.t(), State.token()) :: :ok
+  @decorate with_span("Hyper.Node.Budget.Hard.drop", include: [:vm_id])
+  def drop(vm_id, token), do: GenServer.call(__MODULE__, {:drop, vm_id, token})
+
+  @doc "Configured caps minus what is currently leased or reserved."
   @spec headroom() :: %{mem: Unit.Information.t(), disk: Unit.Information.t()}
   @decorate with_span("Hyper.Node.Budget.Hard.headroom")
   def headroom, do: GenServer.call(__MODULE__, :headroom)
 
   # Server callbacks
 
-  @impl true
-  def init(_opts) do
-    {:ok, State.zero()}
+  defmodule Server do
+    @moduledoc false
+    @type t :: %__MODULE__{
+            ledger: State.t(),
+            leasers: %{reference() => {Hyper.Vm.Id.t(), State.token()}}
+          }
+    defstruct ledger: nil, leasers: %{}
   end
 
   @impl true
-  def handle_call({:can_run, spec}, _from, state) do
-    {:reply, fits(state, spec), state}
-  end
+  def init(_opts), do: {:ok, %Server{ledger: State.new()}}
 
   @impl true
-  def handle_call({:ingest, spec}, _from, state) do
-    case fits(state, spec) do
-      :ok -> {:reply, :ok, State.bump(state, spec)}
-      {:error, _} = err -> {:reply, err, state}
+  def handle_call({:lease, vm_id, spec, leaser}, _from, s) do
+    ttl_ms = Unit.Time.as_ms(Config.get().boot_lease_ttl)
+
+    case State.lease(s.ledger, vm_id, spec, caps(), now_ms() + ttl_ms) do
+      {:ok, token, ledger} ->
+        ref = Process.monitor(leaser)
+        _ = Process.send_after(self(), :sweep, ttl_ms)
+        republish()
+
+        {:reply, {:ok, token},
+         %{s | ledger: ledger, leasers: Map.put(s.leasers, ref, {vm_id, token})}}
+
+      {:error, _reason} = err ->
+        {:reply, err, s}
     end
   end
 
   @impl true
-  def handle_call({:egress, spec}, _from, state) do
-    {:reply, :ok, State.cut(state, spec)}
+  def handle_call({:claim, vm_id, owner}, _from, s) do
+    owner_ref = Process.monitor(owner)
+
+    case State.claim(s.ledger, vm_id, owner_ref) do
+      {:ok, ledger} ->
+        # `claim` only ever moves an entry between kinds, never what it adds up
+        # to (the conservation law in HardStatePropertiesTest), so the
+        # NodeState a republish here would gossip is always identical to what
+        # `lease` already published. Only publish when that stops being true.
+        _ = if State.allocated(ledger) != State.allocated(s.ledger), do: republish()
+        {:reply, :ok, %{s | ledger: ledger, leasers: forget_leaser(s.leasers, vm_id)}}
+
+      {:error, :no_lease} = err ->
+        Process.demonitor(owner_ref, [:flush])
+        {:reply, err, s}
+    end
   end
 
   @impl true
-  def handle_call({:reserve, spec, owner}, _from, state) do
-    case fits(state, spec) do
-      :ok ->
-        ref = Process.monitor(owner)
-        state = state |> State.bump(spec) |> State.track(ref, spec)
-        republish()
-        {:reply, :ok, state}
-
-      {:error, _} = err ->
-        {:reply, err, state}
+  def handle_call({:drop, vm_id, token}, _from, s) do
+    ledger = State.drop(s.ledger, vm_id, token)
+
+    # A drop carrying a stale token changes nothing, and must not retire the
+    # leaser's monitor — that monitor is still the lease's release path.
+    if ledger == s.ledger do
+      {:reply, :ok, s}
+    else
+      republish()
+      {:reply, :ok, %{s | ledger: ledger, leasers: forget_leaser(s.leasers, vm_id)}}
     end
   end
 
   @impl true
-  def handle_call(:headroom, _from, state) do
-    config = Config.get()
+  def handle_call(:headroom, _from, s) do
+    caps = caps()
+    %{mem: mem, disk: disk} = State.allocated(s.ledger)
+    {:reply, %{mem: caps.mem - mem, disk: caps.disk - disk}, s}
+  end
 
-    headroom = %{
-      mem: config.mem_max - state.mem_allocated,
-      disk: config.disk_max - state.disk_allocated
-    }
+  @impl true
+  def handle_info({:DOWN, ref, :process, _pid, _reason}, s), do: {:noreply, handle_down(s, ref)}
 
-    {:reply, headroom, state}
+  @impl true
+  def handle_info(:sweep, s) do
+    {expired, ledger} = State.expire(s.ledger, now_ms())
+    _ = if expired != [], do: republish()
+    {:noreply, %{s | ledger: ledger, leasers: forget_leasers(s.leasers, expired)}}
   end
 
   @impl true
-  def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do
-    case State.untrack(state, ref) do
-      {nil, state} ->
-        {:noreply, state}
+  def handle_info(_msg, s), do: {:noreply, s}
+
+  # A :DOWN is either the process that took a lease, or the owner of a claimed
+  # reservation. A leaser's death drops its lease outright; an owner's death
+  # converts the reservation back into a grace lease so a `:transient` restart
+  # can re-claim the same capacity.
+  @spec handle_down(Server.t(), reference()) :: Server.t()
+  defp handle_down(s, ref) do
+    case Map.pop(s.leasers, ref) do
+      {{vm_id, token}, leasers} ->
+        ledger = State.drop(s.ledger, vm_id, token)
+        _ = if ledger != s.ledger, do: republish()
+        %{s | ledger: ledger, leasers: leasers}
+
+      {nil, _leasers} ->
+        release_owner(s, ref)
+    end
+  end
+
+  @spec release_owner(Server.t(), reference()) :: Server.t()
+  defp release_owner(s, ref) do
+    case State.vm_id_for_owner(s.ledger, ref) do
+      nil ->
+        s
 
-      {spec, state} ->
+      vm_id ->
+        grace_ms = Unit.Time.as_ms(Config.get().restart_grace)
+        _ = Process.send_after(self(), :sweep, grace_ms)
         republish()
-        {:noreply, State.cut(state, spec)}
+        %{s | ledger: State.release(s.ledger, vm_id, now_ms() + grace_ms)}
     end
   end
 
-  @impl true
-  def handle_info(_msg, state), do: {:noreply, state}
+  @spec forget_leaser(%{reference() => {Hyper.Vm.Id.t(), State.token()}}, Hyper.Vm.Id.t()) ::
+          %{reference() => {Hyper.Vm.Id.t(), State.token()}}
+  defp forget_leaser(leasers, vm_id), do: forget_leasers(leasers, [vm_id])
+
+  @spec forget_leasers(%{reference() => {Hyper.Vm.Id.t(), State.token()}}, [Hyper.Vm.Id.t()]) ::
+          %{reference() => {Hyper.Vm.Id.t(), State.token()}}
+  defp forget_leasers(leasers, vm_ids) do
+    gone = MapSet.new(vm_ids)
+
+    Enum.reduce(leasers, %{}, fn {ref, {vm_id, token}}, acc ->
+      if MapSet.member?(gone, vm_id) do
+        Process.demonitor(ref, [:flush])
+        acc
+      else
+        Map.put(acc, ref, {vm_id, token})
+      end
+    end)
+  end
 
   # Re-publish this node's NodeState after any reservation change. Guarded so
   # Hard runs standalone when no advertiser is present.
@@ -196,15 +371,12 @@ defmodule Hyper.Node.Budget.Hard do
     end
   end
 
-  # Whether reserving `spec` keeps both totals within the node's configured caps.
-  @spec fits(State.t(), Instance.Spec.t()) :: :ok | {:error, term()}
-  defp fits(state, spec) do
+  @spec caps() :: State.caps()
+  defp caps do
     config = Config.get()
-
-    cond do
-      state.mem_allocated + spec.mem > config.mem_max -> {:error, :mem_exhausted}
-      state.disk_allocated + spec.disk > config.disk_max -> {:error, :disk_exhausted}
-      true -> :ok
-    end
+    %{mem: config.mem_max, disk: config.disk_max}
   end
+
+  @spec now_ms() :: integer()
+  defp now_ms, do: System.monotonic_time(:millisecond)
 end
diff --git a/lib/hyper/node/budget/node_state.ex b/lib/hyper/node/budget/node_state.ex
index aa9af303..0d15a9d1 100644
--- a/lib/hyper/node/budget/node_state.ex
+++ b/lib/hyper/node/budget/node_state.ex
@@ -5,7 +5,7 @@ defmodule Hyper.Node.Budget.NodeState do
 
   Approximate by design: hard headroom is exact at publish time but soft load is
   an EWMA that drifts and the record gossips with lag. The authoritative decision
-  is always the owning node's `Hyper.Node.Budget.admit/2`. Each record carries the
+  is always the owning node's `Hyper.Node.Budget.lease/2`. Each record carries the
   node's load *and* its ceilings, so a scheduler anywhere can evaluate fit without
   knowing the target's config or core count.
   """
@@ -73,7 +73,7 @@ defmodule Hyper.Node.Budget.NodeState do
   Whether this snapshot's node can hold `spec`: hard memory/disk headroom plus the
   soft cpu/disk-bw/net-bw load ceilings. A pure predicate over the published
   snapshot; the authoritative check is still the owning node's
-  `Hyper.Node.Budget.admit/2`.
+  `Hyper.Node.Budget.lease/2`.
   """
   @spec fits?(t(), Spec.t()) :: boolean()
   def fits?(state, spec), do: hard_fits?(state, spec) and soft_fits?(state, spec)
diff --git a/lib/hyper/node/fire_vmm.ex b/lib/hyper/node/fire_vmm.ex
index f33d70bf..11211299 100644
--- a/lib/hyper/node/fire_vmm.ex
+++ b/lib/hyper/node/fire_vmm.ex
@@ -20,6 +20,13 @@ defmodule Hyper.Node.FireVMM do
        final usage window before the daemon removes the cgroup.
 
   Strategy is `:one_for_one`: the four children are restarted independently.
+
+  `init/1` claims this VM's budget lease (`Hyper.Node.Budget.claim/2`) before
+  starting any child, so a VM that reaches `{:ok, pid}` is always accounted.
+  That path needs a real jail and a live budget server, so it is covered by the
+  `:integration` suite (`test/e2e/vm_lifecycle_test.exs`) rather than a unit
+  test; the hermetic half — that `try_run/3` does not itself own the
+  reservation — is pinned by `test/hyper/node/try_run_admission_test.exs`.
   """
 
   use Supervisor
@@ -77,34 +84,36 @@ defmodule Hyper.Node.FireVMM do
 
   @impl true
   def init(opts) do
-    # Self-register the cluster routing entry here rather than via a start name;
-    # see `Hyper.Cluster.Routing.register_self/1`. A fresh random vm_id never
-    # collides, so `:already_registered` only happens against a stale dead
-    # incarnation - decline the start and let the supervisor retry clean.
-    case Hyper.Cluster.Routing.register_self({opts.vm_id, :supervisor}) do
-      :ok ->
-        children = [
-          # Client must be registered before Core: Core starts the State machine,
-          # which calls Client.run while waiting for the daemon's API. Client
-          # depends only on vm_id (an independent peer), so no reverse dependency.
-          {Client, %Client.Opts{vm_id: opts.vm_id}},
-          {Core, opts},
-          {Relay,
-           %{
-             vm_id: opts.vm_id,
-             vsock_uds: Jailer.host_vsock(opts.vm_id),
-             listen_path: Agent.relay_socket_path(opts.vm_id)
-           }},
-          # Last on purpose: children stop in reverse start order, so the meter
-          # stops first at teardown and flushes its final usage window while
-          # Core's Daemon (and the cgroup it removes) is still alive.
-          {Meter, %Meter.Opts{vm_id: opts.vm_id, cgroup_dir: Jailer.cgroup_dir(opts.vm_id)}}
-        ]
-
-        Supervisor.init(children, strategy: :one_for_one)
+    # Self-register the cluster routing entry and claim this VM's budget here
+    # rather than from the placing caller. `DynamicSupervisor.start_child`
+    # returns only after this function does, so a successful start already
+    # implies both — which is what leaves no window in which a booted VM is
+    # unaccounted. A fresh random vm_id never collides, so `:already_registered`
+    # only happens against a stale dead incarnation; `:no_lease` means nothing
+    # admitted this VM. Decline the start either way.
+    with :ok <- Hyper.Cluster.Routing.register_self({opts.vm_id, :supervisor}),
+         :ok <- Hyper.Node.Budget.claim(opts.vm_id, self()) do
+      children = [
+        # Client must be registered before Core: Core starts the State machine,
+        # which calls Client.run while waiting for the daemon's API. Client
+        # depends only on vm_id (an independent peer), so no reverse dependency.
+        {Client, %Client.Opts{vm_id: opts.vm_id}},
+        {Core, opts},
+        {Relay,
+         %{
+           vm_id: opts.vm_id,
+           vsock_uds: Jailer.host_vsock(opts.vm_id),
+           listen_path: Agent.relay_socket_path(opts.vm_id)
+         }},
+        # Last on purpose: children stop in reverse start order, so the meter
+        # stops first at teardown and flushes its final usage window while
+        # Core's Daemon (and the cgroup it removes) is still alive.
+        {Meter, %Meter.Opts{vm_id: opts.vm_id, cgroup_dir: Jailer.cgroup_dir(opts.vm_id)}}
+      ]
 
-      {:error, _} ->
-        :ignore
+      Supervisor.init(children, strategy: :one_for_one)
+    else
+      {:error, _reason} -> :ignore
     end
   end
 
diff --git a/lib/hyper/node/img/mutable.ex b/lib/hyper/node/img/mutable.ex
index c722205a..016ea284 100644
--- a/lib/hyper/node/img/mutable.ex
+++ b/lib/hyper/node/img/mutable.ex
@@ -1,25 +1,18 @@
 defmodule Hyper.Node.Img.Mutable do
   @moduledoc """
-  The per-VM mutable rootfs. On start it activates (or reuses) the image's
-  read-only `Img.Server`, takes a reference on it, reads the composed device's
-  size, and asks the node `ThinPool` for a thin volume with that device as a
-  read-only external origin. `blk_path/1` is the mutable host device the VM
-  boots from (staged into the jail by `mknod` from this path).
-
-  Mutable layers live in their own `DynamicSupervisor`, separate from the shared
-  read-only `Img.Server`s; the firecracker VM is handed this layer directly and
-  cannot be booted from a bare `Hyper.Img`.
-
-  Monitor-refcounted like `Img.Server`/`Layer.Server`: the VM supervisor holds
-  it; when the last holder dies it idle-reaps, destroying its thin volume in
-  `terminate/2` and releasing the image (which, if it was the last holder, tears
-  down the RO chain in turn).
+  Module which manages the lifecycle of a single mutable filesystem layer. A mutable layer is a
+  layer on top of a dm-thin chain of immutable layers.
+
+  For example, you can have a base immutable layer `ubuntu-20.04`, another immutable layer which
+  installs `curl`. When the user creates a new VM, Hyper will stack the `ubuntu-20.04` base, the
+  `curl` layer and a new _mutable_ layer which the user can edit (write to, effectively).
+
+  This is all copy-on-write.
   """
 
-  # `:temporary` is load-bearing: on idle this server destroys its per-VM thin
-  # volume in `terminate/2`, so a `:permanent` restart would resurrect the dm
-  # device it just tore down. See the reconciliation TODO in `Hyper.Node.Reaper`
-  # for why coupling resource lifetime to process lifetime is a smell.
+  # `:temporary` is load-bearing: on idle this server destroys its per-VM thin volume in
+  # `terminate/2`, so a `:permanent` restart would resurrect the dm device it just tore down. See
+  # the reconciliation TODO in `Hyper.Node.Reaper`.
   use GenServer, restart: :temporary
 
   alias Hyper.Node.Img
diff --git a/lib/hyper/vm.ex b/lib/hyper/vm.ex
index 219a9c3a..523efd5c 100644
--- a/lib/hyper/vm.ex
+++ b/lib/hyper/vm.ex
@@ -78,7 +78,7 @@ defmodule Hyper.Vm do
 
   @doc false
   # Which fast_fork refusals mean "no room on the parent's node" — the errors
-  # try_run/Budget.admit emits, plus the scheduler's aggregate — as opposed to
+  # try_run/Budget.lease emits, plus the scheduler's aggregate — as opposed to
   # faults that re-placement cannot fix.
   @spec capacity_error?(term()) :: boolean()
   def capacity_error?(reason), do: reason in @capacity_errors
diff --git a/test/e2e/vm_lifecycle_test.exs b/test/e2e/vm_lifecycle_test.exs
index 9d3dd852..968dd636 100644
--- a/test/e2e/vm_lifecycle_test.exs
+++ b/test/e2e/vm_lifecycle_test.exs
@@ -5,6 +5,11 @@ defmodule Hyper.E2e.VmLifecycleTest do
   - `OciLoader.load/1` of a real registry image yields a bootable image id;
   - `create_vm/1` boots a guest whose per-VM writable dm volume
     (`Mutable.dm_name/1`) exists while the VM runs;
+  - the boot actually claims the node's hard budget: headroom drops by the
+    instance type's `mem` while the VM runs, and comes back once it is stopped
+    and `restart_grace` has elapsed — the one assertion this branch's central
+    invariant (`Hyper.Node.FireVMM.init/1` claiming its lease) actually needs
+    and did not have;
   - the guest agent answers `exec` with the command's captured output;
   - `stop_image_vm/1` reclaims the writable volume (no dm leak);
   - stopping flushes a final metering window: the VM's recorded compute
@@ -15,12 +20,15 @@ defmodule Hyper.E2e.VmLifecycleTest do
   provisioned per docs/cookbook/install.md (CI: the `integration` job).
   """
   use ExUnit.Case, async: false
+  use Unit.Operators
 
   import Ecto.Query
   import Hyper.E2e
 
   alias Hyper.Img.Db.Repo
   alias Hyper.Metering.Usage
+  alias Hyper.Node.Budget.Hard
+  alias Hyper.Vm.Instance
   alias Unit.Time
 
   @moduletag :integration
@@ -33,6 +41,9 @@ defmodule Hyper.E2e.VmLifecycleTest do
   test "load -> create_vm -> exec -> stop reclaims the VM's dm volume" do
     assert {:ok, img_id} = Hyper.Img.OciLoader.load(@image)
 
+    mem_before = Hard.headroom().mem
+    instance_mem = Instance.spec(:micro).mem
+
     # :micro, not the :base default — :base asks for 32 GiB of disk budget,
     # which the default node budget (4 GiB) refuses with :no_capacity on the
     # small CI runner.
@@ -49,6 +60,12 @@ defmodule Hyper.E2e.VmLifecycleTest do
     assert MapSet.member?(dm_devices(), rw_dev),
            "expected writable dm volume #{rw_dev} while the VM is running"
 
+    # `create_vm/1` only returns once `FireVMM.init/1` has claimed this VM's
+    # lease (`DynamicSupervisor.start_child` waits on `init/1`), so the drop is
+    # visible immediately — no poll needed here, unlike the release below.
+    assert Hard.headroom().mem == mem_before - instance_mem,
+           "hard budget headroom did not drop by the booted instance's mem"
+
     assert {:ok, %{stdout: out, exit_code: 0}} =
              await_exec(vm, ["/bin/echo", "hello from guest"])
 
@@ -59,6 +76,14 @@ defmodule Hyper.E2e.VmLifecycleTest do
     assert poll_until(fn -> not MapSet.member?(dm_devices(), rw_dev) end, :timer.seconds(90)),
            "writable dm volume #{rw_dev} leaked after stop_image_vm"
 
+    # A clean stop still turns the reservation back into a lease for
+    # `restart_grace` (so a `:transient` FireVMM restart could reclaim it) —
+    # it only actually expires once that grace elapses. 90s matches the other
+    # polls in this test: generous headroom over the default 5s grace for a
+    # runner busy with the rest of this job's E2E fleet.
+    assert poll_until(fn -> Hard.headroom().mem == mem_before end, :timer.seconds(90)),
+           "hard budget headroom did not return after stop_image_vm + restart_grace"
+
     # The Meter is the FireVMM supervisor's LAST child: at stop it terminates
     # first and flushes a final usage window while the cgroup still exists.
     # Because the meter baselines its accumulator at init (meter start), the
diff --git a/test/hyper/cfg/budget_test.exs b/test/hyper/cfg/budget_test.exs
index 9fa64d92..44d1d7b3 100644
--- a/test/hyper/cfg/budget_test.exs
+++ b/test/hyper/cfg/budget_test.exs
@@ -30,7 +30,9 @@ defmodule Hyper.Cfg.BudgetTest do
              disk_bw_cap: Unit.Bandwidth.gibps(1),
              disk_bw_max_load: 0.8,
              net_bw_cap: Unit.Bandwidth.gibps(1),
-             net_bw_max_load: 0.8
+             net_bw_max_load: 0.8,
+             boot_lease_ttl: Unit.Time.s(300),
+             restart_grace: Unit.Time.s(5)
            }
   end
 
@@ -59,13 +61,22 @@ defmodule Hyper.Cfg.BudgetTest do
     assert config.cpu_max_cap == nil
   end
 
+  test "the lease timings are overridable from the [budget] TOML table" do
+    Toml.put_cache(%{"budget" => %{"boot_lease_ttl" => "30s", "restart_grace" => "250ms"}})
+
+    assert {:ok, config} = Budget.load()
+    assert config.boot_lease_ttl == Unit.Time.s(30)
+    assert config.restart_grace == Unit.Time.ms(250)
+  end
+
   # Refusal contracts on bad budget values: each must fail load with a specific
   # error naming the offending key, never silently coerce or crash. Table-driven:
   # one assertion shape, rows differ only in the bad env and expected error.
   @bad_budgets [
     {[mem_max: 123], {:error, {:bad_value, :mem_max, 123}}},
     {[mem_max: "notabytes"], {:error, {:bad_value, :mem_max, "notabytes"}}},
-    {[cpu_max_load: "high"], {:error, {:not_a_number, :cpu_max_load, "high"}}}
+    {[cpu_max_load: "high"], {:error, {:not_a_number, :cpu_max_load, "high"}}},
+    {[boot_lease_ttl: "forever"], {:error, {:bad_value, :boot_lease_ttl, "forever"}}}
   ]
 
   for {env, expected} <- @bad_budgets do
diff --git a/test/hyper/node/budget/hard_state_properties_test.exs b/test/hyper/node/budget/hard_state_properties_test.exs
index 1ebf0b65..87a08f72 100644
--- a/test/hyper/node/budget/hard_state_properties_test.exs
+++ b/test/hyper/node/budget/hard_state_properties_test.exs
@@ -1,67 +1,358 @@
 defmodule Hyper.Node.Budget.HardStatePropertiesTest do
   @moduledoc """
-  Algebraic laws of the pure `Hyper.Node.Budget.Hard.State` accumulator: `cut`
-  is the inverse of `bump` on every spec, bumps accumulate additively (so the
-  running total is order-independent), and `track`/`untrack` round-trip a
-  reservation by reference. The example tests spot-check single values; these
-  pin the laws across the domain.
+  Laws of the pure ledger core, `Hyper.Node.Budget.Hard.State`.
+
+  The ledger holds two kinds of entry against one node's caps:
+
+    * a **lease** — capacity granted to a boot that has not happened yet, held
+      against a placing caller, carrying an expiry;
+    * a **reservation** — capacity held against a live VM, created by `claim`ing
+      an existing lease.
+
+  Both occupy capacity. That is the contract the whole design rests on: a VM
+  that is *about to exist* must be as visible to admission as one that already
+  does, or a concurrent herd boots against headroom nothing has taken.
+
+  The laws pinned here:
+
+    * **Inverse** — `drop` undoes `lease` exactly; `release` then `claim`
+      restores the allocation unchanged.
+    * **Conservation** — `claim` moves an entry between kinds without changing
+      what is allocated; allocation is always the sum of live entries, in any
+      order.
+    * **Refusal** — `lease` refuses exactly when the spec would cross a cap, and
+      leases exhaust caps identically to reservations. `claim` of an unknown
+      vm_id is refused rather than silently reserving.
+    * **Never under-reserve** — no operation releases capacity out from under a
+      claimed reservation: not `drop`, not a stale token, not `expire`.
+
+  The last family is the important one. Over-reserving costs capacity and heals;
+  under-reserving means a VM is running that the ledger cannot see, which is how
+  the host reaches the OOM killer with headroom to spare on paper.
   """
+
   use ExUnit.Case, async: true
   use ExUnitProperties
+  use Unit.Operators
 
   alias Hyper.Node.Budget.Hard.State
   alias Hyper.Vm.Instance.Spec
-  alias Unit.{Bandwidth, Information}
-
-  # Only `mem`/`disk` matter to bump/cut; the other fields are along for the ride.
-  defp spec do
-    gen all(mem_mib <- integer(0..1_000_000), disk_mib <- integer(0..1_000_000)) do
-      %Spec{
-        vcpus: 1,
-        mem: Information.mib(mem_mib),
-        disk: Information.mib(disk_mib),
-        disk_bw: Bandwidth.zero(),
-        net_bw: Bandwidth.zero()
-      }
+  alias Unit.Bandwidth
+  alias Unit.Information
+
+  # Expiries are monotonic milliseconds, compared by `expire/2` against a `now`
+  # the caller supplies — so every law here is pure, with no clock and no sleep.
+  @never 1_000_000_000
+  @long_ago -1
+
+  describe "inverse laws" do
+    property "dropping a lease restores the allocation it took" do
+      check all({s, caps} <- spec_within_caps(), id <- vm_id()) do
+        empty = State.new()
+        {:ok, token, leased} = State.lease(empty, id, s, caps, @never)
+
+        assert State.allocated(State.drop(leased, id, token)) == State.allocated(empty)
+      end
+    end
+
+    property "releasing a claimed reservation and re-claiming it restores the allocation" do
+      check all({s, caps} <- spec_within_caps(), id <- vm_id()) do
+        {:ok, _token, leased} = State.lease(State.new(), id, s, caps, @never)
+        {:ok, claimed} = State.claim(leased, id, ref())
+        before = State.allocated(claimed)
+
+        rebound =
+          claimed
+          |> State.release(id, @never)
+          |> State.claim(id, ref())
+          |> ok!()
+
+        assert State.allocated(rebound) == before
+      end
+    end
+  end
+
+  describe "conservation laws" do
+    property "claiming a lease does not change what is allocated" do
+      check all({s, caps} <- spec_within_caps(), id <- vm_id()) do
+        {:ok, _token, leased} = State.lease(State.new(), id, s, caps, @never)
+        {:ok, claimed} = State.claim(leased, id, ref())
+
+        assert State.allocated(claimed) == State.allocated(leased)
+      end
+    end
+
+    property "allocation is the sum of live entries, whatever order they arrived in" do
+      check all({specs, caps} <- specs_within_caps(), shuffled <- shuffled(specs)) do
+        assert allocate_all(specs, caps) == allocate_all(shuffled, caps)
+
+        total_mem = specs |> Enum.map(&Information.as_bytes(&1.mem)) |> Enum.sum()
+        total_disk = specs |> Enum.map(&Information.as_bytes(&1.disk)) |> Enum.sum()
+
+        %{mem: mem, disk: disk} = allocate_all(specs, caps)
+        assert Information.as_bytes(mem) == total_mem
+        assert Information.as_bytes(disk) == total_disk
+      end
+    end
+  end
+
+  describe "refusal contract" do
+    property "a lease is refused exactly when it would cross a cap" do
+      check all({held, s, caps} <- held_plus_spec()) do
+        state = lease_all(held, caps)
+
+        %{mem: mem, disk: disk} = State.allocated(state)
+        fits? = mem + s.mem <= caps.mem and disk + s.disk <= caps.disk
+
+        case State.lease(state, "vm-candidate", s, caps, @never) do
+          {:ok, _token, _state} -> assert fits?
+          {:error, reason} -> assert not fits? and reason in [:mem_exhausted, :disk_exhausted]
+        end
+      end
+    end
+
+    property "leases exhaust caps exactly as reservations do" do
+      check all(%{mem: mem_mib, disk: disk_mib, count: k} <- sized_for_k()) do
+        caps = %{mem: Information.mib(mem_mib * k), disk: Information.mib(disk_mib * k)}
+        s = spec(mem_mib, disk_mib)
+
+        # k leases fill the node with nothing claimed at all.
+        state = lease_all(List.duplicate(s, k), caps)
+
+        assert {:error, reason} = State.lease(state, "vm-overflow", s, caps, @never)
+        assert reason in [:mem_exhausted, :disk_exhausted]
+      end
+    end
+
+    property "claiming a vm_id with no lease is refused, never silently reserved" do
+      check all(id <- vm_id()) do
+        empty = State.new()
+
+        assert {:error, :no_lease} = State.claim(empty, id, ref())
+        assert State.allocated(empty) == State.allocated(State.new())
+      end
+    end
+  end
+
+  describe "never under-reserve" do
+    property "dropping a claimed vm_id does not release its reservation" do
+      check all({s, caps} <- spec_within_caps(), id <- vm_id()) do
+        {:ok, token, leased} = State.lease(State.new(), id, s, caps, @never)
+        {:ok, claimed} = State.claim(leased, id, ref())
+
+        assert State.allocated(State.drop(claimed, id, token)) == State.allocated(claimed)
+      end
+    end
+
+    property "a stale token never drops the lease that replaced it" do
+      check all({s, caps} <- spec_within_caps(), id <- vm_id()) do
+        {:ok, stale, leased} = State.lease(State.new(), id, s, caps, @never)
+        {:ok, claimed} = State.claim(leased, id, ref())
+
+        # The owner died and the entry went back to a grace lease with a FRESH
+        # token. The placing caller's `drop`, arriving late, must not take it.
+        regraced = State.release(claimed, id, @never)
+
+        assert State.allocated(State.drop(regraced, id, stale)) == State.allocated(regraced)
+      end
+    end
+
+    property "expire removes only leases past their deadline, never a reservation" do
+      check all(mem_mib <- integer(1..512), disk_mib <- integer(1..512)) do
+        s = spec(mem_mib, disk_mib)
+        caps = %{mem: Information.mib(mem_mib * 2), disk: Information.mib(disk_mib * 2)}
+
+        # BOTH entries carry an expired deadline; only the unclaimed one may go.
+        # Claiming is what makes a deadline stop applying.
+        state =
+          State.new()
+          |> lease!("vm-expiring", s, caps, @long_ago)
+          |> lease!("vm-claimed", s, caps, @long_ago)
+          |> State.claim("vm-claimed", ref())
+          |> ok!()
+
+        {expired, swept} = State.expire(state, 0)
+
+        assert expired == ["vm-expiring"]
+        assert State.allocated(swept) == State.allocated(lease_all([s], caps))
+      end
+    end
+
+    property "leasing a vm_id that already holds capacity is refused, not overwritten" do
+      check all({s, caps} <- spec_within_caps(), id <- vm_id()) do
+        {:ok, _token, leased} = State.lease(State.new(), id, s, caps, @never)
+
+        assert {:error, :already_held} = State.lease(leased, id, s, caps, @never)
+
+        {:ok, claimed} = State.claim(leased, id, ref())
+        assert {:error, :already_held} = State.lease(claimed, id, s, caps, @never)
+      end
+    end
+
+    property "re-claiming a claimed vm_id rebinds it without double-counting" do
+      check all({s, caps} <- spec_within_caps(), id <- vm_id()) do
+        first = ref()
+        second = ref()
+
+        {:ok, _token, leased} = State.lease(State.new(), id, s, caps, @never)
+        {:ok, once} = State.claim(leased, id, first)
+        {:ok, twice} = State.claim(once, id, second)
+
+        assert State.allocated(twice) == State.allocated(once)
+        assert State.vm_id_for_owner(twice, second) == id
+        assert State.vm_id_for_owner(twice, first) == nil
+      end
+    end
+
+    property "no interleaving of operations exceeds the caps or loses an entry" do
+      check all({ops, s, caps} <- op_sequence()) do
+        {state, model} =
+          Enum.reduce(ops, {State.new(), %{}}, fn op, acc -> apply_op(op, s, caps, acc) end)
+
+        %{mem: mem, disk: disk} = State.allocated(state)
+
+        assert mem <= caps.mem, "leases + reservations exceeded the memory cap"
+        assert disk <= caps.disk, "leases + reservations exceeded the disk cap"
+
+        assert Information.as_bytes(mem) == map_size(model) * Information.as_bytes(s.mem)
+        assert Information.as_bytes(disk) == map_size(model) * Information.as_bytes(s.disk)
+      end
     end
   end
 
-  property "cut undoes bump for any spec" do
-    check all(s <- spec()) do
-      state = State.zero() |> State.bump(s) |> State.cut(s)
-      assert state.mem_allocated == Information.zero()
-      assert state.disk_allocated == Information.zero()
+  # The model mirrors only the BOOKKEEPING (which vm_ids hold capacity), never
+  # the fit rule: it records a lease when State grants one and skips it when
+  # State refuses. The cap invariant above is what actually tests the fit rule,
+  # so nothing here recomputes the implementation.
+  defp apply_op({:lease, id}, s, caps, {state, model}) do
+    case State.lease(state, id, s, caps, @never) do
+      {:ok, token, state} -> {state, Map.put(model, id, {:leased, token})}
+      {:error, _} -> {state, model}
     end
   end
 
-  property "the running total is the sum of bumped specs (hence order-independent)" do
-    check all(specs <- list_of(spec(), max_length: 20)) do
-      state = Enum.reduce(specs, State.zero(), fn s, acc -> State.bump(acc, s) end)
+  defp apply_op({:claim, id}, _s, _caps, {state, model}) do
+    case State.claim(state, id, ref()) do
+      {:ok, state} -> {state, Map.put(model, id, :claimed)}
+      {:error, :no_lease} -> {state, model}
+    end
+  end
 
-      total_mem = specs |> Enum.map(&Information.as_bytes(&1.mem)) |> Enum.sum()
-      total_disk = specs |> Enum.map(&Information.as_bytes(&1.disk)) |> Enum.sum()
+  defp apply_op({:drop, id}, _s, _caps, {state, model}) do
+    case Map.get(model, id) do
+      {:leased, token} -> {State.drop(state, id, token), Map.delete(model, id)}
+      _ -> {state, model}
+    end
+  end
+
+  defp op_sequence do
+    gen all(
+          mem_mib <- integer(1..64),
+          disk_mib <- integer(1..64),
+          slots <- integer(1..6),
+          # Unique by construction, not by rejection: `uniq_list_of` over a
+          # size-dependent generator raises TooManyDuplicatesError at small
+          # generation sizes rather than re-generating, which aborts the run.
+          ids <- map(integer(1..6), fn n -> Enum.map(1..n, &"vm-#{&1}") end),
+          ops <-
+            list_of(
+              tuple({member_of([:lease, :claim, :drop]), member_of(ids)}),
+              max_length: 30
+            )
+        ) do
+      s = spec(mem_mib, disk_mib)
+      caps = %{mem: Information.mib(mem_mib * slots), disk: Information.mib(disk_mib * slots)}
+      {ops, s, caps}
+    end
+  end
+
+  defp spec(mem_mib, disk_mib) do
+    %Spec{
+      vcpus: 1,
+      mem: Information.mib(mem_mib),
+      disk: Information.mib(disk_mib),
+      disk_bw: Bandwidth.zero(),
+      net_bw: Bandwidth.zero()
+    }
+  end
+
+  # Caps are built FROM the spec (spec + slack) rather than generated and
+  # filtered, so a fitting pair is produced by construction every time.
+  defp spec_within_caps do
+    gen all(
+          mem_mib <- integer(0..8192),
+          disk_mib <- integer(0..8192),
+          slack_mem <- integer(0..8192),
+          slack_disk <- integer(0..8192)
+        ) do
+      {spec(mem_mib, disk_mib),
+       %{
+         mem: Information.mib(mem_mib + slack_mem),
+         disk: Information.mib(disk_mib + slack_disk)
+       }}
+    end
+  end
 
-      assert Information.as_bytes(state.mem_allocated) == total_mem
-      assert Information.as_bytes(state.disk_allocated) == total_disk
+  defp specs_within_caps do
+    gen all(
+          pairs <- list_of(tuple({integer(0..512), integer(0..512)}), max_length: 12),
+          slack <- integer(0..1024)
+        ) do
+      specs = Enum.map(pairs, fn {m, d} -> spec(m, d) end)
+      mem = pairs |> Enum.map(&elem(&1, 0)) |> Enum.sum()
+      disk = pairs |> Enum.map(&elem(&1, 1)) |> Enum.sum()
+
+      {specs, %{mem: Information.mib(mem + slack), disk: Information.mib(disk + slack)}}
     end
   end
 
-  property "untrack returns exactly the spec track stored, leaving the rest unchanged" do
-    check all(s <- spec()) do
-      base = State.zero()
-      ref = make_ref()
-      tracked = State.track(base, ref, s)
-      assert {^s, rest} = State.untrack(tracked, ref)
-      assert rest.reservations == base.reservations
+  # A pre-held set plus a candidate spec, with caps that may or may not fit it —
+  # the boundary is what the refusal property is probing, so caps are NOT sized
+  # to guarantee a fit here.
+  defp held_plus_spec do
+    gen all(
+          held <- list_of(tuple({integer(0..256), integer(0..256)}), max_length: 8),
+          cand_mem <- integer(0..512),
+          cand_disk <- integer(0..512),
+          cap_mem <- integer(0..2048),
+          cap_disk <- integer(0..2048)
+        ) do
+      {Enum.map(held, fn {m, d} -> spec(m, d) end), spec(cand_mem, cand_disk),
+       %{mem: Information.mib(cap_mem), disk: Information.mib(cap_disk)}}
     end
   end
 
-  property "untrack of a ref that was never tracked yields nil and an unchanged state" do
-    check all(s <- spec()) do
-      ref = make_ref()
-      other = make_ref()
-      state = State.track(State.zero(), ref, s)
-      assert {nil, ^state} = State.untrack(state, other)
+  defp sized_for_k do
+    gen all(mem <- integer(1..512), disk <- integer(1..512), count <- integer(1..8)) do
+      %{mem: mem, disk: disk, count: count}
     end
   end
+
+  defp shuffled(list), do: map(constant(list), &Enum.shuffle/1)
+
+  defp vm_id, do: map(positive_integer(), &"vm-#{&1}")
+
+  defp ref, do: make_ref()
+
+  # Lease every spec that fits, ignoring refusals — the caller supplies caps
+  # that make the outcome meaningful.
+  defp lease_all(specs, caps) do
+    specs
+    |> Enum.with_index()
+    |> Enum.reduce(State.new(), fn {s, i}, acc ->
+      case State.lease(acc, "vm-held-#{i}", s, caps, @never) do
+        {:ok, _token, acc} -> acc
+        {:error, _} -> acc
+      end
+    end)
+  end
+
+  defp allocate_all(specs, caps), do: specs |> lease_all(caps) |> State.allocated()
+
+  defp lease!(state, id, s, caps, expires_at) do
+    {:ok, _token, state} = State.lease(state, id, s, caps, expires_at)
+    state
+  end
+
+  defp ok!({:ok, state}), do: state
 end
diff --git a/test/hyper/node/budget/hard_test.exs b/test/hyper/node/budget/hard_test.exs
new file mode 100644
index 00000000..3062c02e
--- /dev/null
+++ b/test/hyper/node/budget/hard_test.exs
@@ -0,0 +1,270 @@
+defmodule Hyper.Node.Budget.HardTest do
+  @moduledoc """
+  Process-level behaviour of `Hyper.Node.Budget.Hard` — the parts the pure
+  ledger laws (`HardStatePropertiesTest`) cannot reach: monitors, the expiry
+  sweep, and the grace that carries a reservation across a VM restart.
+
+  A lease is released by three independent mechanisms, each covering what the
+  others cannot:
+
+    * the **monitor** on the leasing process — a crash, in microseconds;
+    * the **TTL** — a leaser that is alive but wedged, which no monitor will
+      ever fire for (an untimed `:erpc.call` into a hung `dmsetup` is the real
+      case);
+    * reconciliation — out of scope here.
+
+  The TTL is deliberately generous in production: expiring a lease under a VM
+  that is still legitimately booting produces an unaccounted running VM, which
+  is the failure direction that reaches the OOM killer. Over-reserving only
+  parks capacity. These tests shorten it to keep the suite fast.
+  """
+
+  use ExUnit.Case, async: false
+  use Unit.Operators
+
+  alias Hyper.Node.Budget.Hard
+  alias Hyper.Vm.Instance.Spec
+  alias Unit.Bandwidth
+  alias Unit.Information
+
+  @vm_mem Information.mib(128)
+  @capacity 4
+
+  setup do
+    saved = :persistent_term.get(Hyper.Cfg.Budget, :unset)
+
+    on_exit(fn ->
+      case saved do
+        :unset -> :persistent_term.erase(Hyper.Cfg.Budget)
+        config -> :persistent_term.put(Hyper.Cfg.Budget, config)
+      end
+    end)
+
+    :ok
+  end
+
+  test "a lease that does not fit the node's caps is refused" do
+    start_budget()
+    fill_node()
+
+    assert {:error, :mem_exhausted} = Hard.lease("vm-overflow", spec())
+  end
+
+  test "headroom counts an unclaimed lease, not just claimed reservations" do
+    start_budget()
+    before = Hard.headroom().mem
+
+    assert {:ok, _token} = Hard.lease("vm-a", spec())
+
+    assert Hard.headroom().mem == before - @vm_mem
+  end
+
+  test "an unclaimed lease is released when the leasing process dies" do
+    start_budget()
+    before = Hard.headroom().mem
+
+    {leaser, {:ok, _token}} = lease_from_another_process("vm-a")
+    assert Hard.headroom().mem == before - @vm_mem
+
+    kill(leaser)
+
+    assert eventually(fn -> Hard.headroom().mem == before end)
+  end
+
+  test "an unclaimed lease is released when its ttl expires, though the leaser lives" do
+    start_budget(boot_lease_ttl: Unit.Time.ms(50))
+    before = Hard.headroom().mem
+
+    {leaser, {:ok, _token}} = lease_from_another_process("vm-wedged")
+    assert Hard.headroom().mem == before - @vm_mem
+
+    assert eventually(fn -> Hard.headroom().mem == before end, 200),
+           "a wedged leaser held its lease past the ttl"
+
+    assert Process.alive?(leaser), "the ttl must not depend on the leaser dying"
+  end
+
+  test "a claimed reservation survives the death of the process that leased it" do
+    start_budget()
+    before = Hard.headroom().mem
+
+    {leaser, {:ok, _token}} = lease_from_another_process("vm-a")
+    vm = spawn_idle()
+    assert :ok = Hard.claim("vm-a", vm)
+
+    kill(leaser)
+
+    # The placing caller is gone; the VM is not. Its capacity must stay held.
+    assert steadily(fn -> Hard.headroom().mem == before - @vm_mem end)
+  end
+
+  test "a claimed reservation is released once its owner dies and the grace elapses" do
+    start_budget(restart_grace: Unit.Time.ms(50))
+    before = Hard.headroom().mem
+
+    {_leaser, {:ok, _token}} = lease_from_another_process("vm-a")
+    vm = spawn_idle()
+    assert :ok = Hard.claim("vm-a", vm)
+
+    kill(vm)
+
+    assert eventually(fn -> Hard.headroom().mem == before end, 200)
+  end
+
+  test "capacity is not released while a dead owner is inside the restart grace" do
+    start_budget(restart_grace: Unit.Time.s(30))
+    before = Hard.headroom().mem
+
+    {_leaser, {:ok, _token}} = lease_from_another_process("vm-a")
+    vm = spawn_idle()
+    assert :ok = Hard.claim("vm-a", vm)
+    kill(vm)
+
+    # A `:transient` FireVMM restart briefly has no live owner. If the ledger
+    # dipped here, a competing placement could take capacity the restarting VM
+    # is about to reclaim — so a competing lease must still be refused.
+    fill_node(from: 1)
+
+    assert {:error, :mem_exhausted} = Hard.lease("vm-intruder", spec())
+    assert Hard.headroom().mem == Information.zero()
+    assert before != Information.zero()
+  end
+
+  test "re-claiming inside the restart grace rebinds the reservation to the new owner" do
+    start_budget(restart_grace: Unit.Time.ms(500))
+    before = Hard.headroom().mem
+
+    {_leaser, {:ok, _token}} = lease_from_another_process("vm-a")
+    first = spawn_idle()
+    assert :ok = Hard.claim("vm-a", first)
+    kill(first)
+
+    restarted = spawn_idle()
+    assert :ok = Hard.claim("vm-a", restarted)
+
+    # The re-claim turned the grace lease back into a reservation, so the grace
+    # deadline no longer applies: capacity is still held well past it. Were the
+    # entry still a lease, it would expire at 500ms and this would fail.
+    assert steadily(fn -> Hard.headroom().mem == before - @vm_mem end, 400)
+
+    # Only the NEW owner's death releases it. Had the claim rebound to the dead
+    # first owner's ref instead, this kill would match nothing and the capacity
+    # would never come back.
+    kill(restarted)
+    assert eventually(fn -> Hard.headroom().mem == before end, 300)
+  end
+
+  test "dropping a lease with its token releases the capacity immediately" do
+    start_budget()
+    before = Hard.headroom().mem
+
+    assert {:ok, token} = Hard.lease("vm-a", spec())
+    assert Hard.headroom().mem == before - @vm_mem
+
+    assert :ok = Hard.drop("vm-a", token)
+
+    assert Hard.headroom().mem == before
+  end
+
+  test "dropping after the VM has claimed does not release the reservation" do
+    start_budget()
+    before = Hard.headroom().mem
+
+    assert {:ok, token} = Hard.lease("vm-a", spec())
+    vm = spawn_idle()
+    assert :ok = Hard.claim("vm-a", vm)
+
+    # The placing caller's normal post-boot drop must be a no-op now.
+    assert :ok = Hard.drop("vm-a", token)
+
+    assert steadily(fn -> Hard.headroom().mem == before - @vm_mem end)
+  end
+
+  test "claiming a vm_id with no lease is refused" do
+    start_budget()
+
+    assert {:error, :no_lease} = Hard.claim("vm-never-leased", spawn_idle())
+  end
+
+  defp start_budget(overrides \\ []) do
+    :persistent_term.put(Hyper.Cfg.Budget, budget_config(overrides))
+    start_supervised!(Hard)
+  end
+
+  # Lease from a process the test can kill without taking itself down, so the
+  # monitor path is exercised for real rather than simulated.
+  defp lease_from_another_process(vm_id) do
+    test = self()
+
+    {:ok, pid} =
+      Task.start(fn ->
+        send(test, {:leased, Hard.lease(vm_id, spec())})
+        Process.sleep(:infinity)
+      end)
+
+    assert_receive {:leased, result}
+    {pid, result}
+  end
+
+  defp spawn_idle, do: spawn(fn -> Process.sleep(:infinity) end)
+
+  defp kill(pid) do
+    ref = Process.monitor(pid)
+    Process.exit(pid, :kill)
+    assert_receive {:DOWN, ^ref, :process, ^pid, _}
+    :ok
+  end
+
+  defp fill_node(opts \\ []) do
+    from = Keyword.get(opts, :from, 0)
+
+    for i <- from..(@capacity - 1) do
+      assert {:ok, _token} = Hard.lease("vm-fill-#{i}", spec())
+    end
+  end
+
+  defp eventually(fun, attempts \\ 50) do
+    cond do
+      fun.() -> true
+      attempts == 0 -> false
+      true -> Process.sleep(2) && eventually(fun, attempts - 1)
+    end
+  end
+
+  # The dual of `eventually`: the condition must hold for the whole window, so a
+  # release that is merely late still fails the test.
+  defp steadily(fun, attempts \\ 25) do
+    cond do
+      not fun.() -> false
+      attempts == 0 -> true
+      true -> Process.sleep(2) && steadily(fun, attempts - 1)
+    end
+  end
+
+  defp spec do
+    %Spec{
+      vcpus: 0.25,
+      mem: @vm_mem,
+      disk: Information.mib(1),
+      disk_bw: Bandwidth.mibps(1),
+      net_bw: Bandwidth.mibps(1)
+    }
+  end
+
+  # Memory is the only binding cap; everything else is set out of reach so a
+  # refusal can only ever mean "the memory ledger refused".
+  defp budget_config(overrides) do
+    %Hyper.Cfg.Budget{
+      mem_max: Information.mib(@capacity * Information.as_mib(@vm_mem)),
+      disk_max: Information.tib(1),
+      cpu_max_load: 1000.0,
+      cpu_max_cap: nil,
+      disk_bw_cap: Bandwidth.gibps(1000),
+      disk_bw_max_load: 1.0,
+      net_bw_cap: Bandwidth.gibps(1000),
+      net_bw_max_load: 1.0,
+      boot_lease_ttl: Keyword.get(overrides, :boot_lease_ttl, Unit.Time.s(300)),
+      restart_grace: Keyword.get(overrides, :restart_grace, Unit.Time.s(5))
+    }
+  end
+end
diff --git a/test/hyper/node/start_vm_ignore_test.exs b/test/hyper/node/start_vm_ignore_test.exs
new file mode 100644
index 00000000..71e8ad43
--- /dev/null
+++ b/test/hyper/node/start_vm_ignore_test.exs
@@ -0,0 +1,144 @@
+defmodule Hyper.Node.StartVmIgnoreTest do
+  @moduledoc """
+  Pins the merge-blocker fix in `Hyper.Node.start_vm_or_release/3`.
+
+  `Hyper.Node.FireVMM.init/1` declines a boot with `:ignore` when either step
+  of its `with` refuses: `Hyper.Cluster.Routing.register_self/1` finds the
+  vm_id already registered (a stale dead incarnation), or
+  `Hyper.Node.Budget.claim/2` has no lease left to claim (the lease expired,
+  or its holder died mid-boot) — either way before any jailer/cgroup child is
+  ever built. This test drives the budget-claim cause, since it needs no more
+  than a dropped lease to trigger. `DynamicSupervisor.start_child` propagates
+  that `:ignore` verbatim, and `start_vm_or_release/3` must turn it into
+  `{:error, _}` rather than crash on an unmatched case clause — a crash would
+  skip the uid and mutable-layer release the ordinary `{:error, _}` arm
+  performs, and `Hyper.Node.Users` has no other way to recover a leaked uid
+  (see `TODO.txt`).
+
+  Reachable hermetically because the decline happens before any real jail
+  work: a dropped lease, a bare `DynamicSupervisor`, and `Hyper.Cluster.Routing`
+  standing in for the app's supervision tree are enough to exercise the real
+  `Hyper.Node.start_vm/1` call site.
+  """
+
+  use ExUnit.Case, async: false
+
+  alias Hyper.Node.Budget.Hard
+  alias Hyper.Node.Users
+  alias Hyper.Vm.Instance.Spec, as: InstanceSpec
+  alias Unit.Bandwidth
+  alias Unit.Information
+
+  @vm_mem Information.mib(128)
+  @uid 100_000
+
+  defmodule FakeMutable do
+    @moduledoc false
+    # Stands in for `Hyper.Node.Img.Mutable`: the declined-start path only ever
+    # calls `release/1` on it. Reports each release to `reporter` so the test
+    # can assert the call actually happened, rather than merely that answering
+    # it didn't crash.
+    use GenServer
+
+    def start_link(reporter), do: GenServer.start_link(__MODULE__, reporter)
+
+    @impl true
+    def init(reporter), do: {:ok, reporter}
+
+    @impl true
+    def handle_call({:release, _pid}, _from, reporter) do
+      send(reporter, :mutable_released)
+      {:reply, :ok, reporter}
+    end
+  end
+
+  setup do
+    saved_budget = :persistent_term.get(Hyper.Cfg.Budget, :unset)
+    saved_toml = Hyper.Cfg.Toml.reload()
+
+    :persistent_term.put(Hyper.Cfg.Budget, budget_config())
+
+    Hyper.Cfg.Toml.put_cache(Map.put(saved_toml, "jails", %{"uid_gid_range" => [@uid, @uid]}))
+
+    on_exit(fn ->
+      case saved_budget do
+        :unset -> :persistent_term.erase(Hyper.Cfg.Budget)
+        config -> :persistent_term.put(Hyper.Cfg.Budget, config)
+      end
+
+      Hyper.Cfg.Toml.put_cache(saved_toml)
+    end)
+
+    _ = Application.ensure_all_started(:horde)
+
+    unless Process.whereis(Hyper.Cluster.Routing) do
+      case start_supervised(Hyper.Cluster.Routing) do
+        {:ok, _pid} -> :ok
+        {:error, {:already_started, _pid}} -> :ok
+      end
+    end
+
+    start_supervised!(Users)
+    start_supervised!(Hard)
+    start_supervised!({DynamicSupervisor, name: Hyper.Node.VMSupervisor, strategy: :one_for_one})
+
+    :ok
+  end
+
+  test "a lease dropped before FireVMM claims it declines the start without leaking the uid" do
+    vm_id = "vm-declined-#{System.unique_integer([:positive])}"
+
+    assert {:ok, token} = Hard.lease(vm_id, budget_spec())
+    assert :ok = Hard.drop(vm_id, token)
+
+    assert {:ok, uid} = Users.claim()
+    assert uid == @uid
+
+    {:ok, mutable} = start_supervised({FakeMutable, self()})
+
+    opts =
+      Hyper.Node.build_opts(
+        vm_id,
+        %Hyper.Vm.Spec{img_id: "img-fake", type: :micro, arch: :x86_64},
+        uid,
+        mutable,
+        "/dev/null"
+      )
+
+    assert {:error, :not_admitted} = Hyper.Node.start_vm_or_release(opts, uid, mutable)
+
+    # The mutable layer's hold must actually be dropped, not just survive
+    # answering a call: `acquire_or_release/2` took this hold before the boot
+    # even reached `start_vm_or_release/3`, and nothing else on this path
+    # drops it.
+    assert_received :mutable_released
+
+    # The single configured uid must be back in the pool, not leaked.
+    assert {:ok, ^uid} = Users.claim()
+  end
+
+  defp budget_spec do
+    %InstanceSpec{
+      vcpus: 0.25,
+      mem: @vm_mem,
+      disk: Information.mib(1),
+      disk_bw: Bandwidth.mibps(1),
+      net_bw: Bandwidth.mibps(1)
+    }
+  end
+
+  defp budget_config do
+    %Hyper.Cfg.Budget{
+      mem_max: Information.gib(1),
+      disk_max: Information.tib(1),
+      cpu_max_load: 1000.0,
+      cpu_max_cap: nil,
+      disk_bw_cap: Bandwidth.gibps(1000),
+      disk_bw_max_load: 1.0,
+      net_bw_cap: Bandwidth.gibps(1000),
+      net_bw_max_load: 1.0,
+      boot_lease_ttl: Unit.Time.s(300),
+      restart_grace: Unit.Time.s(5)
+    }
+  end
+end
diff --git a/test/hyper/node/try_run_admission_test.exs b/test/hyper/node/try_run_admission_test.exs
new file mode 100644
index 00000000..ff6ca456
--- /dev/null
+++ b/test/hyper/node/try_run_admission_test.exs
@@ -0,0 +1,270 @@
+defmodule Hyper.Node.TryRunAdmissionTest do
+  @moduledoc """
+  `Hyper.Node.try_run/3` is the node's authoritative admission gate: the
+  scheduler picks a candidate from a stale gossip snapshot, and the target node
+  confirms. Two different contracts hang off it, and only one is about the
+  ledger.
+
+    * **Ledger consistency** — admitted reservations never exceed `mem_max` /
+      `disk_max`. `Hyper.Node.Budget.Hard` is a single GenServer, so this holds
+      by serialization, and held even before the fix.
+
+    * **Physical containment** — at no *instant* do more VMs exist on this
+      machine than the budget admits, and a refused placement consumes no
+      physical resources at all. The ledger is a proxy for real RAM; a VM that
+      is running consumes it whether or not anything has reserved it.
+
+  The second is what keeps the host off the OOM killer, and it is what these
+  tests pin. `try_run/3` takes `start_fun` as an argument and the fake boot
+  parks until the test releases it, so the boot window is observable without
+  KVM — which is what a real firecracker boot (uid claim, dm-thin snapshot,
+  jailer exec, guest init) occupies for hundreds of milliseconds.
+
+  On the handoff tests: the fake `start_fun` calls `Hard.claim/2` because that is
+  what `Hyper.Node.FireVMM.init/1` does for a real VM, and
+  `DynamicSupervisor.start_child` returns only after a child's `init/1` has — so
+  `{:ok, pid}` coming back from a real `start_fun` already implies the claim
+  happened. The assertions are not on the fake: they are on whether `try_run`
+  holds the reservation itself (it must not) and whether the ledger follows
+  claims rather than boots (it must).
+  """
+
+  use ExUnit.Case, async: false
+  use Unit.Operators
+
+  alias Hyper.Node.Budget.Hard
+  alias Hyper.Vm.Instance.Spec
+  alias Unit.Bandwidth
+  alias Unit.Information
+
+  @vm_mem Information.mib(128)
+  @capacity 4
+  @herd 12
+  @boot_window_ms 400
+
+  setup do
+    saved = :persistent_term.get(Hyper.Cfg.Budget, :unset)
+    :persistent_term.put(Hyper.Cfg.Budget, budget_config())
+
+    on_exit(fn ->
+      case saved do
+        :unset -> :persistent_term.erase(Hyper.Cfg.Budget)
+        config -> :persistent_term.put(Hyper.Cfg.Budget, config)
+      end
+    end)
+
+    start_supervised!(Sys.Mon)
+    start_supervised!(Hard)
+    :ok
+  end
+
+  describe "physical containment under a herd" do
+    test "never puts more VMs on the machine at once than the budget admits" do
+      %{boots: boots} = run_herd()
+
+      assert boots <= @capacity,
+             """
+             #{boots} VMs were simultaneously running on a node budgeted for #{@capacity}.
+             #{boots} x #{Information.as_mib(@vm_mem)} MiB = \
+             #{boots * Information.as_mib(@vm_mem)} MiB physically resident against a \
+             #{Information.as_mib(budget_config().mem_max)} MiB cap.
+             """
+    end
+
+    test "a refused placement never invokes start_fun" do
+      %{boots: boots, results: results} = run_herd()
+
+      assert Enum.count(results, &match?({:ok, _}, &1)) == @capacity
+
+      assert boots == @capacity,
+             "start_fun ran #{boots} times for #{@capacity} admissions: " <>
+               "#{boots - @capacity} VMs were built only to be thrown away"
+    end
+
+    test "the ledger admits exactly the node's capacity" do
+      %{results: results} = run_herd()
+
+      assert Enum.count(results, &match?({:ok, _}, &1)) == @capacity
+      assert Enum.count(results, &(&1 == {:error, :mem_exhausted})) == @herd - @capacity
+    end
+  end
+
+  describe "lease lifetime around the boot" do
+    test "a boot that fails releases the capacity it was granted" do
+      before = Hard.headroom().mem
+
+      assert {:error, :boom} =
+               Hyper.Node.try_run("vm-fails", spec(), fn -> {:error, :boom} end)
+
+      assert Hard.headroom().mem == before
+    end
+
+    test "a boot that raises releases the capacity it was granted" do
+      before = Hard.headroom().mem
+
+      assert_raise RuntimeError, fn ->
+        Hyper.Node.try_run("vm-raises", spec(), fn -> raise "kernel missing" end)
+      end
+
+      assert Hard.headroom().mem == before
+    end
+
+    test "a caller that dies mid-boot releases the capacity it was granted" do
+      before = Hard.headroom().mem
+      parent = self()
+
+      caller =
+        spawn(fn ->
+          Hyper.Node.try_run("vm-abandoned", spec(), fn ->
+            send(parent, :booting)
+            Process.sleep(:infinity)
+          end)
+        end)
+
+      assert_receive :booting
+      assert Hard.headroom().mem == before - @vm_mem
+
+      Process.exit(caller, :kill)
+
+      assert eventually(fn -> Hard.headroom().mem == before end)
+    end
+
+    test "the reservation outlives the placing caller" do
+      before = Hard.headroom().mem
+      vm = spawn_idle()
+
+      task =
+        Task.async(fn ->
+          Hyper.Node.try_run("vm-claims", spec(), fn ->
+            :ok = Hard.claim("vm-claims", vm)
+            {:ok, vm}
+          end)
+        end)
+
+      assert {:ok, ^vm} = Task.await(task)
+
+      # `try_run` has returned and its caller is gone. Had try_run held the
+      # reservation against itself rather than dropping a lease, this releases.
+      assert steadily(fn -> Hard.headroom().mem == before - @vm_mem end)
+    end
+
+    test "a boot that never claims leaves no reservation behind" do
+      before = Hard.headroom().mem
+      vm = spawn_idle()
+
+      task =
+        Task.async(fn ->
+          Hyper.Node.try_run("vm-silent", spec(), fn -> {:ok, vm} end)
+        end)
+
+      assert {:ok, ^vm} = Task.await(task)
+
+      # The ledger follows claims, not boots: nothing claimed, so once the
+      # placing caller's lease is gone the capacity comes back rather than
+      # lingering as an orphan entry.
+      assert eventually(fn -> Hard.headroom().mem == before end)
+    end
+  end
+
+  # Fire `@herd` concurrent placements at the node, holding every fake boot open
+  # until the whole herd has arrived (or `@boot_window_ms` passes, so an
+  # implementation that admits fewer than `@herd` does not hang).
+  #
+  # `boots` is both the number of `start_fun` invocations and the peak number of
+  # concurrently-live VM processes, because every fake boot parks until released.
+  defp run_herd do
+    parent = self()
+
+    callers =
+      for i <- 1..@herd do
+        Task.async(fn ->
+          Hyper.Node.try_run("vm-herd-#{i}", spec(), boot(parent))
+        end)
+      end
+
+    booted = await_boots(@herd, @boot_window_ms)
+    for {caller, _vm} <- booted, do: send(caller, :release)
+
+    results = Task.await_many(callers, 5_000)
+    for {_caller, vm} <- booted, do: Process.exit(vm, :kill)
+
+    %{boots: length(booted), results: results}
+  end
+
+  # Stands in for a real boot: a live process representing the VM's physical
+  # footprint, announced to the test and held open for the duration of the boot.
+  defp boot(parent) do
+    fn ->
+      vm = spawn_idle()
+      send(parent, {:booted, self(), vm})
+
+      receive do
+        :release -> {:ok, vm}
+      after
+        @boot_window_ms -> {:ok, vm}
+      end
+    end
+  end
+
+  defp spawn_idle, do: spawn(fn -> Process.sleep(:infinity) end)
+
+  defp await_boots(max, window_ms) do
+    deadline = System.monotonic_time(:millisecond) + window_ms
+    await_boots(max, deadline, [])
+  end
+
+  defp await_boots(0, _deadline, acc), do: acc
+
+  defp await_boots(remaining, deadline, acc) do
+    receive do
+      {:booted, caller, vm} -> await_boots(remaining - 1, deadline, [{caller, vm} | acc])
+    after
+      max(deadline - System.monotonic_time(:millisecond), 0) -> acc
+    end
+  end
+
+  defp eventually(fun, attempts \\ 50) do
+    cond do
+      fun.() -> true
+      attempts == 0 -> false
+      true -> Process.sleep(2) && eventually(fun, attempts - 1)
+    end
+  end
+
+  # The dual of `eventually`: the condition must hold for the whole window, so a
+  # release that is merely late still fails the test.
+  defp steadily(fun, attempts \\ 25) do
+    cond do
+      not fun.() -> false
+      attempts == 0 -> true
+      true -> Process.sleep(2) && steadily(fun, attempts - 1)
+    end
+  end
+
+  defp spec do
+    %Spec{
+      vcpus: 0.25,
+      mem: @vm_mem,
+      disk: Information.gib(1),
+      disk_bw: Bandwidth.mibps(1),
+      net_bw: Bandwidth.mibps(1)
+    }
+  end
+
+  # Memory is the only binding constraint: every other cap is set far out of
+  # reach so the soft monitors' live readings cannot decide the outcome.
+  defp budget_config do
+    %Hyper.Cfg.Budget{
+      mem_max: Information.mib(@capacity * Information.as_mib(@vm_mem)),
+      disk_max: Information.tib(1),
+      cpu_max_load: 1000.0,
+      cpu_max_cap: nil,
+      disk_bw_cap: Bandwidth.gibps(1000),
+      disk_bw_max_load: 1.0,
+      net_bw_cap: Bandwidth.gibps(1000),
+      net_bw_max_load: 1.0,
+      boot_lease_ttl: Unit.Time.s(300),
+      restart_grace: Unit.Time.s(5)
+    }
+  end
+end