Never skip the last upstream; judge warm calls on what could be cached, per bot

Review fixes for the cache learning:
- Only skip an upstream when the model's endpoint list is known and another upstream remains, so the
  ignore list can never cover every upstream and fail all calls with 'no endpoints'.
- Judge a warm call against min(this prompt, previous prompt): a big new tool result can only be
  cached from the call after it, so it no longer counts as a miss.
- Miss streaks are per bot, so one bot's hit does not hide another's misses and ordinary misses from
  several bots do not add up to a skip.
- A pin to a skipped upstream (re-pinned by a call in flight) is dropped from the request.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: regnull

backend/openbot/runtime/upstream.py

@@ -112,7 +112,7 @@ def __init__(self, fetch: Callable[[str], Awaitable[dict[str, str]]] = fetch_end
         self._slugs: dict[str, tuple[float, dict[str, str]]] = {}
         # One lock per model: a slow fetch for one model must not hold up the others' runs.
         self._locks: dict[str, asyncio.Lock] = {}
-        self._misses: dict[tuple[str, str], int] = {}  # (model, slug) -> consecutive warm misses
+        self._misses: dict[tuple[str, str, str], int] = {}  # (bot, model, slug) -> consecutive warm misses
         self._ignored: dict[str, dict[str, float]] = {}  # model -> {slug: skipped until (monotonic)}
 
     def pinned(self, bot_id: str, model: str) -> str | None:
@@ -134,6 +134,14 @@ async def slug_for(self, model: str, name: str) -> str:
                 self._slugs[model] = (time.monotonic(), slugs)
         return slugs.get(name.lower()) or guess_slug(name)
 
+    def _others_remain(self, model: str, slug: str) -> bool:
+        """Whether skipping `slug` still leaves `model` an upstream to route to. Unknown when the
+        endpoint list could not be fetched; then nothing is skipped, since an ignore list covering
+        every upstream would fail every call to the model with "no endpoints"."""
+        cached = self._slugs.get(model)
+        known = set(cached[1].values()) if cached else set()
+        return bool(known - set(self.ignored(model)) - {slug})
+
     def ignored(self, model: str) -> list[str]:
         """Upstreams currently skipped for `model`, expired skips dropped."""
         entries = self._ignored.get(model, {})
@@ -142,17 +150,24 @@ def ignored(self, model: str) -> list[str]:
             del entries[slug]
         return sorted(entries)
 
-    def observe_cache(self, model: str, slug: str, prompt_tokens: int, cached_tokens: int) -> None:
-        """Judge one warm call: a hit clears the upstream's miss streak; MISS_LIMIT misses in a row
-        skip it for this model and unpin every bot pinned to it."""
-        key = (model, slug)
-        if prompt_tokens <= 0 or cached_tokens >= CACHE_HIT_SHARE * prompt_tokens:
+    def observe_cache(self, bot_id: str, model: str, slug: str, cacheable_tokens: int, cached_tokens: int) -> None:
+        """Judge one warm call by one bot. `cacheable_tokens` is what the upstream could have had
+        cached: the part of this prompt the previous call already sent, not the whole prompt (a 30k
+        tool result can only be cached from the call after it). A hit clears the bot's miss streak;
+        MISS_LIMIT misses in a row skip the upstream for this model and unpin every bot pinned to it.
+        Streaks are per bot, so one bot's hit does not hide another's misses, nor do ordinary misses
+        from several bots add up to a skip."""
+        key = (bot_id, model, slug)
+        if cacheable_tokens <= 0 or cached_tokens >= CACHE_HIT_SHARE * cacheable_tokens:
             self._misses.pop(key, None)
             return
         self._misses[key] = self._misses.get(key, 0) + 1
         if self._misses[key] < MISS_LIMIT:
             return
         del self._misses[key]
+        if not self._others_remain(model, slug):
+            log.info("OpenRouter upstream %s barely caches %s, but it is the last one left: keeping it", slug, model)
+            return
         entries = self._ignored.setdefault(model, {})
         entries[slug] = time.monotonic() + IGNORE_TTL
         while len(entries) > MAX_IGNORED:
@@ -191,6 +206,7 @@ def __init__(self, pins: UpstreamPins, bot_id: str, model: str, settings) -> Non
         self.pins, self.bot_id, self.model, self.settings = pins, bot_id, model, settings
         # One middleware per run: the upstream of this run's previous call, to tell warm calls apart.
         self._last_slug: str | None = None
+        self._last_prompt = 0  # input tokens of that call: the most a warm call can have cached
 
     def _explicit(self) -> bool:
         return bool(getattr(self.settings, "openrouter_provider_order", None))
@@ -200,6 +216,9 @@ def _pinned(self, request: ModelRequest) -> ModelRequest:
             return request
         slug = self.pins.pinned(self.bot_id, self.model)
         ignored = self.pins.ignored(self.model)
+        if slug in ignored:
+            # A call still in flight when its upstream was skipped re-pinned the bot to it.
+            slug = None
         provider: dict[str, Any] = {}
         if slug is not None:
             provider.update(order=[slug], allow_fallbacks=True)
@@ -228,8 +247,9 @@ async def awrap_model_call(
         if name and not self._explicit():
             slug = await self.pins.record(self.bot_id, self.model, name)
             usage = getattr(reply, "usage_metadata", None) or {}
+            prompt = int(usage.get("input_tokens") or 0)
             if slug == self._last_slug and usage:
                 cached = int((usage.get("input_token_details") or {}).get("cache_read") or 0)
-                self.pins.observe_cache(self.model, slug, int(usage.get("input_tokens") or 0), cached)
-            self._last_slug = slug
+                self.pins.observe_cache(self.bot_id, self.model, slug, min(prompt, self._last_prompt), cached)
+            self._last_slug, self._last_prompt = slug, prompt
         return response

backend/tests/test_upstream.py

@@ -128,28 +128,33 @@ async def test_an_upstream_that_keeps_missing_warm_calls_is_skipped_and_unpinned
     pins = UpstreamPins(_fetch)
     await pins.record("bot1", "glm", "Z.AI")
     await pins.record("bot2", "glm", "Z.AI")
-    pins.observe_cache("glm", "z-ai", 20_000, 0)
-    pins.observe_cache("glm", "z-ai", 20_000, 0)
-    pins.observe_cache("glm", "z-ai", 20_000, 15_000)  # a hit resets the streak
-    pins.observe_cache("glm", "z-ai", 20_000, 700)
-    pins.observe_cache("glm", "z-ai", 20_000, 0)
+    pins.observe_cache("bot1", "glm", "z-ai", 20_000, 0)
+    pins.observe_cache("bot1", "glm", "z-ai", 20_000, 0)
+    pins.observe_cache("bot1", "glm", "z-ai", 20_000, 15_000)  # a hit resets the streak
+    pins.observe_cache("bot1", "glm", "z-ai", 20_000, 700)
+    pins.observe_cache("bot1", "glm", "z-ai", 20_000, 0)
     assert pins.ignored("glm") == [] and pins.pinned("bot1", "glm") == "z-ai"
-    pins.observe_cache("glm", "z-ai", 20_000, 0)
+    pins.observe_cache("bot1", "glm", "z-ai", 20_000, 0)
     assert pins.ignored("glm") == ["z-ai"]
     assert pins.pinned("bot1", "glm") is None and pins.pinned("bot2", "glm") is None
     assert pins.ignored("other-model") == []  # skips are per model
     clock[0] += 6 * 3600 + 1
     assert pins.ignored("glm") == []  # and lapse, so the upstream gets another chance
 
 
-def test_the_skip_list_is_capped_so_routing_never_runs_out(monkeypatch):
+async def test_the_skip_list_is_capped_so_routing_never_runs_out(monkeypatch):
     clock = [1000.0]
     monkeypatch.setattr("openbot.runtime.upstream.time.monotonic", lambda: clock[0])
-    pins = UpstreamPins(_fetch)
+
+    async def many(m):
+        return {f"host {i}": f"host{i}" for i in range(20)}
+
+    pins = UpstreamPins(many)
+    await pins.slug_for("glm", "host 0")  # loads the endpoint list
     for i in range(12):
         clock[0] += 1
         for _ in range(3):
-            pins.observe_cache("glm", f"host{i}", 20_000, 0)
+            pins.observe_cache("bot1", "glm", f"host{i}", 20_000, 0)
     skipped = pins.ignored("glm")
     assert len(skipped) == 10 and "host0" not in skipped and "host11" in skipped
 
@@ -188,3 +193,75 @@ async def handler(request):
     for _ in range(8):
         await mw.awrap_model_call(_request(model()), handler)
     assert pins.ignored("z-ai/glm-5.3-flash") == []
+
+
+async def test_the_last_upstream_left_is_never_skipped():
+    """An ignore list naming every upstream would fail every call to the model with "no endpoints"."""
+
+    async def two(m):
+        return {"z.ai": "z-ai", "deepinfra": "deepinfra"}
+
+    pins = UpstreamPins(two)
+    await pins.slug_for("glm", "Z.AI")
+    for slug in ("z-ai", "deepinfra"):
+        for _ in range(3):
+            pins.observe_cache("bot1", "glm", slug, 20_000, 0)
+    assert pins.ignored("glm") == ["z-ai"]  # deepinfra is all that is left, so it stays
+
+    async def offline(m):
+        raise OSError("down")
+
+    blind = UpstreamPins(offline)
+    await blind.slug_for("glm", "Z.AI")
+    for _ in range(3):
+        blind.observe_cache("bot1", "glm", "z-ai", 20_000, 0)
+    assert blind.ignored("glm") == []  # without the endpoint list it cannot tell, so it skips nothing
+
+
+async def test_warm_calls_are_judged_on_what_could_have_been_cached(settings):
+    """A big tool result makes the next prompt much longer; only the previous call's part of it can be
+    cached, so 8k cached of a 38k prompt that followed an 8k one is a full hit, not a miss."""
+    settings.openrouter_provider_order = []
+    pins = UpstreamPins(_fetch)
+    mw = StickyUpstreamMiddleware(pins, "bot1", "z-ai/glm-5.3-flash", settings)
+    # Each warm call has under 25% of its whole prompt cached, but all of the previous call's.
+    calls = iter([(8_000, 0), (38_000, 8_000), (160_000, 38_000), (700_000, 160_000)])
+
+    async def handler(request):
+        prompt, cached = next(calls)
+        return AIMessage("ok", response_metadata={UPSTREAM_KEY: ["Z.AI"]}, usage_metadata=_usage(prompt, cached))
+
+    for _ in range(4):
+        await mw.awrap_model_call(_request(model()), handler)
+    assert pins.ignored("z-ai/glm-5.3-flash") == []
+
+
+async def test_miss_streaks_are_per_bot():
+    pins = UpstreamPins(_fetch)
+    await pins.slug_for("glm", "Z.AI")
+    for bot in ("bot1", "bot2", "bot3"):  # one ordinary miss each does not add up to a skip
+        pins.observe_cache(bot, "glm", "z-ai", 20_000, 0)
+    pins.observe_cache("bot1", "glm", "z-ai", 20_000, 0)
+    pins.observe_cache("bot2", "glm", "z-ai", 20_000, 15_000)  # bot2's hit does not clear bot1's streak
+    pins.observe_cache("bot1", "glm", "z-ai", 20_000, 0)
+    assert pins.ignored("glm") == ["z-ai"]
+
+
+async def test_a_pin_to_a_skipped_upstream_is_not_sent(settings):
+    """A call in flight when its upstream was skipped re-pins the bot to it; the request must not then
+    name the same upstream in both order and ignore."""
+    settings.openrouter_provider_order = []
+    pins = UpstreamPins(_fetch)
+    await pins.slug_for("z-ai/glm-5.3-flash", "Z.AI")
+    for _ in range(3):
+        pins.observe_cache("bot2", "z-ai/glm-5.3-flash", "z-ai", 20_000, 0)
+    await pins.record("bot1", "z-ai/glm-5.3-flash", "Z.AI")
+    mw = StickyUpstreamMiddleware(pins, "bot1", "z-ai/glm-5.3-flash", settings)
+    sent = []
+
+    async def handler(request):
+        sent.append(request.model_settings["extra_body"]["provider"])
+        return AIMessage("ok")
+
+    await mw.awrap_model_call(_request(model()), handler)
+    assert sent == [{"ignore": ["z-ai"]}]