feat: adopt host-owned reliability/perf (engine #15) + bump core

llm_router_host.py

@@ -376,16 +376,18 @@ async def _resolve_call_async(self, request: dict, call_override=None) -> dict:
         override, then the async hook, then the sync hook as a last resort."""
         key = (request.get("provider_id"), request.get("model_family"))
         if key in self._mock_responses:
-            return self._mock_responses[key]
-        if call_override is not None:
+            result = self._mock_responses[key]
+        elif call_override is not None:
             result = await call_override(request)
         elif self._async_call_hook is not None:
             result = await self._async_call_hook(request)
         else:
             result = self._call_hook(request)
         # Fold the outcome here (not in the hook) so the streaming/override path —
         # all of opencode's traffic, and every flow node — feeds route_latency /
-        # reliability / tool_capability too. Mocks return above, unmeasured.
+        # reliability / the call count too, the host-owned perf the algebra reads
+        # and the market view surfaces (#15). Mocks fold as well, so a mocked call
+        # is measured exactly like a live one.
         _fold_route_outcome(request, result)
         return result
 
@@ -714,16 +716,20 @@ def _fold_route_outcome(request: dict, result: dict) -> None:
         return
     peer_id = request.get("peer_id") or offer.get("peer_id")
     ok = bool(result.get("ok"))
+    # One route identity for reliability, latency and the call count: the peer for
+    # marketplace routes, or the provider itself for partner/gateway routes (no
+    # peer_id), so OpenRouter/OpenAI is comparable to a peer's. The engine no
+    # longer folds reliability for ANY route (#15), so the host folds it for all
+    # of them — not just marketplace — and the market perf view reads it back.
+    rkey = _route_reliability.route_key(pid, fam, peer_id or pid)
+    _route_reliability.observe(rkey, ok)
+    _route_latency.observe(rkey, result.get("latency_ms"), ok)
+    # Learned tool capability is a marketplace concern only (static/partner routes
+    # declare their capabilities in config), so keep it peer-scoped.
     if peer_id:
-        rkey = _route_reliability.route_key(pid, fam, peer_id)
-        _route_reliability.observe(rkey, ok)
         _route_tool_capability.observe(
             rkey, bool(request.get("tools")),
             bool((result.get("response") or {}).get("tool_calls")))
-    # Latency is keyed on the peer, or on the provider itself for partner/gateway
-    # routes (no peer_id), so OpenRouter/OpenAI latency is comparable to a peer's.
-    lkey = _route_reliability.route_key(pid, fam, peer_id or pid)
-    _route_latency.observe(lkey, result.get("latency_ms"), ok)
 
 
 def make_async_call_provider(

route_reliability.py

@@ -26,6 +26,11 @@
 
 _lock = threading.Lock()
 _rates: dict[str, float] = {}
+# Per-route observation count. The engine no longer folds an EMA (reliability is
+# host-owned, #15), so the host owns the "how many calls have we made on this
+# route" count too — surfaced in the market perf view and asserted by the
+# concurrency invariant.
+_counts: dict[str, int] = {}
 
 
 def route_key(provider_id: str, model_family: str, peer_id: str) -> str:
@@ -39,19 +44,31 @@ def observe(key: str, ok: bool) -> None:
     with _lock:
         cur = _rates.get(key)
         _rates[key] = s if cur is None else _ALPHA * s + (1.0 - _ALPHA) * cur
+        _counts[key] = _counts.get(key, 0) + 1
 
 
 def success_rate(key: str) -> float | None:
     """The route's folded success rate, or None if never observed."""
     return _rates.get(key)
 
 
+def count(key: str) -> int:
+    """How many outcomes have been folded for this route (0 if never observed)."""
+    return _counts.get(key, 0)
+
+
 def snapshot() -> dict[str, float]:
     with _lock:
         return dict(_rates)
 
 
+def snapshot_counts() -> dict[str, int]:
+    with _lock:
+        return dict(_counts)
+
+
 def reset() -> None:
     """Test hook."""
     with _lock:
         _rates.clear()
+        _counts.clear()

shim.py

@@ -307,22 +307,38 @@ def market_view():
         actually called that provider|family. Internal — the dashboard
         fetches this server-side; /x/* is hidden from consumers."""
         import sources as _sources
+        import route_reliability as _rr
+        import route_latency as _rl
         catalog = host.catalog() or {}
         models = catalog.get("models") or {}
         state = host.dump_state() or {}
-        ema = state.get("ema_metrics") or {}
+        ema = state.get("ema_metrics") or {}   # still carries seeded price + credits
         disabled = state.get("disabled_providers") or {}
         marketplace_pids = {
             pid for pid, p in (catalog.get("providers") or {}).items()
             if isinstance(p, dict) and p.get("discovery") == "marketplace"}
 
+        # Live perf is host-owned now (#15): the engine no longer folds an EMA, so
+        # build it from the host's per-route measurements (route_reliability /
+        # route_latency / the call count), aggregated across the peers/route ids
+        # that serve a given provider|family. None until the router has called it.
+        _rates = _rr.snapshot()
+        _counts = _rr.snapshot_counts()
+        _lats = _rl.snapshot()
+
         def _perf(provider, family):
-            m = ema.get(f"{provider}|{family}") or {}
-            if not m.get("n"):
+            prefix = f"{provider}|{family}|"
+            keys = [k for k in _counts if k.startswith(prefix)]
+            total = sum(_counts[k] for k in keys)
+            if not total:
                 return None
-            return {"success_rate": m.get("success_rate_ewma"),
-                    "latency_ms": m.get("ema_latency_ms"),
-                    "calls": m.get("n")}
+            sr = sum(_rates[k] * _counts[k] for k in keys if k in _rates)
+            sr_calls = sum(_counts[k] for k in keys if k in _rates)
+            lt = sum(_lats[k] * _counts[k] for k in keys if k in _lats)
+            lt_calls = sum(_counts[k] for k in keys if k in _lats)
+            return {"success_rate": (sr / sr_calls) if sr_calls else None,
+                    "latency_ms": round(lt / lt_calls) if lt_calls else None,
+                    "calls": total}
 
         def _antseed_row(r, family, book):
             via = (r.get("tradable_via") or [None])[0]

tests/test_async_concurrency.py

@@ -70,8 +70,13 @@ async def hook(req):
 
 @pytest.mark.asyncio
 async def test_shared_state_is_coherent_under_concurrency():
-    """All coroutines share one router state; EMA metric count must equal the
-    number of calls with no lost updates (single-thread invariant)."""
+    """All coroutines fold into one host-owned reliability state; the per-route
+    observation count must equal the number of calls with no lost updates
+    (single-thread invariant). Reliability is host-owned now (#15), so the count
+    lives in route_reliability, not the engine EMA."""
+    import route_reliability as rr
+    rr.reset()
+
     async def hook(req):
         await asyncio.sleep(0.01)
         return {"ok": True, "response": {"text": "ok"}}
@@ -80,6 +85,5 @@ async def hook(req):
     contract = {"profile": "default", "messages": [{"role": "user", "content": "hi"}]}
     await asyncio.gather(*(host.execute_async(dict(contract)) for _ in range(N)))
 
-    state = host.dump_state()
-    total_n = sum(m.get("n", 0) for m in (state.get("ema_metrics") or {}).values())
+    total_n = sum(rr.snapshot_counts().values())
     assert total_n == N, f"expected {N} recorded calls, got {total_n} (lost updates?)"

tests/test_host.py

@@ -153,21 +153,27 @@ def test_marketplace_discovery_merges_offers_into_pool(host):
 # Per-call ranking is now a raw `policy_ir` scorer (see test_policy_ir.py).
 
 
-def test_min_tok_s_filters_unbenched_candidates(host):
-    # min_tok_s=39 should keep only candidates with observed tok_s >= 39 in metrics.
-    # Per metrics.example: hermes@comput3=42.1, llama@io_net=40.0 → both pass.
-    # deepseek@comput3=38.0 fails. Others have no metrics → fail.
+def test_min_tok_s_filters_on_stamped_throughput(host):
+    # Engine #15: throughput is host-measured and stamped per candidate (like
+    # price); the engine reads cand.tok_s and no longer seeds it from metrics.
+    # A candidate stamped above the floor passes; one below it, or one with no
+    # measured throughput at all, is rejected with reason min_tok_s.
+    base = {"model_family": "fam", "served_model_id": "fam", "capabilities": {},
+            "tier": "fallback", "api_kind": "openai_compatible", "discovery": "static"}
+    fast = {**base, "provider_id": "p_fast", "tok_s": 50.0}
+    slow = {**base, "provider_id": "p_slow", "tok_s": 10.0}
+    unstamped = {**base, "provider_id": "p_unstamped"}  # no tok_s -> default 0
     ranked, rejected = host.rank({
         "prompt": "x",
         "profile": "default",
         "requirements": {"min_tok_s": 39},
+        "extra_candidates": [fast, slow, unstamped],
     })
-    surviving = {(r["candidate"]["provider_id"], r["candidate"]["model_family"])
-                 for r in ranked}
-    assert ("comput3", "hermes-3-405b") in surviving
-    assert ("io_net", "llama-3.3-70b") in surviving
-    assert ("comput3", "deepseek-v3") not in surviving
-    # at least one rejection with reason min_tok_s
+    surviving = {r["candidate"]["provider_id"] for r in ranked}
+    assert "p_fast" in surviving
+    assert "p_slow" not in surviving
+    assert "p_unstamped" not in surviving
+    # static catalog candidates carry no measured tok_s either, so they too fail
     reasons = {r["reason"] for r in rejected}
     assert "min_tok_s" in reasons
 
@@ -266,12 +272,13 @@ async def override(r):                     # a streamed result: ok + latency_ms
     assert rl.latency_ms(k) == 12000           # latency folded from the streamed call
     assert rr.success_rate(k) == 1.0
 
-    # a mock short-circuits BEFORE the fold, so test measurements stay pure
+    # A mock folds too now (#15: the host owns perf, so a mocked call is measured
+    # exactly like a live one — the engine no longer folds a separate EMA).
     rr.reset(); rl.reset()
     host.set_mock_response("antseed", "glm-5.2",
                            {"ok": True, "latency_ms": 5, "response": {}})
     asyncio.run(host._resolve_call_async(req))
-    assert rl.latency_ms(k) is None
+    assert rl.latency_ms(k) == 5
 
 
 def test_fold_uses_top_level_route_identity(host):

tests/test_route_reliability.py

@@ -29,24 +29,6 @@ def test_ema_seeds_then_decays():
     assert rr.success_rate(k) == pytest.approx(0.36)
 
 
-class _Resp:
-    def __init__(self, content):
-        self.status_code = 200
-        self._c = content
-
-    def json(self):
-        return {"choices": [{"message": {"content": self._c}, "finish_reason": "stop"}],
-                "usage": {}}
-
-
-class _Client:
-    def __init__(self, content):
-        self._content = content
-
-    async def post(self, url, json=None, headers=None, timeout=None):
-        return _Resp(self._content)
-
-
 def _req(peer_id, family="m"):
     return {
         "api_kind": "openai_compatible", "base_url": "http://s/v1",
@@ -56,21 +38,16 @@ def _req(peer_id, family="m"):
     }
 
 
-@pytest.mark.asyncio
-async def test_call_folds_route_reliability_on_success():
+# Folding moved out of the call backend into _fold_route_outcome, which runs once
+# per resolved call (direct + streaming/flow paths) so all traffic feeds the same
+# host-owned EMAs the algebra reads (offer.success_rate). Drive it directly.
+def test_fold_route_outcome_on_success():
     rr.reset()
-    H._PEER_GATES.clear()
-    call = H.make_async_call_provider(client=_Client("hello"))
-    r = await call(_req("peerGood"))
-    assert r["ok"] is True
+    H._fold_route_outcome(_req("peerGood"), {"ok": True, "latency_ms": 10})
     assert rr.success_rate(rr.route_key("antseed", "m", "peerGood")) == 1.0
 
 
-@pytest.mark.asyncio
-async def test_call_folds_route_reliability_on_empty_content():
+def test_fold_route_outcome_on_failure():
     rr.reset()
-    H._PEER_GATES.clear()
-    call = H.make_async_call_provider(client=_Client(""))   # empty -> bad_response
-    r = await call(_req("peerBad"))
-    assert r["ok"] is False and r["error_kind"] == "bad_response"
+    H._fold_route_outcome(_req("peerBad"), {"ok": False, "error_kind": "bad_response"})
     assert rr.success_rate(rr.route_key("antseed", "m", "peerBad")) == 0.0