an API that cannot be asked is not a pod that failed (v0.3.99)

A request for an embedding model rented a pod, which began pulling a 9.65 GB
image that had been pushed twenty minutes earlier — a five-minute cold pull, and
the ORDINARY case for the first pod on a newly tagged image. Two minutes in,
RunPod's control plane answered HTTP 503 once. Three defects then fired in
sequence, and the pod that was coming up fine became an untracked GPU billing at
$1.09/hr:

  21:36:58  pod created — booting
  21:38:00  still no port after 62s (status=RUNNING) — image pull in progress
  21:38:50  boot FAILED: RunPod API HTTP 503: Service Unavailable
  21:38:53  POST /v1/embeddings  502

1. THE BOOT WAIT CONDEMNED THE POD FOR THE API'S FAILURE. Any exception out of
   the wait is read as "this pod failed to boot", and `get_pod` raised straight
   through the poll loop. A 503 from the control plane means the question went
   unanswered; it says nothing about the machine. RunpodTransient (a subclass,
   so every existing `except RunpodError` is unchanged) now marks 5xx/408/425/429
   and transport errors, and the wait keeps polling to the deadline the caller
   set, saying once that the API is quiet and again when it comes back. A wait
   that never got a single answer raises transient too — "unknown, not
   known-bad" — rather than handing the teardown path a verdict nobody reached.
   401 and 4xx stay final: those will say the same thing in five seconds.

2. THE TEARDOWN RAN THROUGH THE SAME DEAD API, INSIDE `except Exception: pass`.
   The attempt most likely to fail is the one that matters most, and its failure
   was dropped on the floor — no log line anywhere named the pod. terminate_pod
   now retries with backoff while the error is transient, and the boot handler
   prints what is still billing, at what rate, and how to stop it. A pod that
   could not be terminated KEEPS its registry entry: that entry is the only
   record it exists. `_give_back` had the mirror image of the bug — it
   unregistered unconditionally, throwing away the record of a live pod — and
   now forgets only what is actually gone. A boot that failed also ledgers the
   minutes it burned; no pool ever held that pod, so nothing else would.

3. THE REAPER COULD NEVER HAVE CLEANED UP AFTER EITHER. Pod age came from
   `runtime.uptimeInSeconds`, which is the CURRENT RUN's uptime and resets on
   every container restart. The leaked pod reported 234s while it was 11 minutes
   old, and its uptime reset every few minutes, so it could never cross the
   1200s grace — the pod that most needs reaping, one nothing tracks that is
   failing over and over, was the one pod the reaper was structurally blind to.
   `lastStatusChange` carries the rental time ("Rented by User: Thu Oct 15 2026
   19:36:58 GMT+0000") and was already in the response; list_pods now asks for
   it. Age is the larger of rental age and uptime — uptime cannot exceed the
   age, so the pair is still a true lower bound, and a pod whose stamp will not
   parse is no worse off than before. A future stamp is refused rather than
   making every pod ancient, and no age at all is still None: the reaper fails
   closed on that.

Found because the image update was the trigger: a new tag is cold in every
datacenter, so the first boot after every release spends minutes exposed to
exactly this. The next pod proved the image itself was fine — 301s of pull, 22s
of warm-up, 200 OK.

27 tests. Suite: 3132 passed, the same 13 known-stale failures.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KnfkxBpcZ1v6ypRFFyhrVM

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.98"
+__version__ = "0.3.99"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/api/runpod_client.py

@@ -38,6 +38,23 @@ class RunpodError(RuntimeError):
     """A RunPod API call failed (network, auth, or GraphQL error)."""
 
 
+class RunpodTransient(RunpodError):
+    """The API could not be ASKED — it says nothing about the pod.
+
+    A 503 from RunPod's control plane, a timeout, a dropped connection: the
+    question never got an answer. That is a different fact from "the pod is
+    bad", and conflating the two cost a live pod once already. A boot wait got
+    HTTP 503 while a 9.6 GB image was still being pulled, read it as a boot
+    failure, tore the pod down through the SAME unavailable API — which also
+    failed — and left an untracked GPU billing until someone looked at the
+    account. The request that started it answered 502 while the machine it had
+    paid for was still coming up.
+
+    A subclass, so every existing ``except RunpodError`` keeps working; callers
+    that can afford to wait catch this one first and retry.
+    """
+
+
 #: Said once per process, not once per pod: the scaler would otherwise repeat it
 #: every boot.
 _GLOBAL_VOLUME_WARNED = False
@@ -137,11 +154,19 @@ class RunpodClient:
                               timeout=timeout,
                               headers={"Content-Type": "application/json"})
         except requests.RequestException as exc:
-            raise RunpodError(f"RunPod API unreachable ({_mask(self._api_key)}): {exc}")
+            # No answer at all — transient by definition.
+            raise RunpodTransient(f"RunPod API unreachable ({_mask(self._api_key)}): {exc}")
         if r.status_code == 401:
             raise RunpodError(f"RunPod rejected the API key ({_mask(self._api_key)}) — 401.")
         if r.status_code >= 400:
-            raise RunpodError(f"RunPod API HTTP {r.status_code}: {r.text[:300]}")
+            # 5xx is RunPod's own trouble and 408/429 is "ask again"; neither is
+            # a statement about what we asked for. A 4xx otherwise IS: a bad id,
+            # a malformed query, a refused request — retrying that only wastes
+            # the deadline.
+            cls = (RunpodTransient
+                   if r.status_code >= 500 or r.status_code in (408, 425, 429)
+                   else RunpodError)
+            raise cls(f"RunPod API HTTP {r.status_code}: {r.text[:300]}")
         try:
             body = r.json()
         except ValueError:
@@ -463,19 +488,49 @@ class RunpodClient:
         except RunpodError:
             raise
 
-    def terminate_pod(self, pod_id: str) -> None:
+    def terminate_pod(self, pod_id: str, attempts: int = 4) -> None:
+        """Stop a pod, retrying while the API is merely unreachable.
+
+        The one call that must not give up quietly. Every other failed call
+        costs a retry; this one costs a GPU that bills until a human notices.
+        The case that matters is precisely the one where a single try fails:
+        the control plane is having trouble, which is also what made the caller
+        decide to tear the pod down in the first place — so the teardown meets
+        the same 503 that caused it.
+
+        Raises after the last attempt. A caller that swallows that exception is
+        choosing to leak a pod, and should say so in the log instead.
+        """
         q = "mutation($input: PodTerminateInput!) { podTerminate(input: $input) }"
-        self._gql(q, {"input": {"podId": pod_id}}, timeout=30.0)
+        delay = 2.0
+        for attempt in range(1, max(1, attempts) + 1):
+            try:
+                self._gql(q, {"input": {"podId": pod_id}}, timeout=30.0)
+                return
+            except RunpodTransient as exc:
+                if attempt >= attempts:
+                    raise
+                print(f"[runpod] terminate {pod_id}: {exc} — retrying in "
+                      f"{delay:.0f}s ({attempt}/{attempts})", flush=True)
+                time.sleep(delay)
+                delay = min(delay * 2, 15.0)
 
     def list_pods(self) -> list:
         """List the account's pods: [{id, name, status, cost_per_hr, image}]."""
         # runtime.uptimeInSeconds is what makes the reaper's grace period real:
         # without an age every pod looked undatable, and the "never reap a pod
         # younger than GRACE" backstop abstained on all of them.
+        # lastStatusChange carries the RENTAL time ("Rented by User: Thu Oct 15
+        # 2026 19:36:58 GMT+0000"), which is the pod's real age. uptime alone is
+        # the current RUN's age and resets on every container restart, so a pod
+        # stuck in a restart loop reported 234s when it was 11 minutes old and
+        # could never cross the reaper's 20-minute grace. The one pod that most
+        # needs reaping — one nothing is tracking, failing over and over — was
+        # the one pod the reaper could never see.
         q = """
         query Pods {
           myself {
-            pods { id name desiredStatus costPerHr imageName
+            pods { id name desiredStatus costPerHr imageName lastStatusChange
                    runtime { uptimeInSeconds } }
           }
         }"""
@@ -487,6 +542,7 @@ class RunpodClient:
                 "id": p.get("id"), "name": p.get("name"),
                 "status": p.get("desiredStatus"), "cost_per_hr": p.get("costPerHr"),
                 "image": p.get("imageName"),
+                "last_change": p.get("lastStatusChange") or "",
             }
             # Absent while a pod is still being scheduled, which is exactly when
             # it must not be mistaken for a leftover — the reaper now treats a
@@ -558,8 +614,29 @@ class RunpodClient:
         started = time.time()
         next_report = started + 60.0
         last = {}
+        # A boot wait is MINUTES long — a cold machine pulling a 10 GB image is
+        # the normal case after every image release — so it is the call most
+        # likely to be in flight when the control plane hiccups. Asking and not
+        # being answered is not the pod failing; keep asking until the deadline
+        # the caller set, and only then say the API was the thing that broke.
+        api_down_since = 0.0
+        last_transient = None
         while time.time() < deadline:
-            info = self.get_pod(pod_id)
+            try:
+                info = self.get_pod(pod_id)
+            except RunpodTransient as exc:
+                last_transient = exc
+                if not api_down_since:
+                    api_down_since = time.time()
+                    print(f"[runpod] pod {pod_id}: RunPod API not answering "
+                          f"({exc}) — still waiting; this says nothing about "
+                          f"the pod", flush=True)
+                time.sleep(poll_every)
+                continue
+            if api_down_since:
+                print(f"[runpod] pod {pod_id}: RunPod API answering again after "
+                      f"{time.time() - api_down_since:.0f}s", flush=True)
+                api_down_since = 0.0
             last = info
             if info.get("status") in ("TERMINATED", "FAILED"):
                 raise RunpodError(f"Pod {pod_id} entered status {info.get('status')} before ready.")
@@ -593,6 +670,13 @@ class RunpodClient:
                       f"(status={info.get('status')} uptime={info.get('uptime_s')}s) "
                       f"— {why}", flush=True)
             time.sleep(poll_every)
+        if api_down_since and not last:
+            # The whole wait went by without one answer. Blaming the pod here
+            # would be a guess, and the caller tears down on a boot failure.
+            raise RunpodTransient(
+                f"Pod {pod_id}: RunPod's API did not answer once in "
+                f"{ready_timeout:.0f}s ({last_transient}) — the pod's state is "
+                f"unknown, not known-bad.")
         raise RunpodError(
             f"Pod {pod_id} did not expose port {port} within {ready_timeout}s "
             f"(last status={last.get('status')} uptime={last.get('uptime_s')}s). "

codai/api/runpod_worker.py

@@ -3372,17 +3372,26 @@ class RunpodPodPool:
                   f"while it was still booting ({why}) — terminating it now "
                   f"rather than paying out the boot for a pod nobody wants",
                   flush=True)
+            gone = True
             try:
                 client.terminate_pod(pod_id)
             except Exception as exc:
-                print(f"[runpod] could not terminate {pod_id}: {exc}", flush=True)
+                gone = False
+                print(f"[runpod] !! could not terminate {pod_id}: {exc} — it is "
+                      f"STILL BILLING; the reaper will take it, or stop it now "
+                      f"with POST /v1/runpod/release", flush=True)
             with _pools_lock:
                 _provisioning_ids.discard(pod_id)
                 _provisioning_info.pop(pod_id, None)
-            try:
-                unregister_pod(pod_id)
-            except Exception:
-                pass
+            # Only forget a pod that is actually gone. Dropping the entry after
+            # a FAILED terminate throws away the registry's record of a pod
+            # that is still running, leaving the name prefix as the only thing
+            # that can still identify it as ours.
+            if gone:
+                try:
+                    unregister_pod(pod_id)
+                except Exception:
+                    pass
 
         # The caller's narrowing, if a lease is in force. Read per provision so
         # a lapsed lease stops constraining the very next pod.
@@ -3481,8 +3490,31 @@ class RunpodPodPool:
                 self._dump_pod_logs(client, pod_id, "boot failed", url)
                 try:
                     client.terminate_pod(pod_id)
-                except Exception:
-                    pass
+                except Exception as term_exc:
+                    # NOT `pass`. This is the moment a pod becomes a leak: the
+                    # teardown runs through the same API that just failed, so
+                    # the attempt most likely to fail is the one that matters,
+                    # and a silent drop left a GPU billing with nothing in any
+                    # log naming it. Say it in the words of the thing that went
+                    # wrong, and leave the registry entry alone — the reaper
+                    # owns a pod it can still see, and that entry is now the
+                    # only record that this pod exists.
+                    print(f"[runpod] !! pod {pod_id} COULD NOT BE TERMINATED "
+                          f"after its boot failed: {term_exc}", flush=True)
+                    print(f"[runpod] !! it is still billing at "
+                          f"${sel['price']:.2f}/hr. The stale-pod reaper will "
+                          f"take it once it is past the grace period; to stop "
+                          f"it now: POST /v1/runpod/release {{\"pod_id\": "
+                          f"\"{pod_id}\"}} or {console}", flush=True)
+                else:
+                    # A boot that failed still ran a GPU for the minutes it
+                    # took to fail, and no pool ever held this pod, so nothing
+                    # else would ever ledger those minutes.
+                    try:
+                        close_bill_from_registry(pod_id, "boot failed")
+                    except Exception:
+                        pass
+                    unregister_pod(pod_id)
                 with _pools_lock:
                     _provisioning_ids.discard(pod_id)
                     _provisioning_info.pop(pod_id, None)
@@ -5527,25 +5559,70 @@ _POD_LOG_MAX_BYTES = int(os.environ.get("CODERAI_POD_LOG_MAX_BYTES", 8 * 1024 *
 _POD_LOG_KEEP_LINES = int(os.environ.get("CODERAI_POD_LOG_KEEP_LINES", 20000))
 
 
-def _pod_age_seconds(pod: dict):
-    """Seconds since the pod was created, or None when RunPod does not say."""
-    for key in ("uptime_s", "uptimeSeconds", "runtimeSeconds"):
-        val = pod.get(key)
-        if isinstance(val, (int, float)) and val >= 0:
-            return float(val)
+def _rental_age_seconds(pod: dict):
+    """Seconds since the pod was RENTED, from RunPod's own timestamps, or None.
+
+    ``lastStatusChange`` is a sentence, not a field: "Rented by User: Thu Oct 15
+    2026 19:36:58 GMT+0000 (Coordinated Universal Time)". The date after the
+    colon is what we want, and only when the sentence is about the rental —
+    a later status change would date the change, not the pod.
+    """
+    from datetime import datetime, timezone
     for key in ("createdAt", "created_at"):
         val = pod.get(key)
         if not val:
             continue
         try:
-            from datetime import datetime, timezone
             return (datetime.now(timezone.utc) - datetime.fromisoformat(
                 str(val).replace("Z", "+00:00"))).total_seconds()
         except Exception:
             continue
+    text = str(pod.get("last_change") or pod.get("lastStatusChange") or "")
+    if "rented" not in text.lower() or ":" not in text:
+        return None
+    stamp = text.split(":", 1)[1].split("(")[0].strip()
+    for fmt in ("%a %b %d %Y %H:%M:%S GMT%z", "%a %b %d %Y %H:%M:%S %Z",
+                "%a %b %d %Y %H:%M:%S"):
+        try:
+            when = datetime.strptime(stamp, fmt)
+        except ValueError:
+            continue
+        if when.tzinfo is None:
+            when = when.replace(tzinfo=timezone.utc)
+        age = (datetime.now(timezone.utc) - when).total_seconds()
+        # A clock that disagrees would otherwise reap on arrival.
+        return age if age >= 0 else None
     return None
 
 
+def _pod_age_seconds(pod: dict):
+    """Seconds since the pod was created, or None when RunPod does not say.
+
+    The RENTAL time first, uptime only as a floor. uptime is how long the
+    CURRENT run has lasted, and it resets whenever the container restarts: a
+    pod that fails and restarts every few minutes reported 234s while it was 11
+    minutes old, so it could never reach REAP_GRACE_SECONDS and the reaper
+    skipped it on every sweep, forever. Observed live — a pod abandoned
+    mid-boot billed at $1.09/hr with nothing tracking it and no sweep that
+    would ever have caught it.
+
+    They are combined rather than swapped: uptime cannot exceed the age, so the
+    larger of the two is still a true lower bound on how long this pod has
+    existed, and keeping it means a pod whose rental stamp is unparsable is no
+    worse off than before.
+    """
+    ages = []
+    rental = _rental_age_seconds(pod)
+    if rental is not None:
+        ages.append(rental)
+    for key in ("uptime_s", "uptimeSeconds", "runtimeSeconds"):
+        val = pod.get(key)
+        if isinstance(val, (int, float)) and val >= 0:
+            ages.append(float(val))
+            break
+    return max(ages) if ages else None
+
+
 def reap_orphans() -> int:
     """Terminate RunPod pods tagged as OURS that no live pool is tracking — stale
     pods from a crash/restart that would otherwise bill silently. Only touches pods

tests/test_runpod_transient_api.py

+"""A control plane that could not be ASKED is not a pod that failed.
+
+Observed live 2026-10-15. A request for an embedding model rented a pod, which
+began pulling a freshly pushed 9.65 GB image — a five-minute cold pull, the
+normal case for the first pod after an image release. Two minutes in, RunPod's
+API answered HTTP 503 once. Three things then went wrong in a row:
+
+  1. the boot wait raised, and the handler reads any exception as "this pod
+     failed to boot" — so a healthy pod, mid-pull, was condemned;
+  2. the teardown went out through the SAME unavailable API, failed, and was
+     swallowed by `except Exception: pass`;
+  3. the stale-pod reaper could never clean up after it, because pod age came
+     from `runtime.uptimeInSeconds` — the CURRENT run's uptime, which resets on
+     every restart. The pod reported 234s while it was 11 minutes old and could
+     never cross the 20-minute grace.
+
+Net: an L40S billing at $1.09/hr that nothing tracked and no sweep would ever
+have found, and a 502 to the caller whose pod was still coming up.
+"""
+
+import time
+import types
+
+import pytest
+
+from codai.api import runpod_worker
+from codai.api.runpod_client import RunpodClient, RunpodError, RunpodTransient
+
+
+# ------------------------------------------------------- transient vs final #
+class _Resp:
+    def __init__(self, status, text='{"message":"nope"}'):
+        self.status_code, self.text = status, text
+
+    def json(self):
+        import json
+        return json.loads(self.text)
+
+
+def _client(monkeypatch, responses):
+    """A client whose transport returns the given responses in order."""
+    cfg = types.SimpleNamespace(api_key="k", api_base="", rest_base="")
+    c = RunpodClient(cfg)
+    seq = list(responses)
+
+    def _post(*a, **k):
+        r = seq.pop(0)
+        if isinstance(r, Exception):
+            raise r
+        return r
+
+    import requests
+    monkeypatch.setattr(requests, "post", _post)
+    return c
+
+
+@pytest.mark.parametrize("status", [500, 502, 503, 504, 408, 429])
+def test_a_server_side_status_is_transient(monkeypatch, status):
+    c = _client(monkeypatch, [_Resp(status)])
+    with pytest.raises(RunpodTransient):
+        c.get_pod("p1")
+
+
+@pytest.mark.parametrize("status", [400, 403, 404, 422])
+def test_a_client_side_status_is_final(monkeypatch, status):
+    """Retrying a bad id or a malformed query only burns the deadline."""
+    c = _client(monkeypatch, [_Resp(status)])
+    with pytest.raises(RunpodError) as e:
+        c.get_pod("p1")
+    assert not isinstance(e.value, RunpodTransient)
+
+
+def test_an_unreachable_api_is_transient(monkeypatch):
+    import requests
+    c = _client(monkeypatch, [requests.ConnectionError("no route")])
+    with pytest.raises(RunpodTransient):
+        c.get_pod("p1")
+
+
+def test_a_rejected_key_is_not_transient(monkeypatch):
+    """401 will say 401 again in five seconds; retrying hides a real problem."""
+    c = _client(monkeypatch, [_Resp(401)])
+    with pytest.raises(RunpodError) as e:
+        c.get_pod("p1")
+    assert not isinstance(e.value, RunpodTransient)
+
+
+def test_transient_is_still_a_runpod_error():
+    """Every existing `except RunpodError` must keep catching it."""
+    assert issubclass(RunpodTransient, RunpodError)
+
+
+# ------------------------------------------------------------- the boot wait #
+def _ready_pod():
+    return {"id": "p1", "status": "RUNNING", "uptime_s": 10, "ready": True,
+            "ports": [{"ip": "1.2.3.4", "isIpPublic": True, "privatePort": 8000,
+                       "publicPort": 22031, "type": "http"}]}
+
+
+def test_the_boot_wait_rides_out_a_transient_api_failure(monkeypatch):
+    """The pod was fine; only the question went unanswered."""
+    cfg = types.SimpleNamespace(api_key="k", api_base="", rest_base="")
+    c = RunpodClient(cfg)
+    calls = []
+
+    def _get_pod(pod_id):
+        calls.append(pod_id)
+        if len(calls) <= 3:
+            raise RunpodTransient("RunPod API HTTP 503")
+        return _ready_pod()
+
+    monkeypatch.setattr(c, "get_pod", _get_pod)
+    monkeypatch.setattr(time, "sleep", lambda *_: None)
+    url = c.wait_ready("p1", 8000, ready_timeout=60, poll_every=0.01)
+    assert url
+    assert len(calls) == 4, "it must keep asking, not give up on the first 503"
+
+
+def test_a_wait_that_never_got_an_answer_says_so(monkeypatch):
+    """The caller tears down on a boot failure, so the distinction decides
+    whether a healthy pod is destroyed."""
+    cfg = types.SimpleNamespace(api_key="k", api_base="", rest_base="")
+    c = RunpodClient(cfg)
+    monkeypatch.setattr(c, "get_pod",
+                        lambda *_: (_ for _ in ()).throw(RunpodTransient("503")))
+    monkeypatch.setattr(time, "sleep", lambda *_: None)
+    with pytest.raises(RunpodTransient) as e:
+        c.wait_ready("p1", 8000, ready_timeout=0.05, poll_every=0.01)
+    assert "not known-bad" in str(e.value)
+
+
+def test_a_pod_that_never_opens_its_port_is_still_a_plain_failure(monkeypatch):
+    """When the API ANSWERS and the pod simply never comes up, that is the
+    pod's fault and the existing behaviour must not change."""
+    cfg = types.SimpleNamespace(api_key="k", api_base="", rest_base="")
+    c = RunpodClient(cfg)
+    monkeypatch.setattr(c, "get_pod", lambda *_: {
+        "id": "p1", "status": "RUNNING", "uptime_s": 5, "ready": False, "ports": []})
+    monkeypatch.setattr(time, "sleep", lambda *_: None)
+    with pytest.raises(RunpodError) as e:
+        c.wait_ready("p1", 8000, ready_timeout=0.05, poll_every=0.01)
+    assert not isinstance(e.value, RunpodTransient)
+
+
+# -------------------------------------------------------------- the teardown #
+def test_terminate_retries_while_the_api_is_down(monkeypatch):
+    """The teardown runs through the API that just failed, so the attempt that
+    matters most is the one most likely to fail."""
+    cfg = types.SimpleNamespace(api_key="k", api_base="", rest_base="")
+    c = RunpodClient(cfg)
+    tries = []
+
+    def _gql(q, v=None, timeout=30.0):
+        tries.append(1)
+        if len(tries) < 3:
+            raise RunpodTransient("RunPod API HTTP 503")
+        return {}
+
+    monkeypatch.setattr(c, "_gql", _gql)
+    monkeypatch.setattr(time, "sleep", lambda *_: None)
+    c.terminate_pod("p1")
+    assert len(tries) == 3
+
+
+def test_terminate_eventually_raises_rather_than_pretending(monkeypatch):
+    """A caller that swallows this is choosing to leak a pod; it must at least
+    be given the chance to say so."""
+    cfg = types.SimpleNamespace(api_key="k", api_base="", rest_base="")
+    c = RunpodClient(cfg)
+    monkeypatch.setattr(c, "_gql",
+                        lambda *a, **k: (_ for _ in ()).throw(RunpodTransient("503")))
+    monkeypatch.setattr(time, "sleep", lambda *_: None)
+    with pytest.raises(RunpodTransient):
+        c.terminate_pod("p1", attempts=2)
+
+
+def test_a_final_error_is_not_retried(monkeypatch):
+    cfg = types.SimpleNamespace(api_key="k", api_base="", rest_base="")
+    c = RunpodClient(cfg)
+    tries = []
+
+    def _gql(*a, **k):
+        tries.append(1)
+        raise RunpodError("RunPod API HTTP 404: no such pod")
+
+    monkeypatch.setattr(c, "_gql", _gql)
+    with pytest.raises(RunpodError):
+        c.terminate_pod("p1", attempts=4)
+    assert len(tries) == 1
+
+
+# ---------------------------------------------------------------- the age #
+RENTED = "Rented by User: Thu Oct 15 2026 19:36:58 GMT+0000 (Coordinated Universal Time)"
+
+
+def test_age_comes_from_the_rental_not_the_current_run():
+    """The live failure in one assertion: 11 minutes old, uptime 234s."""
+    import datetime
+    rented = (datetime.datetime.now(datetime.timezone.utc)
+              - datetime.timedelta(minutes=11))
+    stamp = rented.strftime("Rented by User: %a %b %d %Y %H:%M:%S GMT+0000 (x)")
+    age = runpod_worker._pod_age_seconds({"last_change": stamp, "uptime_s": 234})
+    assert age is not None
+    assert age > 600, "a restart-looping pod must not look 234 seconds old"
+
+
+def test_a_restart_looping_pod_crosses_the_reap_grace():
+    """The whole point: the pod that most needs reaping was the one pod the
+    reaper could never see."""
+    import datetime
+    rented = (datetime.datetime.now(datetime.timezone.utc)
+              - datetime.timedelta(seconds=runpod_worker.REAP_GRACE_SECONDS + 300))
+    stamp = rented.strftime("Rented by User: %a %b %d %Y %H:%M:%S GMT+0000 (x)")
+    age = runpod_worker._pod_age_seconds({"last_change": stamp, "uptime_s": 60})
+    assert age >= runpod_worker.REAP_GRACE_SECONDS
+
+
+def test_uptime_still_works_when_there_is_no_rental_stamp():
+    assert runpod_worker._pod_age_seconds({"uptime_s": 900}) == 900
+
+
+def test_an_unparsable_stamp_falls_back_rather_than_guessing():
+    age = runpod_worker._pod_age_seconds({"last_change": "Rented by User: sometime",
+                                          "uptime_s": 120})
+    assert age == 120
+
+
+def test_no_age_at_all_is_still_none():
+    """The reaper fails CLOSED on None, so this must not become 0."""
+    assert runpod_worker._pod_age_seconds({"status": "RUNNING"}) is None
+
+
+def test_a_status_change_that_is_not_a_rental_is_not_an_age():
+    """'Stopped by user' dates the stop, not the pod."""
+    assert runpod_worker._rental_age_seconds(
+        {"last_change": "Stopped by User: Thu Oct 15 2026 19:36:58 GMT+0000 (x)"}) is None
+
+
+def test_a_future_rental_stamp_is_refused():
+    """A host clock behind RunPod's would otherwise make every pod ancient."""
+    import datetime
+    later = (datetime.datetime.now(datetime.timezone.utc)
+             + datetime.timedelta(hours=2))
+    stamp = later.strftime("Rented by User: %a %b %d %Y %H:%M:%S GMT+0000 (x)")
+    assert runpod_worker._rental_age_seconds({"last_change": stamp}) is None
+
+
+def test_the_real_timestamp_runpod_sends_parses():
+    """Copied verbatim from the account listing, sentence and all."""
+    assert runpod_worker._rental_age_seconds({"last_change": RENTED}) is not None