runpod: a release that cannot be honoured yet is remembered, and a pod that

refuses our token is not an asset (v0.3.17)

Both from watching a real OCR run.

THE LOST INTENT. A client asked for the fleet back while a page was still being
read. The pods correctly stayed — work is never dropped to save money — but the
intent was then forgotten: the pool fell back to idle_timeout_s (900s there) and
billed for a fleet nobody had asked to keep. It took a second teardown call,
four minutes later, to actually release it. The pool now remembers: pods kept
only because they are working are retired the moment they fall idle, the same
"do not wait out the timeout" treatment a closed schedule window already gets.
Asking for warm pods again cancels it.

Also fixes the reply and the log, which said "floor 1" when the 1 came from work
in flight. That reads as though the operator had asked for the pod — the opposite
of what it meant. Demand and configuration are now reported separately
("needed" and "floor"), because they are two different reasons a pod survived.

THE ADOPTED POD THAT 401s. After a restart the pool adopts the pod it left
running instead of renting another, and carries that pod's bearer token across
through the registry — when one was recorded. With none recorded it fell back to
our own token and never checked the result, so a pod launched with a different
one answered 401 to every request for the rest of its paid life. Seen today,
right after a restart: "reusing the pod already running" followed by 401 on
every OCR page. The health probe that found it proves nothing here — a coderai
pod answers /healthz to anybody.

Adoption now probes an AUTHENTICATED path with the token it would use, and a pod
that refuses it is terminated rather than adopted: with one registry per
deployment (0.3.15) an unrecorded key is a lost key, so that pod can never serve
anyone, and paying for it is the only outcome worse than killing it. An
unreachable pod is NOT treated as unauthorised — that stays with the health
machinery, which has a budget for it.
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.16"
+__version__ = "0.3.17"
 __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

@@ -2128,6 +2128,11 @@ class RunpodPodPool:
         #: waits out scale_down_after_s from here, so a momentary dip in a burst
         #: does not cost a pod that is wanted again seconds later.
         self._surplus_since = 0.0
+        #: A client said "give them back" and something was still working, so the
+        #: pods could not go yet. Without remembering it the intent was simply
+        #: lost: the pool fell back to idle_timeout_s (15 minutes on the Digesta
+        #: orchestrator) and billed for a fleet nobody had asked to keep.
+        self._release_pending = False
         # One log line per window transition, not one every scaler tick.
         self._sched_closed_logged = False
 
@@ -2373,6 +2378,28 @@ class RunpodPodPool:
     # NEXT candidate rather than failing the request.
     MAX_PROVISION_ATTEMPTS = 3
 
+    def _pod_answers_us(self, url: str) -> bool:
+        """Whether the pod accepts the token we would use for it.
+
+        Probes an AUTHENTICATED path, because the unauthenticated one proves
+        nothing: a coderai pod answers /healthz to anybody, which is how a pod
+        that 401s every real request passed the probe that adopted it.
+        """
+        base = (url or "").rstrip("/")
+        if not base:
+            return False
+        try:
+            from codai.api import pod_http
+            headers = ({"Authorization": f"Bearer {self.api_key}"}
+                       if self.api_key else {})
+            reply = pod_http.get(base + "/v1/models", headers=headers, timeout=20)
+        except Exception as exc:
+            # Unreachable is not "unauthorised": leave that to the health
+            # machinery rather than destroying a pod over one failed probe.
+            print(f"[runpod] token probe against {base} failed: {exc}", flush=True)
+            return True
+        return reply.status_code not in (401, 403)
+
     def _adopt_shared_pod(self) -> bool:
         """Take over a pod already running for this pool, instead of renting.
 
@@ -2396,6 +2423,27 @@ class RunpodPodPool:
                     return False
             self.api_key = shared_key
             print(f"[runpod] adopting pod {shared_id}'s bearer token", flush=True)
+        if not self._pod_answers_us(shared_url):
+            # A coderai pod's /healthz needs no token, so the probe that found
+            # this pod says nothing about whether we can USE it. With no key
+            # recorded we fall back to our own — and if that is not the token the
+            # pod was launched with, it answers 401 to every request for the rest
+            # of its paid life. Seen today: a pod adopted after a restart 401'd
+            # every OCR page.
+            #
+            # One registry per deployment means no recorded key is no key at all:
+            # the token is lost and the pod can never serve anyone. Paying for it
+            # is the only outcome worse than terminating it.
+            print(f"[runpod] pod {shared_id} refuses our bearer token (its own "
+                  f"was never recorded) — terminating rather than paying for a "
+                  f"pod nobody can use", flush=True)
+            try:
+                from codai.api.runpod_client import RunpodClient as _RC
+                unregister_pod(shared_id)
+                _RC(self.account).terminate_pod(shared_id)
+            except Exception as exc:
+                print(f"[runpod] could not terminate {shared_id}: {exc}", flush=True)
+            return False
         print(f"[runpod] reusing the pod already running for {self.model_key!r} "
               f"({shared_id}) instead of renting another", flush=True)
         handle = PodHandle(pod_id=shared_id, url=shared_url,
@@ -2796,6 +2844,10 @@ class RunpodPodPool:
         cap = max(1, int(self.mcfg.max_pods))
         want = max(0, int(pods))
         granted = min(want, cap)
+        if granted:
+            # Asking for warm pods cancels a pending release: the client changed
+            # its mind, and retiring what it just asked for would be absurd.
+            self._release_pending = False
         ttl = max(60, int(getattr(self.mcfg, "client_scale_ttl_s", 1800) or 1800))
         with self._cv:
             self._client_floor = granted
@@ -2902,11 +2954,24 @@ class RunpodPodPool:
                     draining.append(victim.pod_id)
             self._surplus_since = 0.0
             kept = len([p for p in self.pods if p.healthy and not p.draining])
+            # Anything still here above the operator's own floor is here because
+            # it is WORKING, not because anyone wants it warm. Remember that, so
+            # it goes the moment it falls idle instead of waiting out
+            # idle_timeout_s.
+            floor = self.effective_min_pods()
+            self._release_pending = bool(kept > floor or draining)
         print(f"[runpod] {self.model_key!r}: client released the fleet — "
               f"{len(terminated)} terminated, {len(draining)} draining, "
-              f"{kept} kept (floor {need})", flush=True)
+              f"{kept} kept ({need} still needed by work in flight, warm floor "
+              f"{floor})"
+              + (" — will retire as soon as they fall idle"
+                 if self._release_pending else ""), flush=True)
         return {"terminated": terminated, "draining": draining, "kept": kept,
-                "floor": need}
+                # `needed` is demand; `floor` is configuration. Reporting demand
+                # AS "floor" read as though the operator had asked for the pod,
+                # which is the opposite of what it meant.
+                "needed": need, "floor": floor,
+                "retire_when_idle": self._release_pending}
 
     def _scale_down(self, now: float) -> None:
         """Retire one surplus pod per pass, once the surplus has persisted.
@@ -2992,6 +3057,11 @@ class RunpodPodPool:
                       f"({schedule_state(self.mcfg).get('window')}) — warm floor 0",
                       flush=True)
                 self._sched_closed_logged = True
+        elif self._release_pending:
+            # A client asked for the fleet back while it was busy. Hold nothing
+            # once the work drains — the same "do not wait out the timeout"
+            # treatment a closed schedule window gets, for the same reason.
+            idle_to = min(idle_to, 1) if idle_to else 1
         else:
             self._sched_closed_logged = False
         with self._cv:
@@ -3009,6 +3079,8 @@ class RunpodPodPool:
         # Keep min_pods warm.
         with self._cv:
             healthy_n = len([p for p in self.pods if p.healthy])
+            if self._release_pending and healthy_n <= keep_min:
+                self._release_pending = False          # satisfied
         if healthy_n < keep_min and not self._closed and not self._provisioning:
             # One already on the way (from here, from ensure_ready, or from a
             # sibling engine sharing this pool) is one pod, not none: topping up

docs/remote-execution.md

@@ -785,8 +785,16 @@ is inferring from a momentary reading and a cold pod costs minutes to replace. A
 client that has just finished a batch is not inferring — it knows no more work is
 coming — so for that call the window and the one-per-pass limit are both skipped,
 and the reply says what happened: `{"terminated": [...], "draining": [...],
-"kept": N, "floor": N}`. It still cannot scale through the operator's `min_pods`,
-and a pod with requests on it drains rather than dying under them.
+"kept": N, "needed": N, "floor": N, "retire_when_idle": bool}`. `needed` is what
+the work in flight still requires; `floor` is what the configuration keeps warm —
+two different reasons a pod survived, and worth telling apart when you are asking
+why one did.
+
+It still cannot scale through the operator's `min_pods`, and a pod with requests
+on it drains rather than dying under them. **The intent is remembered**: if the
+release could not be honoured because pods were working, `retire_when_idle` comes
+back true and those pods are retired the moment they fall idle, rather than
+waiting out `idle_timeout_s`. Asking for warm pods again cancels it.
 
 **Which surplus pod goes.** By default the least loaded, and among equals the one
 used least recently — its prefix cache is the coldest, so retiring it throws away

tests/test_runpod_reaper_grace.py

@@ -103,3 +103,81 @@ def test_the_resolved_path_is_logged_so_a_split_is_visible(capsys, monkeypatch,
     rw._REGISTRY_PATH_LOGGED = False
     rw._pod_registry_path()
     assert "pod registry:" in capsys.readouterr().out
+
+
+# ------------------------------- a pod we cannot talk to is not an asset #
+# After a restart the pool adopts the pod it left running rather than renting
+# another. The pod only answers the bearer token it was LAUNCHED with, and the
+# registry carries that token across for exactly this reason — but when no key
+# was recorded the code fell back to our own and never checked the result. Seen
+# today, right after a restart: "reusing the pod already running" followed by
+# "Inference error: 401 — Invalid API key" on every page, for the rest of its
+# paid life. The probe that found the pod proves nothing about this: a coderai
+# pod answers /healthz to anybody.
+
+
+class _Reply:
+    def __init__(self, status):
+        self.status_code = status
+
+
+def _pool_for_probe(monkeypatch, status):
+    import threading
+    pool = rw.RunpodPodPool.__new__(rw.RunpodPodPool)
+    pool._cv = threading.Condition()
+    pool.model_key = "m"
+    pool.api_key = "ours"
+    pool.pods = []
+    from codai.api import pod_http
+    seen = {}
+
+    def _get(url, headers=None, timeout=None, **kw):
+        seen["url"] = url
+        seen["headers"] = headers or {}
+        return _Reply(status)
+    monkeypatch.setattr(pod_http, "get", _get)
+    return pool, seen
+
+
+def test_an_authenticated_path_is_what_gets_probed(monkeypatch):
+    """/healthz needs no token on a coderai pod, so probing it answers the wrong
+    question."""
+    pool, seen = _pool_for_probe(monkeypatch, 200)
+    assert pool._pod_answers_us("https://10.0.0.5:8000") is True
+    assert seen["url"].endswith("/v1/models")
+    assert seen["headers"]["Authorization"] == "Bearer ours"
+
+
+def test_a_401_means_the_pod_is_not_ours_to_use(monkeypatch):
+    pool, _ = _pool_for_probe(monkeypatch, 401)
+    assert pool._pod_answers_us("https://10.0.0.5:8000") is False
+
+
+def test_a_403_counts_too(monkeypatch):
+    pool, _ = _pool_for_probe(monkeypatch, 403)
+    assert pool._pod_answers_us("https://10.0.0.5:8000") is False
+
+
+def test_unreachable_is_not_unauthorised(monkeypatch):
+    """One failed probe must not destroy a pod that is merely still booting —
+    that is the health machinery's job, with its own budget."""
+    import threading
+    from codai.api import pod_http
+    pool = rw.RunpodPodPool.__new__(rw.RunpodPodPool)
+    pool._cv = threading.Condition()
+    pool.api_key = "ours"
+
+    def _boom(*a, **kw):
+        raise OSError("connection refused")
+    monkeypatch.setattr(pod_http, "get", _boom)
+    assert pool._pod_answers_us("https://10.0.0.5:8000") is True
+
+
+def test_adoption_refuses_and_terminates_a_pod_that_rejects_us():
+    """Paying for a pod nobody can use is the only outcome worse than killing
+    it, and with one registry per deployment an unrecorded key is a lost key."""
+    import inspect
+    src = inspect.getsource(rw.RunpodPodPool._adopt_shared_pod)
+    assert "_pod_answers_us" in src
+    assert "terminate_pod" in src
+    assert "return False" in src

tests/test_runpod_scale_down.py

@@ -362,3 +362,62 @@ def test_the_surplus_timer_is_reset_so_the_next_pass_starts_clean():
     pool._surplus_since = 123.0
     pool.shed_now()
     assert pool._surplus_since == 0.0
+
+
+# ------------------------------- a release that could not be honoured is kept #
+# A client asked for the fleet back while a page was still being read. The pods
+# could not go yet — correctly, work is never dropped to save money — but the
+# intent was then LOST: the pool fell back to idle_timeout_s (900s on the Digesta
+# orchestrator) and billed for a fleet nobody had asked to keep. Seen live:
+# "0 terminated, 0 draining, 1 kept", and the pod was still there four minutes
+# later with nothing running on it.
+
+
+def test_a_release_blocked_by_work_is_remembered():
+    pool = _pool(threshold=10, pods=[_pod("busy", 2)])
+    out = pool.shed_now()
+    assert out["kept"] == 1
+    assert out["retire_when_idle"] is True
+    assert pool._release_pending is True
+
+
+def test_demand_is_reported_as_demand_not_as_a_floor():
+    """The log and the reply said "floor 1" when the 1 came from work in flight,
+    which reads as though the operator had asked for the pod."""
+    pool = _pool(threshold=10, min_pods=0, pods=[_pod("busy", 2)])
+    out = pool.shed_now()
+    assert out["needed"] == 1          # demand
+    assert out["floor"] == 0           # configuration
+
+
+def test_nothing_is_remembered_when_the_release_fully_succeeded():
+    pool = _pool(pods=[_pod("idle", 0)])
+    out = pool.shed_now()
+    assert out["retire_when_idle"] is False
+    assert pool._release_pending is False
+
+
+def test_asking_for_pods_again_cancels_the_pending_release():
+    """The client changed its mind; retiring what it just asked for is absurd."""
+    pool = _pool(threshold=10, pods=[_pod("busy", 2)])
+    pool.shed_now()
+    assert pool._release_pending is True
+    pool.set_client_floor(2)
+    assert pool._release_pending is False
+
+
+def test_releasing_to_zero_leaves_the_intent_standing():
+    pool = _pool(threshold=10, pods=[_pod("busy", 2)])
+    pool.shed_now()
+    pool.set_client_floor(0)
+    assert pool._release_pending is True
+
+
+def test_a_pending_release_collapses_the_idle_timeout():
+    """The mechanism: the pod is retired as soon as it falls idle instead of
+    waiting out idle_timeout_s — the same treatment a closed schedule window
+    already gets, for the same reason."""
+    import inspect
+    src = inspect.getsource(rw.RunpodPodPool.maintain)
+    assert "elif self._release_pending:" in src
+    assert "idle_to = min(idle_to, 1)" in src