runpod: the bearer was read before the pod was acquired, so adoption 401'd (v0.3.58)

Reported from the Digesta side as "the pinned pod image boots, but coderai calls
its vLLM endpoint with a token the pod rejects — every page 401, every page
empty". The 401 was real and the pages were really lost. It was not the image,
and not a mismatch between what coderai holds and what the pod wants — the pod
answers 200 to its recorded token, before and after.

It was the ORDER of two reads. `RunpodBackend.load_model` snapshotted
`pool.api_key` into `self._headers`, and `_acquire_endpoint()` — which is what
triggers provisioning, and therefore what runs `_adopt_shared_pod` and REPLACES
`pool.api_key` — runs after it. A pod for this pool was already running (ours,
from before a restart), so the pool adopted its token rather than renting a
second A40:

  22:27:15 [runpod] adopting pod sncrmkqdnd7c57's bearer token
  22:27:15 [runpod] reusing the pod already running … instead of renting another
  22:27:15 Inference error: Error code: 401 - Invalid API key.

Every request that crossed that moment carried the token from before it. All 64
pages in flight failed; the same lease answered 200 minutes later, which is why
it looked like a property of the run rather than of one instant in it.

`_auth()` now reads the pool at send time, and the three sites that used the
snapshot go through it (chat completions, streaming, openai_endpoint).
Serverless keeps the snapshot: it has no pool and no adoption event. The
placement candidate also reads the key AFTER acquiring rather than before,
since acquiring is the thing that can change it.

Behind that, the OCR gateway now retries a 401/403 once against a freshly
chosen upstream. A pod only answers to the token it was launched with, so a
stale credential is not recoverable downstream — and throwing a page away for a
credential that was already correct by the time the error arrived is the
expensive half of this bug. A second refusal is returned as itself, not retried
forever.

Also fixes a regression in 0.3.56: `register_pod` re-stamped `created` on every
write, but an adoption re-registers an existing pod — so a pod adopted hourly
looked newly born hourly, and the reaper's ledger attribution (which 0.3.56
computes from `created`) would have under-counted its hours. An existing
`created` is now preserved.

9 tests in tests/test_pod_token_adoption_race.py, including the ordering pinned
against the source so no send site can quietly go back to the snapshot.
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.57"
+__version__ = "0.3.58"
 __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

@@ -4387,18 +4387,28 @@ def register_pod(pod_id: str, pool_key: str = "", url: str = "",
     """
     if not pod_id:
         return
-    _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.
-                      "hourly_usd": float(hourly_usd or 0.0)})
-        or True)
+
+    def _write(d):
+        # An ADOPTION re-registers a pod that already exists, so "created" must
+        # survive the new owner: taking time.time() again would reset the age
+        # the reaper's ledger attribution is computed from, and a pod adopted
+        # every hour would look newly born every hour.
+        was = d.get(str(pod_id))
+        born = float((was or {}).get("created") or 0.0) if isinstance(was, dict) else 0.0
+        d[str(pod_id)] = {"pool": str(pool_key), "pid": os.getpid(),
+                          "proc": _PROC_ID, "at": time.time(),
+                          # Distinct from "at", which is a heartbeat: the
+                          # hours a reaped pod billed can only be computed from
+                          # when it was CREATED, and that must survive a new
+                          # owner.
+                          "created": born or 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.
+                          "hourly_usd": float(hourly_usd or 0.0)}
+        return True
+
+    _registry_update(_write)
 
 
 def touch_registered_pods() -> int:

codai/backends/runpod.py

@@ -127,6 +127,10 @@ class RunpodBackend(ModelBackend):
             # configured) and every request we send must carry it.
             _key = getattr(self._pool, "api_key", "")
             self._headers = {"Authorization": f"Bearer {_key}"} if _key else {}
+            # NOTE: that snapshot is only a starting value. A pool can CHANGE its
+            # token after this point — it adopts the token of a pod already
+            # running rather than renting a second one — so every send reads the
+            # pool again through _auth(). See _auth().
             # Eagerly warm a pod only when min_pods >= 1; otherwise provision lazily
             # on the first request (min_pods=0 = scale-to-zero when idle).
             if self._mcfg.min_pods >= 1:
@@ -178,6 +182,27 @@ class RunpodBackend(ModelBackend):
         self._last_pod = pod
         return url.rstrip("/") + "/v1", (lambda: self._pool.release(pod))
 
+    def _auth(self) -> dict:
+        """The bearer header for THIS send, read from the pool now.
+
+        Not the snapshot taken in load_model. A pool may replace its token
+        between loading the model and sending a request: when a pod for this
+        pool is already running — a sibling engine's, or ours from before a
+        restart — the pool adopts that pod's token instead of renting a second
+        card, and `pool.api_key` changes under us.
+
+        A request that read the header before that happened carried the old
+        token and the pod answered 401. Observed live: 64 OCR pages in flight
+        when a pool adopted a running pod all failed with "Invalid API key",
+        every one of them lost, while the same lease worked seconds later. The
+        snapshot stays for serverless, which has no pool and no such event.
+        """
+        pool = getattr(self, "_pool", None)
+        if pool is None:
+            return dict(self._headers or {})
+        key = getattr(pool, "api_key", "") or ""
+        return {"Authorization": f"Bearer {key}"} if key else {}
+
     def openai_endpoint(self):
         """``(base_url ending /v1, api_key)`` for a client that will speak to this
         model over HTTP ITSELF, instead of going through us.
@@ -193,7 +218,7 @@ class RunpodBackend(ModelBackend):
         answering rather than treating it as permanent."""
         base, release = self._acquire_endpoint()
         try:
-            auth = (self._headers or {}).get("Authorization", "")
+            auth = (self._auth() or {}).get("Authorization", "")
             key = auth[7:].strip() if auth[:7].lower() == "bearer " else ""
             return base, key
         finally:
@@ -283,7 +308,7 @@ class RunpodBackend(ModelBackend):
                     if tools:
                         payload["tools"] = tools
                     r = pod_http.post(base + "/chat/completions", json=payload,
-                                      headers=self._headers, timeout=3600)
+                                      headers=self._auth(), timeout=3600)
                 except requests.exceptions.ConnectionError:
                     # Nothing answered at all: the pod is gone. Once, on a
                     # fresh pod — a second refusal is a real outage.
@@ -368,7 +393,7 @@ class RunpodBackend(ModelBackend):
         loop = asyncio.get_event_loop()
         queue: asyncio.Queue = asyncio.Queue()
         _SENTINEL = object()
-        headers = self._headers
+        headers = self._auth()
 
         def _handle_line(line: str) -> bool:
             if not line or not line.startswith("data:"):

codai/models/placement.py

@@ -296,7 +296,11 @@ def _pod_candidate(model_id: str) -> Optional[Candidate]:
 
     def _acquire():
         base, release = backend._acquire_endpoint()
-        auth = (backend._headers or {}).get("Authorization", "")
+        # AFTER the acquire, and from the live accessor: acquiring is what can
+        # change the pool's token (it adopts a running pod's instead of renting
+        # another), and the snapshot taken at load_model time would be the old
+        # one — which the pod answers 401 to.
+        auth = (backend._auth() or {}).get("Authorization", "")
         key = auth[7:].strip() if auth[:7].lower() == "bearer " else ""
         # _unreachable() drops the pod the backend just handed out, which is
         # what stops the next page choosing the same dead one.

codai/ocr/gateway.py

@@ -417,6 +417,20 @@ async def chat_completions(token: str, request: Request):
                         await client.aclose()
                     except Exception:
                         pass
+            if reply.status_code in (401, 403) and attempt == 1:
+                # The upstream refused our credential. A pod only answers to the
+                # token it was launched with, and a pool CHANGES its token when
+                # it adopts a pod that is already running instead of renting a
+                # second card — so a page in flight across that moment carries
+                # the token from before it. Re-choosing reads the pool again,
+                # which is enough; this is the last line of defence behind
+                # RunpodBackend._auth(), because the alternative is a page
+                # thrown away for a credential that was already correct by the
+                # time the error arrived. 64 pages were lost that way.
+                print(f"[ocr] upstream {chosen.base} refused our token "
+                      f"(HTTP {reply.status_code}) — re-resolving and sending "
+                      f"this page once more", flush=True)
+                continue
             # The reply goes back as it came: a re-serialised response is how
             # logprobs got lost in the first place.
             media = reply.headers.get("content-type", "application/json")

tests/test_pod_token_adoption_race.py

+"""A pool that adopts a running pod's token must not lose the pages in flight.
+
+Live incident, 2026-10-13 22:27 on the Digesta orchestrator. A surya run started,
+and a pod for the same pool was already running (ours, from before a restart), so
+the pool adopted it instead of renting a second A40:
+
+  22:27:15 [runpod] adopting pod sncrmkqdnd7c57's bearer token
+  22:27:15 [runpod] reusing the pod already running … instead of renting another
+  22:27:15 Inference error: Error code: 401 - Invalid API key.
+
+Every one of the 64 pages in flight failed with 401 and was lost; the same lease
+answered 200 seven minutes later, and the pod answered 200 to its recorded token
+throughout. The cause was the ORDER of two reads: `RunpodBackend.load_model`
+snapshotted `pool.api_key` into `self._headers`, and `_acquire_endpoint()` — the
+call that triggers the adoption and therefore CHANGES `pool.api_key` — runs
+after it. A request that crossed that moment carried the token from before it.
+
+A pod only ever answers to the token it was launched with, so this is not
+recoverable downstream: the page is thrown away for a credential that was
+already correct by the time the error arrived.
+"""
+
+import pathlib
+import sys
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+sys.path.insert(0, str(ROOT / "tests"))
+
+from codai.backends.runpod import RunpodBackend                    # noqa: E402
+
+
+class _Pool:
+    def __init__(self, key):
+        self.api_key = key
+
+
+def _backend(pool, snapshot):
+    b = RunpodBackend.__new__(RunpodBackend)
+    b._pool = pool
+    b._headers = {"Authorization": f"Bearer {snapshot}"} if snapshot else {}
+    return b
+
+
+# --- the live accessor ------------------------------------------------------
+
+def test_the_header_is_read_from_the_pool_not_the_snapshot():
+    pool = _Pool("adopted-token")
+    b = _backend(pool, "stale-token")
+    assert b._auth() == {"Authorization": "Bearer adopted-token"}
+
+
+def test_a_token_adopted_after_load_is_used():
+    # Exactly the incident: load_model snapshots, then the acquire adopts.
+    pool = _Pool("own-generated")
+    b = _backend(pool, "own-generated")
+    pool.api_key = "the-running-pods-token"       # _adopt_shared_pod does this
+    assert b._auth()["Authorization"] == "Bearer the-running-pods-token"
+
+
+def test_a_pool_with_no_token_sends_no_authorization_header():
+    # Some servers reject an empty bearer outright, so the header must be absent
+    # rather than present and blank.
+    assert _backend(_Pool(""), "") == {} or _backend(_Pool(""), "")._auth() == {}
+
+
+def test_serverless_keeps_its_snapshot():
+    """Serverless has no pool and no adoption event; its header is the account's."""
+    b = RunpodBackend.__new__(RunpodBackend)
+    b._pool = None
+    b._headers = {"Authorization": "Bearer account-key"}
+    assert b._auth() == {"Authorization": "Bearer account-key"}
+
+
+def test_the_accessor_is_what_every_send_uses():
+    """Pinned against the source: three sites used the snapshot, and any one of
+    them left behind would reproduce the incident on that path."""
+    src = (ROOT / "codai/backends/runpod.py").read_text()
+    assert "headers=self._auth()" in src                 # chat completions
+    assert "headers = self._auth()" in src               # streaming
+    assert "(self._auth() or {}).get(\"Authorization\"" in src   # openai_endpoint
+    # …and the placement candidate reads it AFTER acquiring, since acquiring is
+    # what can change it.
+    pl = (ROOT / "codai/models/placement.py").read_text()
+    after = pl.split("base, release = backend._acquire_endpoint()")[1][:400]
+    assert "backend._auth()" in after
+
+
+# --- the gateway's last line of defence ------------------------------------
+
+def test_a_401_is_retried_once_against_a_freshly_chosen_upstream(monkeypatch):
+    import asyncio
+    import httpx
+    from codai.ocr import gateway
+    from test_ocr_fanout import (_CapturingClient, _Req, _cand, _no_node_load,
+                                 _fixed_candidates, _reset, SURYA_BODY)
+
+    _reset()
+    calls = []
+
+    class _RefusesOnce(_CapturingClient):
+        # The gateway reuses its pooled client on the second attempt and asks
+        # whether it is still open; the shared stub is only ever used once.
+        is_closed = False
+
+        async def post(self, url, content=None, headers=None):
+            calls.append(headers.get("Authorization"))
+
+            class _R:
+                status_code = 401 if len(calls) == 1 else 200
+                headers = {"content-type": "application/json"}
+                content = b'{"ok": true}'
+            return _R()
+
+    monkeypatch.setattr(httpx, "AsyncClient", _RefusesOnce)
+    _no_node_load(monkeypatch)
+    c = _cand("node:a", base="http://a/v1")
+    c.api_key = "k"
+    _fixed_candidates(monkeypatch, [c])
+    token = gateway.issue("m", "m")
+    reply = asyncio.run(gateway.chat_completions(token, _Req(SURYA_BODY)))
+    # The page is served, not lost.
+    assert reply.status_code == 200
+    assert len(calls) == 2
+
+
+def test_a_second_401_is_reported_rather_than_retried_forever(monkeypatch):
+    import asyncio
+    import httpx
+    from codai.ocr import gateway
+    from test_ocr_fanout import (_CapturingClient, _Req, _cand, _no_node_load,
+                                 _fixed_candidates, _reset, SURYA_BODY)
+
+    _reset()
+    calls = []
+
+    class _AlwaysRefuses(_CapturingClient):
+        is_closed = False
+
+        async def post(self, url, content=None, headers=None):
+            calls.append(1)
+
+            class _R:
+                status_code = 401
+                headers = {"content-type": "application/json"}
+                content = b'{"error": "nope"}'
+            return _R()
+
+    monkeypatch.setattr(httpx, "AsyncClient", _AlwaysRefuses)
+    _no_node_load(monkeypatch)
+    c = _cand("node:a", base="http://a/v1")
+    c.api_key = "k"
+    _fixed_candidates(monkeypatch, [c])
+    token = gateway.issue("m", "m")
+    reply = asyncio.run(gateway.chat_completions(token, _Req(SURYA_BODY)))
+    # A real credential problem must surface as itself, not as a retry loop.
+    assert reply.status_code == 401
+    assert len(calls) == 2
+
+
+# --- the registry must not forget a pod's age when it changes hands ---------
+
+def test_adoption_keeps_the_original_created_timestamp(monkeypatch, tmp_path):
+    import json
+    import time
+
+    from codai.api import runpod_worker as rw
+
+    path = tmp_path / "runpod-pods.json"
+    born = time.time() - 3600
+    path.write_text(json.dumps({"p1": {"pool": "m", "proc": "old-1",
+                                       "created": born, "at": born,
+                                       "url": "u", "key": "k"}}))
+    monkeypatch.setattr(rw, "_pod_registry_path", lambda: str(path))
+    rw.register_pod("p1", pool_key="m", url="u", api_key="k", hourly_usd=0.59)
+    info = json.loads(path.read_text())["p1"]
+    # Re-stamping it would make an hourly-adopted pod look newly born every
+    # hour, and the reaper's ledger attribution is computed from this.
+    assert abs(info["created"] - born) < 1
+    assert info["proc"] == rw._PROC_ID
+    assert info["at"] > born + 3000
+
+
+def test_a_pod_registered_for_the_first_time_gets_a_created_stamp(monkeypatch, tmp_path):
+    import json
+    import time
+
+    from codai.api import runpod_worker as rw
+
+    path = tmp_path / "runpod-pods.json"
+    path.write_text("{}")
+    monkeypatch.setattr(rw, "_pod_registry_path", lambda: str(path))
+    rw.register_pod("p2", pool_key="m", url="u", api_key="k")
+    info = json.loads(path.read_text())["p2"]
+    assert abs(info["created"] - time.time()) < 5