runpod: a warm floor is not satisfied by a pod that no longer exists (v0.3.44)

`keep_warm` stopped re-renting whenever a pod went away behind the pool's
back — terminated from the console, reclaimed as spot, lost with its host —
and the scaler said nothing about it. Both halves had one cause: nothing in
a pool ever re-checks a pod's health. A handle's `healthy` flag goes false
only when something tried to USE the pod and failed, or when the pool itself
retired it, and a warm-floor pod is precisely the pod with no traffic to
fail. So `maintain()` counted the handle, found `healthy_n >= keep_min`, and
neither topped up nor had anything to log. The floor stayed "satisfied" by a
machine that had not existed for hours, and the first real request paid the
full ~12-minute cold boot that keep_warm is bought to avoid.

(The class docstring claimed the scaler "re-checks health" for months. It
never has. Corrected rather than implemented: a timer-probe would have to
decide what an unanswered probe means, and that decision is what killed
seven A40s in v0.3.41.)

`reconcile_pools` instead reads the account listing the stale-pod reaper
already fetches every ~30s — authoritative, free, and unable to mistake a
busy pod for a dead one. Only positive evidence acts:

  * a dead status (TERMINATED/EXITED/DEAD/FAILED) drops the pod at once;
  * a mere absence drops it only after two listings in a row, because the
    API is eventually consistent and one page omitting a live pod used to be
    enough to unregister it — a reappearance resets the count;
  * a pod that served a request in the last 60s is kept whatever the listing
    says, which cannot mask this bug since a warm-floor pod has no traffic;
  * a provisioning pod is shielded, and a FAILED listing never gets here at
    all, so an API outage cannot empty every pool and re-rent the fleet.

`PodPool.forget()` is `discard()` minus the kill: nothing is left to
terminate, but the bill is closed in the ledger exactly as a real
termination closes it, since the pod was billed up to its death. A borrowed
`(shared)` pod is dropped without a ledger entry — its owner holds the bill.
The registry is pruned through the same two-sweep rule it used to skip.

The top-up now announces itself before spending, because a warm floor
silently failing is indistinguishable from a warm floor that is fine.

21 new tests in tests/test_runpod_warm_floor_rerent.py pin the caution, not
just the fix: the failure mode in the other direction is forgetting a pod
that is alive and billing. test_runpod_client_scale's source assertion moved
to the new seam.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PxYHWbCAjjjD7A1rhFKtaW

codai/__init__.py

@@ -16,7 +16,7 @@
 
 # Canonical product version for CoderAI — single source of truth. Both the API
 # metadata and the admin web UI read from here.
-__version__ = "0.3.43"
+__version__ = "0.3.44"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/api/runpod_worker.py

@@ -2176,8 +2176,18 @@ class RunpodPodPool:
       ``max_pods``, subject to the cost caps) and bumps its in-flight count.
     * ``release()`` drops the in-flight count and stamps last-used.
     * the shared scaler thread (``_scaler_loop``) tears down idle pods to
-      ``min_pods`` after ``idle_timeout_s``, re-checks health, reaps dead pods,
-      and keeps ``min_pods`` warm.
+      ``min_pods`` after ``idle_timeout_s``, reconciles the pool against what
+      RunPod actually has, and keeps ``min_pods`` warm.
+
+    Note what the pool does NOT do: probe its own pods on a timer. Nothing here
+    re-checks health (this docstring claimed it did for months, and it never
+    has). A handle's ``healthy`` flag only goes false when something tried to
+    USE the pod and failed, or when the pool itself retired it. So a pod that
+    dies with no traffic on it — which is precisely a warm-floor pod — stays
+    ``healthy=True`` forever and the floor is satisfied by a machine that no
+    longer exists. ``reconcile_pools`` closes that, from the account listing the
+    reaper already fetches: authoritative, free, and incapable of mistaking a
+    busy pod for a dead one.
     """
 
     def __init__(self, model_key, account_cfg, mcfg: "RunpodModelConfig", served_name,
@@ -3004,6 +3014,46 @@ class RunpodPodPool:
         print(f"[runpod] pod {pod.pod_id} dropped from the pool: {reason}", flush=True)
         threading.Thread(target=self._terminate, args=(pod, reason), daemon=True).start()
 
+    def forget(self, pod: PodHandle, reason: str = "gone") -> bool:
+        """Drop a pod that no longer exists. Like ``discard``, minus the kill.
+
+        ``discard`` terminates, because it is called when OUR use of a pod
+        failed and the pod may well still be running and billing. This one is
+        for a pod RunPod has positively told us is gone — terminated from the
+        console, reclaimed as spot, lost with its host — where there is nothing
+        left to terminate and an API call would only error. The bill still has
+        to be closed, so the ledger entry is written here exactly as
+        ``_terminate`` writes it.
+
+        Returns True if this call is what removed it.
+        """
+        with self._cv:
+            if pod not in self.pods:
+                return False
+            pod.healthy = False
+            self.pods.remove(pod)
+        if getattr(pod, "gpu", "") == "(shared)":
+            # Another process owns the bill; it writes the ledger entry.
+            print(f"[runpod] borrowed pod {pod.pod_id} is gone ({reason}) — "
+                  f"dropped from {self.model_key!r}", flush=True)
+            return True
+        from codai.api import runpod_ledger
+        # To NOW, which over-counts by however long it was dead before we
+        # noticed. The same approximation _terminate makes, and the reconcile
+        # runs every ~30s, so the error is bounded by that.
+        cost = (time.time() - pod.started_at) / 3600.0 * pod.hourly_usd
+        unregister_pod(pod.pod_id)
+        try:
+            runpod_ledger.record_spend(
+                self.model_key, cost,
+                {"pod_id": pod.pod_id, "gpu": pod.gpu, "is_spot": pod.is_spot})
+        except Exception as exc:
+            print(f"[runpod] ledger for gone pod {pod.pod_id}: {exc}", flush=True)
+        print(f"[runpod] pod {pod.pod_id} for {self.model_key!r} is gone from "
+              f"RunPod ({reason}) — dropped from the pool; billed ~${cost:.3f}. "
+              f"The warm floor is re-checked on the next scaler pass.", flush=True)
+        return True
+
     def _terminate(self, pod: PodHandle, reason: str):
         if getattr(pod, "gpu", "") == "(shared)":
             # Borrowed from another engine: it owns the lifecycle and the bill.
@@ -3305,6 +3355,12 @@ class RunpodPodPool:
                 if self._provisioning:
                     return
                 self._provisioning = True
+            # Say so before spending. A warm floor silently failing to be
+            # re-established is indistinguishable, from outside, from a warm
+            # floor that is fine — which is how a pod terminated out of band
+            # went unnoticed until the next request paid a cold boot.
+            print(f"[runpod] {self.model_key!r}: warm floor {keep_min}, "
+                  f"{healthy_n} pod(s) healthy — renting one", flush=True)
             try:
                 if not self._adopt_shared_pod():
                     self._provision_one()
@@ -3965,6 +4021,90 @@ def registered_pod_ids() -> set:
     return set(data.keys()) if isinstance(data, dict) else set()
 
 
+#: Pod ids a successful listing did not mention, from the PREVIOUS sweep. A pod
+#: has to be missing from two listings in a row before the pool gives up on it:
+#: the RunPod API is eventually consistent, a partial page or a brief backend
+#: hiccup can omit a pod that is running perfectly well, and the cost of
+#: believing one such listing is a second rented GPU. Guarded by _pools_lock.
+_absent_before: set = set()
+
+#: Statuses that are RunPod telling us, in so many words, that the machine is
+#: finished. No second opinion needed for these.
+_DEAD_STATUSES = ("TERMINATED", "EXITED", "DEAD", "FAILED")
+
+
+def reconcile_pools(live_status: dict, candidates=None) -> set:
+    """Reconcile pools (and the registry) against what RunPod actually has.
+
+    Returns the set of pod ids now CONFIRMED gone — the caller unregisters
+    those; any that were in a pool have already been forgotten here.
+
+    The bug this exists for: ``keep_warm`` would not re-rent after a pod was
+    terminated out of band, and nothing was logged. Both halves had the same
+    cause — a pool only learns a pod is bad by FAILING to use it, and a
+    warm-floor pod by definition has no traffic to fail. ``maintain()`` counted
+    it as healthy, so ``healthy_n < keep_min`` was never true, so it never
+    topped up and never had anything to say. The floor stayed "satisfied" by a
+    machine that had not existed for hours, and the first real request paid a
+    full cold boot — the exact cost keep_warm is there to avoid.
+
+    ``live_status`` maps pod id -> status string, from one successful account
+    listing. Only positive evidence acts: a dead status immediately, an absence
+    only when the previous sweep saw it absent too. An empty or failed listing
+    must never reach this function — see the caller.
+    """
+    global _absent_before
+    with _pools_lock:
+        pools = list(_pools.values())
+        previously_absent = set(_absent_before)
+        shielded = set(_provisioning_ids)
+    #: id -> the pool handle holding it, for the ids we are judging.
+    owned = {}
+    for pool in pools:
+        with pool._cv:
+            for h in pool.pods:
+                owned[str(h.pod_id)] = (pool, h)
+    ids = {str(c) for c in (candidates or ())} | set(owned)
+    absent_now, confirmed = set(), set()
+    for pid in sorted(ids):
+        if pid in shielded:
+            continue
+        ph = owned[pid][1] if pid in owned else None
+        status = live_status.get(pid)
+        if status is None:
+            absent_now.add(pid)
+            if pid not in previously_absent:
+                # First listing without it. Say so when it is a pod we are
+                # actively holding — one that flickers in and out of the
+                # account listing is itself worth knowing about — but never act
+                # on one data point.
+                if pid in owned:
+                    print(f"[runpod] pod {pid} ({owned[pid][0].model_key!r}) is not "
+                          f"in the RunPod account listing; confirming on the next "
+                          f"sweep before dropping it", flush=True)
+                continue
+            # A pod that served a request seconds ago exists, whatever the
+            # listing says — traffic is better evidence than an API page. Only
+            # the absence branch gets this: an explicit dead status is RunPod
+            # telling us directly, and is trusted even mid-request. A warm-floor
+            # pod, the case this whole function is for, has no traffic to speak
+            # for it, so the guard never masks it.
+            if (time.time() - float(getattr(ph, "last_used", 0) or 0)) < 60.0:
+                continue
+            reason = "not in the account listing twice running"
+        elif str(status).upper() in _DEAD_STATUSES:
+            reason = f"status {str(status).upper()}"
+        else:
+            continue
+        confirmed.add(pid)
+        if pid in owned:
+            pool, h = owned[pid]
+            pool.forget(h, reason)
+    with _pools_lock:
+        _absent_before = absent_now
+    return confirmed
+
+
 def _known_pod_ids() -> set:
     """Every pod id this DEPLOYMENT owns — not just this process's."""
     ids = set()
@@ -4035,12 +4175,22 @@ def reap_orphans() -> int:
         return 0
     known = _known_pod_ids()
     reaped = 0
-    # Drop registry entries for pods RunPod no longer has. They are shielded
-    # from reaping by being "known", and they accumulate for the life of the
-    # install — including pods terminated out of band.
-    live = {str(p.get("id")) for p in pods if p.get("id")}
-    for gone in (known - live):
-        unregister_pod(gone)
+    # Reconcile what we BELIEVE we have against what the account actually has,
+    # before reaping anything. This listing is the only place a pod that died
+    # out of band becomes visible: a pool never probes its own pods, so a
+    # warm-floor pod with no traffic on it is never found dead by use. Pools
+    # drop such a handle (and close its bill) here; registry entries for pods
+    # RunPod no longer has go the same way — they are shielded from reaping by
+    # being "known", so they would otherwise accumulate for the life of the
+    # install. An empty listing is not evidence of an empty account, so a
+    # successful-but-empty response is left to the two-sweep rule inside.
+    live_status = {str(p.get("id")): str(p.get("status") or "")
+                   for p in pods if p.get("id")}
+    try:
+        for gone in reconcile_pools(live_status, known):
+            unregister_pod(gone)
+    except Exception as exc:
+        print(f"[runpod-reaper] reconcile: {exc}", flush=True)
     for p in pods:
         pid, name = p.get("id"), (p.get("name") or "")
         status = str(p.get("status") or "").upper()

docs/runpod.md

@@ -334,6 +334,56 @@ Consequences worth knowing:
 
 `atexit` also tears down every pool on a clean shutdown.
 
+### The same sweep reconciles the pools (a pod that dies behind our back)
+
+The listing is not only used to find pods that should not exist. It is also the
+only place a pod that **should** exist and no longer does becomes visible.
+
+Nothing in a pool probes its own pods on a timer. A `PodHandle`'s `healthy` flag
+goes false when something tried to **use** the pod and failed, or when the pool
+itself retired it — which means a pod with no traffic on it is never discovered
+dead. That is exactly a warm-floor pod. So a `keep_warm` pod terminated from the
+RunPod console, reclaimed as spot, or lost with its host left the pool believing
+its floor was satisfied, **indefinitely and silently**: `maintain()` counted the
+handle, found `healthy >= min_pods`, and neither topped up nor logged. The first
+real request then paid a full cold boot — around 12 minutes on the Digesta
+orchestrator — which is the precise cost `keep_warm` is bought to avoid.
+
+Each successful listing is now reconciled against what the pools believe:
+
+- A pod RunPod reports as `TERMINATED`/`EXITED`/`DEAD`/`FAILED` is dropped
+  immediately. That is RunPod telling us directly; a second opinion adds nothing.
+- A pod simply **missing** from the listing is dropped only after it is missing
+  from **two listings in a row** (~30s apart). The API is eventually consistent
+  and pages; one response that omits a running pod used to be enough to
+  unregister it, and believing it costs a second rented GPU for work already
+  being done. A pod that reappears resets the count.
+- A pod that **served a request in the last 60s** is kept whatever the listing
+  says — traffic is better evidence than an API page. This cannot mask the bug
+  above, because a warm-floor pod has no traffic to speak for it.
+- A pod still being **provisioned** is shielded, and a **failed** listing never
+  reaches the reconcile at all: `reap_orphans` returns early on a `RunpodError`,
+  so an API outage cannot empty every pool and re-rent the fleet.
+
+Dropping such a handle does **not** call the API — there is nothing left to
+terminate — but it does close the bill: the ledger entry is written exactly as a
+real termination writes it, estimated up to the moment we noticed, because a
+pod that died out of band was billed right up to its death. A borrowed
+(`(shared)`) pod is dropped without a ledger entry; its owner holds the bill.
+
+Once the handle is gone, the next scaler pass sees the pool below its floor and
+says so before it spends:
+
+```
+[runpod] pod qb9ap… for 'pool:digesta-qwen38' is gone from RunPod (status TERMINATED)
+         — dropped from the pool; billed ~$0.412
+[runpod] 'pool:digesta-qwen38': warm floor 1, 0 pod(s) healthy — renting one
+```
+
+A pool with `min_pods: 0` re-rents nothing: a pod dying out of band there is
+simply gone, and renting a replacement nobody asked for is how an idle install
+bills overnight.
+
 ---
 
 ## 5. Budgets and the ledger

tests/test_runpod_client_scale.py

@@ -467,10 +467,18 @@ def test_a_booting_pod_from_a_previous_process_counts():
 
 def test_the_reaper_prunes_entries_for_pods_that_are_gone():
     """A terminated-but-registered id stays 'known', so it is shielded from
-    reaping and accumulates for the life of the install."""
+    reaping and accumulates for the life of the install.
+
+    The pruning moved into `reconcile_pools`, which judges the pools' own pods
+    and the registry from the same listing — and gates an ABSENCE on two
+    listings in a row, because the API is eventually consistent and one page
+    that omits a live pod used to be enough to unregister it. Behaviour pinned
+    in tests/test_runpod_warm_floor_rerent.py; what matters here is that
+    reap_orphans still feeds it the listing and acts on what comes back."""
     src = __import__("pathlib").Path(rw.__file__).read_text()
     blk = src.split("def reap_orphans")[1].split("def _scaler_loop")[0]
-    assert "known - live" in blk and "unregister_pod(gone)" in blk
+    assert "reconcile_pools(live_status, known)" in blk
+    assert "unregister_pod(gone)" in blk
 
 
 # ------------------------------- the pid in a container is not an identity

tests/test_runpod_warm_floor_rerent.py

+"""A warm floor satisfied by a machine that no longer exists.
+
+`keep_warm` exists to pay for an idle GPU so that the first real request does
+not wait out a cold boot — on the Digesta orchestrator, ~12 minutes of image
+pull and warm-up. It stopped doing that whenever a pod went away behind the
+pool's back: terminated from the RunPod console, reclaimed as spot, lost with
+its host. The pool kept the floor "satisfied" indefinitely and said nothing.
+
+The cause is one line that was never written: nothing re-checks a pod's health.
+A `PodHandle.healthy` flag goes false only when something tried to USE the pod
+and failed (`discard`), or when the pool itself retired it. A warm-floor pod is
+precisely the pod with no traffic on it, so there is no use to fail — and
+`maintain()` counted it, found `healthy_n >= keep_min`, and never topped up and
+never had anything to log. (`RunpodPodPool`'s own docstring claimed the scaler
+"re-checks health" for months. It never has.)
+
+The fix reconciles pools against the account listing the stale-pod reaper
+already fetches every ~30 seconds: authoritative, free, and incapable of
+mistaking a busy pod for a dead one. What these tests pin is the caution around
+it, because the failure mode in the other direction is forgetting a pod that is
+alive and billing: only positive evidence acts, an absence needs two listings in
+a row, a failed listing is not an empty one, and recent traffic outvotes the
+API.
+"""
+
+import pathlib
+import sys
+import threading
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from codai.api import runpod_worker as rw                          # noqa: E402
+
+
+def _pool(key="pool:digesta-qwen38", min_pods=1, pods=()):
+    pool = rw.RunpodPodPool.__new__(rw.RunpodPodPool)
+    pool._cv = threading.Condition()
+    pool.model_key = key
+    pool._closed = False
+    pool._provisioning = False
+    pool._surplus_since = 0.0
+    pool._release_pending = False
+    pool._sched_closed_logged = False
+    pool.pods = list(pods)
+    cfg = rw.RunpodModelConfig()
+    cfg.min_pods = min_pods
+    cfg.keep_warm = min_pods >= 1
+    cfg.idle_timeout_s = 300
+    cfg.schedule_enabled = False
+    pool.mcfg = cfg
+    pool.effective_min_pods = lambda: min_pods
+    pool.account = None
+    return pool
+
+
+def _pod(pid, last_used=0.0, inflight=0, hourly_usd=0.59, gpu="A40"):
+    return rw.PodHandle(pod_id=pid, url=f"https://{pid}:22000", hourly_usd=hourly_usd,
+                        started_at=0.0, gpu=gpu, inflight=inflight,
+                        last_used=last_used)
+
+
+def _install(monkeypatch, pool, ledger=None, unregistered=None):
+    """Put `pool` in the module's pool table and stub the two side effects
+    `forget` has: the spend ledger and the shared registry."""
+    monkeypatch.setattr(rw, "_pools", {pool.model_key: pool}, raising=False)
+    monkeypatch.setattr(rw, "_provisioning_ids", set(), raising=False)
+    monkeypatch.setattr(rw, "_absent_before", set(), raising=False)
+    monkeypatch.setattr(rw, "unregister_pod",
+                        lambda pid: (unregistered if unregistered is not None
+                                     else []).append(pid))
+
+    class _Ledger:
+        @staticmethod
+        def record_spend(key, cost, meta):
+            if ledger is not None:
+                ledger.append((key, round(cost, 4), meta.get("pod_id")))
+
+    monkeypatch.setitem(sys.modules, "codai.api.runpod_ledger", _Ledger)
+
+
+# -------------------------------------------- what the pool knows on its own #
+def test_nothing_in_the_pool_ever_rechecks_a_pods_health():
+    """The whole bug in one assertion. If a health probe is ever added, this
+    test should fail and be replaced — but it must not be *assumed*."""
+    src = (ROOT / "codai/api/runpod_worker.py").read_text()
+    body = src.split("def maintain(self)")[1].split("\n    def ")[0]
+    assert "healthy = True" not in body          # it never revives one either
+    # The only ways a handle leaves a pool: use failed, retirement, reconcile.
+    assert "reconcile" in src
+
+
+def test_a_dead_warm_pod_keeps_the_floor_satisfied_without_reconcile():
+    """Before the fix, this was the end of the story: one healthy-looking
+    handle, floor of 1, nothing to do, nothing to say."""
+    pool = _pool(min_pods=1, pods=[_pod("ghost")])
+    healthy = len([p for p in pool.pods if p.healthy])
+    assert healthy == 1 and healthy >= pool.effective_min_pods()
+
+
+# ------------------------------------------------- only positive evidence acts #
+def test_an_explicitly_terminated_pod_is_dropped_at_once(monkeypatch):
+    """RunPod saying TERMINATED is RunPod telling us directly. No confirmation
+    round, because there is nothing a second listing could add."""
+    pool = _pool(pods=[_pod("ghost")])
+    _install(monkeypatch, pool)
+    gone = rw.reconcile_pools({"ghost": "TERMINATED"})
+    assert gone == {"ghost"}
+    assert pool.pods == []
+
+
+def test_every_finished_status_counts_as_finished(monkeypatch):
+    for status in ("TERMINATED", "EXITED", "DEAD", "FAILED", "terminated"):
+        pool = _pool(pods=[_pod("ghost")])
+        _install(monkeypatch, pool)
+        assert rw.reconcile_pools({"ghost": status}) == {"ghost"}, status
+
+
+def test_a_running_pod_is_left_alone(monkeypatch):
+    pool = _pool(pods=[_pod("good")])
+    _install(monkeypatch, pool)
+    assert rw.reconcile_pools({"good": "RUNNING"}) == set()
+    assert [p.pod_id for p in pool.pods] == ["good"]
+
+
+# ------------------------------------------------- absence needs two listings #
+def test_one_listing_without_the_pod_is_not_enough(monkeypatch):
+    """The RunPod API is eventually consistent and pages; a pod can be missing
+    from one response and running perfectly well. Believing it costs a second
+    rented GPU for work already being done."""
+    pool = _pool(pods=[_pod("maybe")])
+    _install(monkeypatch, pool)
+    assert rw.reconcile_pools({"other": "RUNNING"}) == set()
+    assert [p.pod_id for p in pool.pods] == ["maybe"]
+
+
+def test_two_listings_without_it_settle_it(monkeypatch):
+    pool = _pool(pods=[_pod("maybe")])
+    _install(monkeypatch, pool)
+    rw.reconcile_pools({"other": "RUNNING"})
+    assert rw.reconcile_pools({"other": "RUNNING"}) == {"maybe"}
+    assert pool.pods == []
+
+
+def test_a_pod_that_reappears_starts_the_count_again(monkeypatch):
+    """A flicker must not accumulate into a death over hours."""
+    pool = _pool(pods=[_pod("flicker")])
+    _install(monkeypatch, pool)
+    rw.reconcile_pools({})                       # absent once
+    rw.reconcile_pools({"flicker": "RUNNING"})   # back
+    assert rw.reconcile_pools({}) == set()       # absent once again, not twice
+    assert [p.pod_id for p in pool.pods] == ["flicker"]
+
+
+def test_recent_traffic_outvotes_the_listing(monkeypatch):
+    """A pod that answered a request seconds ago exists. This guard cannot mask
+    the bug being fixed: a warm-floor pod has no traffic to speak for it."""
+    import time
+    pool = _pool(pods=[_pod("busy", last_used=time.time())])
+    _install(monkeypatch, pool)
+    rw.reconcile_pools({})
+    assert rw.reconcile_pools({}) == set()
+    assert [p.pod_id for p in pool.pods] == ["busy"]
+
+
+def test_traffic_does_not_save_a_pod_runpod_calls_terminated(monkeypatch):
+    import time
+    pool = _pool(pods=[_pod("busy", last_used=time.time(), inflight=3)])
+    _install(monkeypatch, pool)
+    assert rw.reconcile_pools({"busy": "TERMINATED"}) == {"busy"}
+
+
+def test_a_pod_still_provisioning_is_shielded(monkeypatch):
+    """It may not be in the listing yet; the pool is mid-create."""
+    pool = _pool(pods=[_pod("booting")])
+    _install(monkeypatch, pool)
+    monkeypatch.setattr(rw, "_provisioning_ids", {"booting"}, raising=False)
+    rw.reconcile_pools({})
+    assert rw.reconcile_pools({}) == set()
+    assert [p.pod_id for p in pool.pods] == ["booting"]
+
+
+# --------------------------------------------------- a failed listing is not data #
+def test_a_listing_that_failed_never_reaches_the_reconcile(monkeypatch):
+    """`reap_orphans` returns on RunpodError before anything is judged.
+    Otherwise an API outage would empty every pool and re-rent the fleet."""
+    from codai.api import runpod_client
+
+    class _Acct:
+        enabled, api_key = True, "k"
+
+    monkeypatch.setattr("codai.models.manager.get_active_runpod_config",
+                        lambda: _Acct, raising=False)
+
+    def _boom(self):
+        raise runpod_client.RunpodError("api down")
+
+    monkeypatch.setattr(runpod_client.RunpodClient, "list_pods", _boom, raising=False)
+    monkeypatch.setattr(runpod_client.RunpodClient, "__init__",
+                        lambda self, acct: None, raising=False)
+    called = []
+    monkeypatch.setattr(rw, "reconcile_pools",
+                        lambda *a, **k: called.append(a) or set())
+    assert rw.reap_orphans() == 0
+    assert called == []
+
+
+# ------------------------------------------------------------- closing the bill #
+def test_forgetting_a_pod_does_not_call_the_api(monkeypatch):
+    """There is nothing left to terminate, and the call would only error."""
+    pool = _pool(pods=[_pod("ghost")])
+    _install(monkeypatch, pool)
+
+    class _Client:
+        def __init__(self, acct):
+            raise AssertionError("forget must not talk to RunPod")
+
+    monkeypatch.setattr("codai.api.runpod_client.RunpodClient", _Client)
+    rw.reconcile_pools({"ghost": "TERMINATED"})
+    assert pool.pods == []
+
+
+def test_the_bill_is_still_recorded_and_the_registry_cleaned(monkeypatch):
+    """A pod that died out of band was billed right up to its death. Dropping
+    the handle without a ledger entry would make the GPU look free."""
+    ledger, unregistered = [], []
+    pod = _pod("ghost", hourly_usd=0.59)
+    pod.started_at = 0.0
+    pool = _pool(pods=[pod])
+    _install(monkeypatch, pool, ledger=ledger, unregistered=unregistered)
+    rw.reconcile_pools({"ghost": "TERMINATED"})
+    assert unregistered == ["ghost"]
+    assert len(ledger) == 1
+    key, cost, pid = ledger[0]
+    assert (key, pid) == (pool.model_key, "ghost") and cost > 0
+
+
+def test_a_borrowed_pod_is_dropped_but_not_billed_here(monkeypatch):
+    """Its owner holds the lifecycle and the bill; double-counting a shared pod
+    would charge one GPU to two pools."""
+    ledger = []
+    pool = _pool(pods=[_pod("shared", gpu="(shared)")])
+    _install(monkeypatch, pool, ledger=ledger)
+    rw.reconcile_pools({"shared": "TERMINATED"})
+    assert pool.pods == [] and ledger == []
+
+
+def test_forget_is_idempotent(monkeypatch):
+    """Two sweeps racing on the same corpse must bill it once."""
+    ledger = []
+    pod = _pod("ghost")
+    pool = _pool(pods=[pod])
+    _install(monkeypatch, pool, ledger=ledger)
+    assert pool.forget(pod, "status TERMINATED") is True
+    assert pool.forget(pod, "status TERMINATED") is False
+    assert len(ledger) == 1
+
+
+# ----------------------------------------------------- and then it re-rents #
+def test_the_floor_is_re_established_once_the_ghost_is_gone(monkeypatch):
+    """The point of the whole exercise: maintain() can now see the pool is
+    below its floor, and says so before it spends."""
+    pool = _pool(min_pods=1, pods=[_pod("ghost")])
+    _install(monkeypatch, pool)
+    rw.reconcile_pools({"ghost": "TERMINATED"})
+
+    rented = []
+    pool._adopt_shared_pod = lambda: False
+    pool._provision_one = lambda: rented.append("rented")
+    pool._drain_pod_log = lambda p: None
+    pool._scale_down = lambda now: None
+    monkeypatch.setattr(rw, "sibling_is_provisioning", lambda key: False)
+    out = []
+    monkeypatch.setattr("builtins.print", lambda *a, **k: out.append(" ".join(map(str, a))))
+    pool.maintain()
+    monkeypatch.undo()
+    assert rented == ["rented"]
+    assert any("warm floor 1" in line and "renting one" in line for line in out), out
+
+
+def test_a_sibling_already_renting_is_one_pod_not_none(monkeypatch):
+    """Topping up against a booting pod is how a warm floor of 1 became two."""
+    pool = _pool(min_pods=1, pods=[])
+    _install(monkeypatch, pool)
+    rented = []
+    pool._adopt_shared_pod = lambda: False
+    pool._provision_one = lambda: rented.append("rented")
+    pool._drain_pod_log = lambda p: None
+    pool._scale_down = lambda now: None
+    monkeypatch.setattr(rw, "sibling_is_provisioning", lambda key: True)
+    pool.maintain()
+    assert rented == []
+
+
+def test_an_existing_pod_elsewhere_is_adopted_before_renting(monkeypatch):
+    """Cheaper than a GPU, and avoids paying twice for one job."""
+    pool = _pool(min_pods=1, pods=[])
+    _install(monkeypatch, pool)
+    rented = []
+    pool._adopt_shared_pod = lambda: True
+    pool._provision_one = lambda: rented.append("rented")
+    pool._drain_pod_log = lambda p: None
+    pool._scale_down = lambda now: None
+    monkeypatch.setattr(rw, "sibling_is_provisioning", lambda key: False)
+    pool.maintain()
+    assert rented == []
+
+
+def test_no_floor_means_no_re_rent(monkeypatch):
+    """A pod dying out of band on a min_pods=0 pool is simply gone. Renting a
+    replacement nobody asked for is how an idle install bills overnight."""
+    pool = _pool(min_pods=0, pods=[_pod("ghost")])
+    _install(monkeypatch, pool)
+    rw.reconcile_pools({"ghost": "TERMINATED"})
+    rented = []
+    pool._adopt_shared_pod = lambda: False
+    pool._provision_one = lambda: rented.append("rented")
+    pool._drain_pod_log = lambda p: None
+    pool._scale_down = lambda now: None
+    monkeypatch.setattr(rw, "sibling_is_provisioning", lambda key: False)
+    pool.maintain()
+    assert rented == []
+
+
+# ------------------------------------------------------- the registry as well #
+def test_registry_entries_for_vanished_pods_are_dropped_too(monkeypatch):
+    """They are shielded from reaping by being "known", so without this they
+    accumulate for the life of the install — and a stale one with a URL is
+    handed out by find_shared_pod as if it were a live pod."""
+    pool = _pool(min_pods=0, pods=[])
+    _install(monkeypatch, pool)
+    rw.reconcile_pools({}, candidates={"orphan"})
+    assert rw.reconcile_pools({}, candidates={"orphan"}) == {"orphan"}