From 887c3970df770c1b06c8676b9a7a285ac8a1b641 Mon Sep 17 00:00:00 2001 From: jmlago Date: Mon, 22 Jun 2026 13:34:51 +0100 Subject: [PATCH] feat: adopt host-owned reliability/perf (engine #15) + bump core submodule MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bump core to unhardcoded-engine main (4409f6f), which retired the engine's reliability/latency/throughput fold — these are host-owned now, read by the algebra only off the candidate. Adapt the host to own them end-to-end: - _fold_route_outcome: fold reliability + latency + a call count for EVERY route (peer for marketplace, provider for partner/gateway), not just marketplace, and fold mocked calls too so a mocked call is measured like a live one. - route_reliability: add a per-route observation counter (count/snapshot_counts) — the engine no longer tracks observation counts. - /x/market perf: rebuild success_rate/latency/calls from the host folds instead of the engine EMA (which no longer carries them; it keeps price + credits). - tests: rewrite the cases that asserted the old engine fold to the host-owned model (min_tok_s now filters on candidate-stamped tok_s; concurrency counts host observations; route-reliability folds via _fold_route_outcome). Note: throughput (tok_s) is not yet host-measured/stamped on marketplace offers, so min_tok_s policies match only candidates with an explicit tok_s. A route_throughput fold mirroring route_latency is the natural follow-up. Co-Authored-By: Claude Opus 4.8 (1M context) --- core | 2 +- llm_router_host.py | 24 +++++++++++++-------- route_reliability.py | 17 +++++++++++++++ shim.py | 28 +++++++++++++++++++------ tests/test_async_concurrency.py | 12 +++++++---- tests/test_host.py | 31 ++++++++++++++++----------- tests/test_route_reliability.py | 37 +++++++-------------------------- 7 files changed, 89 insertions(+), 62 deletions(-) diff --git a/core b/core index 8acfa9c..4409f6f 160000 --- a/core +++ b/core @@ -1 +1 @@ -Subproject commit 8acfa9c7eb2667dd0ae148ce4e5453c24f7e2049 +Subproject commit 4409f6f03419535b9dcad9372475c97700a2407a diff --git a/llm_router_host.py b/llm_router_host.py index 67f44e8..f361fab 100644 --- a/llm_router_host.py +++ b/llm_router_host.py @@ -376,8 +376,8 @@ 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) @@ -385,7 +385,9 @@ async def _resolve_call_async(self, request: dict, call_override=None) -> dict: 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( diff --git a/route_reliability.py b/route_reliability.py index ecb9271..0f6c223 100644 --- a/route_reliability.py +++ b/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,6 +44,7 @@ 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: @@ -46,12 +52,23 @@ def success_rate(key: str) -> float | None: 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() diff --git a/shim.py b/shim.py index 0828e33..560f311 100644 --- a/shim.py +++ b/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] diff --git a/tests/test_async_concurrency.py b/tests/test_async_concurrency.py index 9cff3ed..214bed6 100644 --- a/tests/test_async_concurrency.py +++ b/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?)" diff --git a/tests/test_host.py b/tests/test_host.py index 6218ed6..26e6757 100644 --- a/tests/test_host.py +++ b/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): diff --git a/tests/test_route_reliability.py b/tests/test_route_reliability.py index e01855f..fceb834 100644 --- a/tests/test_route_reliability.py +++ b/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