runpod: a registry entry must stop vouching for a pod when its owner dies (v0.3.56)

A qwen38 pool pod (A40, $0.59/hr) was found running at 21:30, outside its
08:00-20:00 schedule window, hours after it should have been down. Its config
was correct and the pool said so every pass — "outside its warm-pod window —
warm floor 0" — while the pod billed on.

Two mechanisms each did their job and left a gap between them. The 21:17
unattended upgrade restarted coderai; the pod survived it. With a warm floor of
0 nothing provisioned, and adoption (`find_shared_pod`) only runs when a pool
actually needs a pod — so no pool tracked it, and no pool applied
`idle_timeout_s` to it. Meanwhile the reaper skips everything in
`_known_pod_ids()`, and the dead process's registry entry was still in there,
because `registered_pod_ids()` returned every key in the file regardless of who
wrote it or whether that process still existed. The pod was simultaneously
unmanaged and shielded. Nothing in the registry could tell the two apart: `at`
was written once at registration and never updated.

`at` is now a heartbeat. `touch_registered_pods()` refreshes this process's own
entries from the scaler loop (every fourth 15 s pass), `live_registered_pod_ids()`
counts ours unconditionally and a sibling's only while its heartbeat is fresh,
and `_known_pod_ids()` uses that. An entry whose owner has been silent for
REAP_GRACE_SECONDS stops shielding its pod, and the reaper treats it like any
other leftover. The grace is 1200 s against a 60 s heartbeat, so a sibling
between passes is never at risk.

The trade this accepts: a process still running but whose maintenance loop has
died stops heartbeating, and a sibling will reap its pods after the grace.
Twenty minutes of silence from the loop that is supposed to manage a pod is
itself the failure.

Also: a pod the reaper kills is now recorded in the ledger, computed from a new
`created` field (distinct from the heartbeat, which moves). Those hours were
invisible before — only `_terminate` recorded spend, and that only runs for a
pod a pool was tracking, which is exactly not this case. And the reaper
unregisters what it terminates, so it cannot try twice.

The pod itself was terminated by hand and its stale entry removed; the account
is back to zero pods.

11 tests in tests/test_runpod_registry_heartbeat.py, including the whole bug in
one assertion: an orphan id is in `registered_pod_ids()` and not in
`_known_pod_ids()`.
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.55"
+__version__ = "0.3.56"
 __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

@@ -4390,6 +4390,10 @@ def register_pod(pod_id: str, pool_key: str = "", url: str = "",
     _registry_update(lambda d: d.__setitem__(
         str(pod_id), {"pool": str(pool_key), "pid": os.getpid(),
                       "proc": _PROC_ID, "at": time.time(),
+                      # Distinct from "at", which is now a heartbeat: the hours
+                      # a reaped pod billed can only be computed from when it
+                      # was CREATED, and that number must survive the owner.
+                      "created": time.time(),
                       "url": str(url or ""), "key": str(api_key or ""),
                       # So a pod adopted after a restart is still attributed its
                       # real $/hr instead of looking free.
@@ -4397,6 +4401,36 @@ def register_pod(pod_id: str, pool_key: str = "", url: str = "",
         or True)
 
 
+def touch_registered_pods() -> int:
+    """Refresh this process's registry entries. Returns how many were touched.
+
+    ``at`` is what tells a sibling's reaper that an entry still has an owner.
+    Without a heartbeat it only ever said when the pod was REGISTERED, so an
+    entry outlived its process forever — and because the reaper skips anything
+    "known", the pod it named was shielded from reaping for the life of the
+    install while no pool was managing it either. Exactly what happened to a
+    qwen38 pod across an unattended upgrade: its pool had a warm floor of 0
+    (outside its schedule window), so nothing adopted it, nothing applied
+    ``idle_timeout_s`` to it, and the reaper left it alone because the dead
+    process's entry still vouched for it. It billed on.
+
+    So: a live owner says so every pass, and silence past REAP_GRACE_SECONDS
+    lets the pod be reaped like any other leftover.
+    """
+    touched = [0]
+
+    def _fn(d):
+        now = time.time()
+        for info in d.values():
+            if isinstance(info, dict) and _is_this_process(info):
+                info["at"] = now
+                touched[0] += 1
+        return bool(touched[0])
+
+    _registry_update(_fn)
+    return touched[0]
+
+
 def sibling_is_provisioning(pool_key: str) -> bool:
     """True when another process has a pod for this pool that is still booting.
 
@@ -4468,6 +4502,32 @@ def registered_pod_ids() -> set:
     return set(data.keys()) if isinstance(data, dict) else set()
 
 
+def live_registered_pod_ids() -> set:
+    """Registered pods whose owner is still around to manage them.
+
+    Ours count unconditionally. A sibling's count while its heartbeat is fresh.
+    Everything else is a pod with no manager, which is the reaper's job.
+    """
+    data = _registry_update(lambda d: False)
+    if not isinstance(data, dict):
+        return set()
+    now, out = time.time(), set()
+    for pod_id, info in data.items():
+        if not isinstance(info, dict):
+            continue
+        if _is_this_process(info) or (now - float(info.get("at") or 0)
+                                      < REAP_GRACE_SECONDS):
+            out.add(pod_id)
+    return out
+
+
+def registry_entry(pod_id: str) -> dict:
+    """One registry entry, or {}."""
+    data = _registry_update(lambda d: False)
+    info = (data or {}).get(str(pod_id))
+    return dict(info) if isinstance(info, dict) else {}
+
+
 #: 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
@@ -4553,14 +4613,21 @@ def reconcile_pools(live_status: dict, candidates=None) -> set:
 
 
 def _known_pod_ids() -> set:
-    """Every pod id this DEPLOYMENT owns — not just this process's."""
+    """Every pod id this DEPLOYMENT owns — not just this process's.
+
+    A registry entry counts only while it still has an owner: ours always, and
+    a sibling's for as long as it keeps heartbeating (see
+    touch_registered_pods). An entry whose process is gone stops vouching for
+    its pod after REAP_GRACE_SECONDS, because otherwise nothing ever reaps it
+    and nothing manages it either — the pod simply bills.
+    """
     ids = set()
     with _pools_lock:
         for pool in _pools.values():
             with pool._cv:
                 ids.update(p.pod_id for p in pool.pods)
         ids.update(_provisioning_ids)
-    ids.update(registered_pod_ids())
+    ids.update(live_registered_pod_ids())
     return ids
 
 
@@ -4669,10 +4736,30 @@ def reap_orphans() -> int:
         if age < REAP_GRACE_SECONDS:
             continue                      # too young to be a leftover
         try:
+            info = registry_entry(pid)
             client.terminate_pod(pid)
             reaped += 1
+            # A pod the reaper kills billed real money, and until now it left no
+            # trace in the ledger: the spend was recorded by _terminate, which
+            # only runs for a pod a pool was tracking. An orphan's hours were
+            # therefore invisible in exactly the case where they are most
+            # surprising.
+            try:
+                from codai.api import runpod_ledger
+                rate = float(info.get("hourly_usd") or 0.0)
+                began = float(info.get("created") or info.get("at") or 0.0)
+                if rate and began:
+                    runpod_ledger.record_spend(
+                        str(info.get("pool") or "(orphan)"),
+                        (time.time() - began) / 3600.0 * rate,
+                        {"pod_id": pid, "gpu": "(orphan)", "reaped": True})
+            except Exception as exc:
+                print(f"[runpod-reaper] could not attribute {pid}'s spend: {exc}",
+                      flush=True)
+            unregister_pod(pid)
             print(f"[runpod-reaper] terminated STALE pod {pid} ({name}) — "
-                  "not tracked by any pool", flush=True)
+                  "not tracked by any pool and its owner stopped reporting",
+                  flush=True)
         except Exception as exc:
             print(f"[runpod-reaper] failed to terminate {pid}: {exc}", flush=True)
     return reaped
@@ -4688,6 +4775,14 @@ def _scaler_loop():
     while True:
         time.sleep(15)
         tick += 1
+        # Say we are still here, for every pod we own. Cheap (one locked file
+        # write), and what keeps a sibling's reaper off our pods — and, when
+        # this process dies, what stops ours being shielded forever.
+        if tick % 4 == 1:
+            try:
+                touch_registered_pods()
+            except Exception as exc:
+                print(f"[runpod] registry heartbeat: {exc}", flush=True)
         try:
             with _pools_lock:
                 pools = list(_pools.values())

docs/runpod.md

@@ -324,6 +324,24 @@ carrying this deployment's tag that is not in a live pool and not currently bein
 provisioned**. That covers pods orphaned by a crash, a kill -9, or a container
 restart mid-provision.
 
+"Not in a live pool" is decided from a shared registry file, and an entry in it
+**shields** the pod it names from reaping — a sibling engine must not kill a pod
+another process is managing. So every process heartbeats its own entries once a
+minute, and an entry goes quiet for `CODERAI_RUNPOD_REAP_GRACE` (1200 s) when
+its process is gone. Without that heartbeat an entry vouched for its pod for
+the life of the install: observed 2026-10-13, where an unattended upgrade
+restarted coderai and the surviving qwen38 pod was then both unmanaged (its
+pool's warm floor was 0 outside its schedule window, so nothing adopted it and
+nothing applied `idle_timeout_s` to it) and un-reapable (the dead process's
+entry still vouched for it). It billed on. A pod the reaper kills is now also
+written to the ledger, from the `created` timestamp in its entry — those hours
+used to be invisible, because only a pool-tracked termination recorded spend.
+
+One caveat that follows: a process whose maintenance loop is dead but which is
+still running stops heartbeating, and a sibling will reap its pods after the
+grace. That is the intended trade — twenty minutes of silence from the loop
+that is supposed to manage a pod is itself the failure.
+
 Consequences worth knowing:
 
 - Pods from **another** `deployment_id` are never touched, so two CoderAI

tests/test_runpod_registry_heartbeat.py

+"""A registry entry must stop vouching for a pod once its owner is gone.
+
+Live incident, 2026-10-13 on the Digesta orchestrator. An unattended upgrade
+restarted coderai at 21:17. A qwen38 pool pod (A40, $0.59/hr) survived the
+restart, and then nothing touched it:
+
+  * its pool's warm floor was 0 — outside its 08:00-20:00 schedule window — so
+    nothing provisioned, and the adoption path (`find_shared_pod`) only runs
+    when a pool actually needs a pod. No pool tracked it, so no pool applied
+    `idle_timeout_s` to it either;
+  * the stale-pod reaper skips anything in `_known_pod_ids()`, and the DEAD
+    process's registry entry was still in there — `registered_pod_ids()`
+    returned every key in the file regardless of who wrote it or whether that
+    process still existed.
+
+So the pod was simultaneously unmanaged and shielded, and billed on. `at` was
+written once at registration and never updated, so it could not tell the two
+cases apart. It is now a heartbeat: an owner refreshes it every scaler pass,
+and silence past REAP_GRACE_SECONDS lets the pod be reaped like any other
+leftover.
+"""
+
+import json
+import pathlib
+import sys
+import time
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from codai.api import runpod_worker as rw                          # noqa: E402
+
+
+def _registry(tmp_path, entries):
+    path = tmp_path / "runpod-pods.json"
+    path.write_text(json.dumps(entries))
+    return path
+
+
+def _patch_registry(monkeypatch, path):
+    monkeypatch.setattr(rw, "_pod_registry_path", lambda: str(path))
+
+
+# --- the heartbeat ----------------------------------------------------------
+
+def test_registering_records_creation_separately_from_last_seen(monkeypatch, tmp_path):
+    path = _registry(tmp_path, {})
+    _patch_registry(monkeypatch, path)
+    rw.register_pod("p1", pool_key="pool:x", url="https://h", hourly_usd=0.59)
+    info = json.loads(path.read_text())["p1"]
+    # "at" moves with the heartbeat, so the hours a reaped pod billed can only
+    # be computed from a timestamp that does not move.
+    assert "created" in info and info["created"] <= info["at"] + 1
+
+
+def test_the_heartbeat_refreshes_only_our_own_entries(monkeypatch, tmp_path):
+    old = time.time() - 5000
+    path = _registry(tmp_path, {
+        "mine": {"pool": "p", "proc": rw._PROC_ID, "at": old, "url": "u"},
+        "theirs": {"pool": "p", "proc": "999-deadbeef", "at": old, "url": "u"},
+    })
+    _patch_registry(monkeypatch, path)
+    assert rw.touch_registered_pods() == 1
+    data = json.loads(path.read_text())
+    assert data["mine"]["at"] > old + 4000
+    # Touching a sibling's entry would be claiming to be alive on its behalf.
+    assert data["theirs"]["at"] == old
+
+
+def test_the_heartbeat_survives_an_empty_registry(monkeypatch, tmp_path):
+    _patch_registry(monkeypatch, _registry(tmp_path, {}))
+    assert rw.touch_registered_pods() == 0
+
+
+# --- what still counts as known --------------------------------------------
+
+def test_our_own_entry_counts_however_old(monkeypatch, tmp_path):
+    # This process manages its own pods directly; the file is a courtesy to
+    # siblings, not the source of truth for us.
+    _patch_registry(monkeypatch, _registry(tmp_path, {
+        "mine": {"pool": "p", "proc": rw._PROC_ID, "at": time.time() - 99999}}))
+    assert rw.live_registered_pod_ids() == {"mine"}
+
+
+def test_a_siblings_fresh_entry_counts(monkeypatch, tmp_path):
+    _patch_registry(monkeypatch, _registry(tmp_path, {
+        "theirs": {"pool": "p", "proc": "999-x", "at": time.time() - 30}}))
+    assert rw.live_registered_pod_ids() == {"theirs"}
+
+
+def test_an_entry_whose_owner_went_quiet_stops_counting(monkeypatch, tmp_path):
+    # The incident: the owner was a process that no longer exists.
+    _patch_registry(monkeypatch, _registry(tmp_path, {
+        "orphan": {"pool": "p", "proc": "41-06956b37e2e9",
+                   "at": time.time() - (rw.REAP_GRACE_SECONDS + 60)}}))
+    assert rw.live_registered_pod_ids() == set()
+    # …while the raw view still has it, which is what the reaper used before.
+    assert rw.registered_pod_ids() == {"orphan"}
+
+
+def test_the_lapse_is_the_reap_grace_not_something_shorter(monkeypatch, tmp_path):
+    """A sibling whose scaler is between passes must not lose its pods.
+
+    The heartbeat runs every fourth 15 s pass, so the margin here is sixty
+    seconds against twenty minutes.
+    """
+    _patch_registry(monkeypatch, _registry(tmp_path, {
+        "theirs": {"pool": "p", "proc": "999-x",
+                   "at": time.time() - (rw.REAP_GRACE_SECONDS - 120)}}))
+    assert rw.live_registered_pod_ids() == {"theirs"}
+
+
+def test_a_malformed_entry_is_ignored_not_fatal(monkeypatch, tmp_path):
+    _patch_registry(monkeypatch, _registry(tmp_path, {"junk": "not a dict"}))
+    assert rw.live_registered_pod_ids() == set()
+
+
+def test_known_ids_uses_the_live_view(monkeypatch, tmp_path):
+    _patch_registry(monkeypatch, _registry(tmp_path, {
+        "orphan": {"pool": "p", "proc": "dead-1",
+                   "at": time.time() - (rw.REAP_GRACE_SECONDS + 60)}}))
+    # The whole bug in one assertion: this used to contain "orphan", and the
+    # reaper skips everything it contains.
+    assert "orphan" not in rw._known_pod_ids()
+
+
+# --- the entry the reaper reads to attribute the spend ----------------------
+
+def test_registry_entry_reads_one_entry(monkeypatch, tmp_path):
+    _patch_registry(monkeypatch, _registry(tmp_path, {
+        "p1": {"pool": "pool:x", "hourly_usd": 0.59, "created": 1000.0}}))
+    info = rw.registry_entry("p1")
+    assert info["pool"] == "pool:x" and info["hourly_usd"] == 0.59
+    assert rw.registry_entry("nope") == {}
+
+
+def test_the_reaper_attributes_and_forgets_what_it_kills(monkeypatch, tmp_path):
+    began = time.time() - 7200          # two hours
+    _patch_registry(monkeypatch, _registry(tmp_path, {
+        "orphan": {"pool": "pool:x", "proc": "dead-1", "hourly_usd": 0.59,
+                   "created": began,
+                   "at": began}}))
+
+    killed, spend = [], []
+
+    class _Client:
+        def list_pods(self):
+            return [{"id": "orphan", "name": "coderai-dep-model-abc",
+                     "status": "RUNNING", "uptime_s": 7200}]
+
+        def terminate_pod(self, pid):
+            killed.append(pid)
+
+    # reap_orphans imports these inside the function, so patch them at source.
+    import codai.api.runpod_client as rc
+    monkeypatch.setattr(rc, "RunpodClient", lambda *a, **k: _Client())
+    monkeypatch.setattr(rw, "_deploy_tag", lambda *a, **k: "dep")
+
+    import codai.api.runpod_ledger as ledger
+    monkeypatch.setattr(ledger, "record_spend",
+                        lambda key, cost, meta: spend.append((key, cost, meta)))
+
+    class _Acct:
+        enabled = True
+        api_key = "k"
+        deployment_id = "dep"
+
+    import codai.models.manager as mm
+    monkeypatch.setattr(mm, "get_active_runpod_config", lambda: _Acct())
+
+    rw.reap_orphans()
+
+    assert killed == ["orphan"]
+    # An orphan's hours used to be invisible: only _terminate recorded spend,
+    # and that runs for a pod a pool was tracking.
+    assert spend and spend[0][0] == "pool:x"
+    assert 1.1 < spend[0][1] < 1.2        # 2h at $0.59
+    assert spend[0][2]["reaped"] is True
+    # And it must not be reaped twice on the next sweep.
+    assert "orphan" not in json.loads((tmp_path / "runpod-pods.json").read_text())