Skip OpenRouter upstreams that do not cache

README.md

@@ -646,7 +646,7 @@ name, if set, is the default the page shows and the value a reset returns to.
 | Context | `history_token_budget` (24000), `history_max_messages` (80) | Thread history included in a run's prompt. |
 | Memory | `memory_reflection_delay` (30) | Seconds to debounce background memory reflection after a run. |
 | Model routing | `bot_model` | OpenRouter model for bots on `auto`, `openrouter` bots with no model, and cloud bots whose provider has no key (default `openai/gpt-4o-mini`). |
-| Model routing | `openrouter_provider_order` | Preferred OpenRouter upstream slugs (e.g. `z-ai`); each upstream has its own prompt cache. Empty (the default): each bot is pinned to the upstream that served its last call, with fallbacks on, so its cache stays warm and the pin follows when that upstream fails. |
+| Model routing | `openrouter_provider_order` | Preferred OpenRouter upstream slugs (e.g. `z-ai`); each upstream has its own prompt cache. Empty (the default): each bot is pinned to the upstream that served its last call, with fallbacks on, so its cache stays warm and the pin follows when that upstream fails. An upstream that keeps answering warm calls with less than a quarter of the prompt cached (three in a row) is skipped for that model for 6 hours, so bots settle on one that caches. |
 | Model routing | `prompt_caching` (true), `direct_anthropic` (true) | Anthropic cache breakpoints on every call; send OpenRouter `anthropic/...` models straight to Anthropic when a key exists so caching covers tool results. |
 | Model retries | `model_retry_max_attempts` (3), `model_retry_base_delay` (2.0), `model_retry_backoff_cap` (60.0) | Retries a model call that failed with a transient upstream provider error (rate limit, 5xx, timeout, "model stopped before completing"); 0 or 1 attempts disables retries. Delays double from the base with jitter up to the cap. Auth/permission/invalid-request errors fail immediately. |
 | Model retries | `model_call_timeout` (120) | Seconds one model call may take, including gaps between streamed chunks, before it fails as a transient error and is retried. The provider SDKs' own silent retries are off, so a dead upstream costs at most `model_retry_max_attempts` × this. 0 means no limit. |

backend/openbot/runtime/upstream.py

@@ -7,6 +7,13 @@
 automatically: it reads which upstream answered each call and pins the bot's next call to it, with
 fallbacks on, so when that upstream fails OpenRouter answers elsewhere and the pin follows.
 
+Staying on one upstream only helps if it caches. OpenRouter's default routing favors the cheapest
+upstreams, and on 2026-10-18 bots pinned that way (OpenInference, Venice, PrimeIntellect) got ~2% of
+their prompt tokens cached. So the pins also learn: an upstream that keeps serving warm calls (same
+upstream as the run's previous call, so the prompt's start should be cached) with little cached is
+skipped for that model for a few hours, via OpenRouter's `provider.ignore`, and bots pinned to it
+move on until they land on one that caches.
+
 OpenRouter reports the upstream by display name ("Z.AI") but routes by slug ("z-ai"); the model's
 public endpoint list maps one to the other.
 """
@@ -31,6 +38,10 @@
 ENDPOINTS_URL = "https://openrouter.ai/api/v1/models/{model}/endpoints"
 ENDPOINTS_TTL = 6 * 3600  # upstreams come and go; a stale map only costs a fallback slug
 ENDPOINTS_RETRY = 600  # after a failed fetch, guess slugs for this long before asking again
+CACHE_HIT_SHARE = 0.25  # a warm call with less than this share of its prompt cached is a miss
+MISS_LIMIT = 3  # consecutive warm misses before an upstream is skipped (one summarization rewrite is fine)
+IGNORE_TTL = 6 * 3600  # how long a non-caching upstream is skipped before it gets another chance
+MAX_IGNORED = 10  # per model; the oldest skip lapses first, so routing never runs out of upstreams
 
 
 def upstream_of(message: BaseMessage) -> str | None:
@@ -92,14 +103,17 @@ async def fetch_endpoint_slugs(model: str) -> dict[str, str]:
 
 
 class UpstreamPins:
-    """Per-(bot, model) pinned upstream slug, plus the cached name-to-slug map per model."""
+    """Per-(bot, model) pinned upstream slug, the cached name-to-slug map per model, and per model the
+    upstreams skipped for not caching."""
 
     def __init__(self, fetch: Callable[[str], Awaitable[dict[str, str]]] = fetch_endpoint_slugs) -> None:
         self._fetch = fetch
         self._pins: dict[tuple[str, str], str] = {}
         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, 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:
         return self._pins.get((bot_id, model))
@@ -120,6 +134,51 @@ 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, {})
+        now = time.monotonic()
+        for slug in [s for s, until in entries.items() if until <= now]:
+            del entries[slug]
+        return sorted(entries)
+
+    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:
+            del entries[min(entries, key=entries.__getitem__)]
+        for pin_key in [k for k, v in self._pins.items() if k[1] == model and v == slug]:
+            del self._pins[pin_key]
+        log.info(
+            "OpenRouter upstream %s barely caches %s (%d warm misses in a row): skipping it for %dh",
+            slug, model, MISS_LIMIT, IGNORE_TTL // 3600,
+        )
+
     async def record(self, bot_id: str, model: str, name: str) -> str:
         slug = await self.slug_for(model, name)
         previous = self._pins.get((bot_id, model))
@@ -145,20 +204,34 @@ class StickyUpstreamMiddleware(AgentMiddleware):
     def __init__(self, pins: UpstreamPins, bot_id: str, model: str, settings) -> None:
         super().__init__()
         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))
 
     def _pinned(self, request: ModelRequest) -> ModelRequest:
-        slug = None if self._explicit() else self.pins.pinned(self.bot_id, self.model)
-        if slug is None:
+        if self._explicit():
+            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)
+        if ignored:
+            provider["ignore"] = ignored
+        if not provider:
             return request
         # bind(extra_body=...) replaces the model's own extra_body wholesale, so carry over what it
         # already sends (e.g. the web-search plugin).
         extra = {
             **(getattr(request.model, "extra_body", None) or {}),
             **(request.model_settings.get("extra_body") or {}),
-            "provider": {"order": [slug], "allow_fallbacks": True},
+            "provider": provider,
         }
         return request.override(model_settings={**request.model_settings, "extra_body": extra})
 
@@ -172,5 +245,11 @@ async def awrap_model_call(
         reply = _reply(response)
         name = upstream_of(reply) if reply is not None else None
         if name and not self._explicit():
-            await self.pins.record(self.bot_id, self.model, name)
+            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.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

@@ -115,3 +115,153 @@ async def handler(request):
     await mw.awrap_model_call(_request(model()), handler)
     assert sent == [None]  # the model's own extra_body (the explicit order) is left alone
     assert pins.pinned("bot1", "z-ai/glm-5.3-flash") == "z-ai"  # and replies do not move the pin
+
+
+def _usage(prompt: int, cached: int) -> dict:
+    return {"input_tokens": prompt, "output_tokens": 10, "total_tokens": prompt + 10,
+            "input_token_details": {"cache_read": cached}}
+
+
+async def test_an_upstream_that_keeps_missing_warm_calls_is_skipped_and_unpinned(monkeypatch):
+    clock = [1000.0]
+    monkeypatch.setattr("openbot.runtime.upstream.time.monotonic", lambda: clock[0])
+    pins = UpstreamPins(_fetch)
+    await pins.record("bot1", "glm", "Z.AI")
+    await pins.record("bot2", "glm", "Z.AI")
+    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("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
+
+
+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])
+
+    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("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
+
+
+async def test_the_middleware_moves_off_a_non_caching_upstream(settings):
+    settings.openrouter_provider_order = []
+    pins = UpstreamPins(_fetch)
+    mw = StickyUpstreamMiddleware(pins, "bot1", "z-ai/glm-5.3-flash", settings)
+    sent = []
+    # Z.AI answers four calls without caching (the first is cold and not judged), then DeepInfra caches.
+    served = iter([("Z.AI", 0)] * 4 + [("DeepInfra", 0), ("DeepInfra", 18_000)])
+
+    async def handler(request):
+        sent.append((request.model_settings.get("extra_body") or {}).get("provider"))
+        name, cached = next(served)
+        return AIMessage("ok", response_metadata={UPSTREAM_KEY: [name]}, usage_metadata=_usage(20_000, cached))
+
+    for _ in range(6):
+        await mw.awrap_model_call(_request(model()), handler)
+    assert sent[0] is None
+    assert sent[1] == sent[2] == sent[3] == {"order": ["z-ai"], "allow_fallbacks": True}
+    assert sent[4] == {"ignore": ["z-ai"]}  # three warm misses: unpinned and skipped
+    assert sent[5] == {"order": ["deepinfra"], "allow_fallbacks": True, "ignore": ["z-ai"]}
+    assert pins.pinned("bot1", "z-ai/glm-5.3-flash") == "deepinfra"
+
+
+async def test_calls_that_switched_upstream_are_cold_and_not_judged(settings):
+    settings.openrouter_provider_order = []
+    pins = UpstreamPins(_fetch)
+    mw = StickyUpstreamMiddleware(pins, "bot1", "z-ai/glm-5.3-flash", settings)
+    served = iter(["Z.AI", "DeepInfra"] * 4)
+
+    async def handler(request):
+        return AIMessage("ok", response_metadata={UPSTREAM_KEY: [next(served)]}, usage_metadata=_usage(20_000, 0))
+
+    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"]}]