diff --git a/host_store.py b/host_store.py index 6e7a166..733f43e 100644 --- a/host_store.py +++ b/host_store.py @@ -100,6 +100,29 @@ def _retention_days() -> int: # on a DB that predates them. #3: the executed route identity + cache tokens. "ALTER TABLE calls ADD COLUMN IF NOT EXISTS tokens_cached BIGINT", "ALTER TABLE calls ADD COLUMN IF NOT EXISTS served_by TEXT", + # Per-ATTEMPT route observations (one row per provider call the engine made, + # including failed fallback tries — a grain `calls` does NOT have: `calls` is + # per-REQUEST, final route only). The RAW from which reliability/latency are + # derived on the fly (route_stats), replacing the in-process EMA dicts. Written + # by the fold site in llm_router_host; route_key identity stays host-internal + # (provider|family|served_by), never enters the signature. + """CREATE TABLE IF NOT EXISTS route_observations ( + id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + ts BIGINT NOT NULL, + provider_id TEXT, + model_family TEXT, + served_by TEXT, + ok BOOLEAN NOT NULL, + latency_ms DOUBLE PRECISION, + tools_requested BOOLEAN, + tool_calls_emitted BOOLEAN + )""", + "CREATE INDEX IF NOT EXISTS idx_route_obs_ts ON route_observations(ts)", + "CREATE INDEX IF NOT EXISTS idx_route_obs_route" + " ON route_observations(provider_id, model_family, served_by, ts)", + # #4c: learned tool capability is derived from these per-attempt signals. + "ALTER TABLE route_observations ADD COLUMN IF NOT EXISTS tools_requested BOOLEAN", + "ALTER TABLE route_observations ADD COLUMN IF NOT EXISTS tool_calls_emitted BOOLEAN", """CREATE TABLE IF NOT EXISTS settings_overrides ( key TEXT PRIMARY KEY, value TEXT NOT NULL, updated_at BIGINT NOT NULL )""", @@ -237,25 +260,32 @@ def insert_call(row: dict[str, Any]) -> None: def _prune() -> None: try: - cutoff = int(time.time()) - _RETENTION_DAYS * 86400 + now = time.time() + # calls.ts is in SECONDS; route_observations.ts is in MILLISECONDS. with _get_pool().connection() as conn: - conn.execute("DELETE FROM calls WHERE ts < %s", (cutoff,)) + conn.execute("DELETE FROM calls WHERE ts < %s", + (int(now) - _RETENTION_DAYS * 86400,)) + conn.execute("DELETE FROM route_observations WHERE ts < %s", + (int(now * 1000) - _RETENTION_DAYS * 86400 * 1000,)) except Exception as exc: # noqa: BLE001 _log.warning("host_store prune failed: %s", exc) -# A bounded background queue + single worker keeps the ledger insert OFF the -# request's latency path; the row is snapshotted (dict) so a later caller mutation -# can't change what gets written, and the queue is capped so a slow DB cannot grow -# an unbounded backlog — best-effort telemetry, so a dropped row is acceptable. -_write_q: "queue.Queue[dict[str, Any]]" = queue.Queue(maxsize=_WRITE_QUEUE_MAX) +# A bounded background queue + single worker keeps every write OFF the request's +# latency path; the payload is snapshotted so a later caller mutation can't change +# what gets written, and the queue is capped so a slow DB cannot grow an unbounded +# backlog — best-effort telemetry, so a dropped row is acceptable. The queue holds +# thunks so it serves both the call ledger and the per-attempt route observations. +_write_q: "queue.Queue" = queue.Queue(maxsize=_WRITE_QUEUE_MAX) def _writer_loop() -> None: while True: - row = _write_q.get() + job = _write_q.get() try: - insert_call(row) + job() + except Exception as exc: # noqa: BLE001 — a bad row must not kill the writer + _log.warning("host_store background write failed: %s", exc) finally: _write_q.task_done() @@ -263,16 +293,28 @@ def _writer_loop() -> None: threading.Thread(target=_writer_loop, name="host-store-writer", daemon=True).start() -def insert_call_async(row: dict[str, Any]) -> None: - """Record a call WITHOUT blocking the caller — enqueue a SNAPSHOT for the - background writer. Drops the row (best-effort) if the queue is full.""" +def _enqueue(job) -> None: try: - _write_q.put_nowait(dict(row)) + _write_q.put_nowait(job) except queue.Full: - _log.warning("host_store: write queue full (%d); dropping ledger row", - _WRITE_QUEUE_MAX) + _log.warning("host_store: write queue full (%d); dropping row", _WRITE_QUEUE_MAX) except Exception as exc: # noqa: BLE001 — never break a request - _log.warning("host_store insert_call_async failed: %s", exc) + _log.warning("host_store enqueue failed: %s", exc) + + +def insert_call_async(row: dict[str, Any]) -> None: + """Record a call WITHOUT blocking the caller — enqueue a SNAPSHOT for the + background writer. Drops the row (best-effort) if the queue is full.""" + snap = dict(row) + _enqueue(lambda: insert_call(snap)) + + +def observe_route_call_async(row: dict[str, Any]) -> None: + """Record one per-ATTEMPT route observation (provider/family/served_by + ok + + latency) off the latency path. The raw from which route_stats() derives + reliability/latency on the fly — replaces the in-process EMA folds.""" + snap = dict(row) + _enqueue(lambda: _insert_route_observation(snap)) def recent_calls(limit: int = 100) -> list[dict[str, Any]]: @@ -416,6 +458,178 @@ def set_consumer_keys(records: dict[str, Any]) -> bool: return False +# ---- route_observations (per-attempt raw; reliability/latency derived on the fly) + +def _insert_route_observation(row: dict[str, Any]) -> None: + """Append one per-attempt route observation. Fail-soft; best-effort telemetry.""" + try: + with _get_pool().connection() as conn: + conn.execute( + "INSERT INTO route_observations" + " (ts, provider_id, model_family, served_by, ok, latency_ms," + " tools_requested, tool_calls_emitted)" + " VALUES (%s,%s,%s,%s,%s,%s,%s,%s)", + (int(row.get("ts") or time.time() * 1000), + row.get("provider_id"), row.get("model_family"), row.get("served_by"), + bool(row.get("ok")), + float(row["latency_ms"]) if row.get("latency_ms") is not None else None, + bool(row.get("tools_requested")), bool(row.get("tool_calls_emitted")))) + except Exception as exc: # noqa: BLE001 — the fold must never break a request + _log.warning("host_store route observation insert failed: %s", exc) + + +def route_stats(window_ms: int = 900_000) -> dict[str, dict[str, Any]]: + """Per-route reliability + latency, DERIVED on the fly from route_observations + over the last `window_ms` — one aggregate query, not per-candidate. Returns + {route_key: {success_rate, latency_ms, count}} where route_key = + provider|family|served_by (host-internal, matches route_reliability.route_key). + Latency averages successful calls only (an error's latency is noise). Fail-soft + -> {} so the offer is left unstamped and the algebra falls back to its default.""" + try: + cutoff = int(time.time() * 1000) - max(0, window_ms) + with _get_pool().connection() as conn: + cur = conn.execute( + "SELECT provider_id, model_family, served_by," + " avg(CASE WHEN ok THEN 1.0 ELSE 0.0 END) AS success_rate," + " avg(latency_ms) FILTER (WHERE ok AND latency_ms IS NOT NULL) AS latency_ms," + " count(*) AS n" + " FROM route_observations WHERE ts >= %s" + " GROUP BY provider_id, model_family, served_by", (cutoff,)) + out: dict[str, dict[str, Any]] = {} + for prov, fam, sby, sr, lat, n in cur.fetchall(): + key = f"{prov}|{fam}|{sby}" + out[key] = { + "success_rate": float(sr) if sr is not None else None, + "latency_ms": int(round(lat)) if lat is not None else None, + "count": int(n)} + return out + except Exception as exc: # noqa: BLE001 — measurement read is best-effort + _log.warning("host_store route_stats failed: %s", exc) + return {} + + +def tool_incapable_routes(window_ms: int = 1_800_000, min_samples: int = 20) -> "set[str]": + """Routes that have proven they ignore tools — derived (#4c) from + route_observations: >= min_samples tools-requests within the last window_ms and + ZERO tool_calls emitted. The window IS the re-test horizon (a route ages out if + it stops being tool-tested); ANY tool_call in the window clears it. Capable is + the default (a route not in this set), so offers_sync drops supports_tools only + for the proven-incapable. Fail-soft -> empty set (everyone stays capable).""" + try: + cutoff = int(time.time() * 1000) - max(0, window_ms) + with _get_pool().connection() as conn: + cur = conn.execute( + "SELECT provider_id, model_family, served_by FROM route_observations" + " WHERE ts >= %s AND tools_requested" + " GROUP BY provider_id, model_family, served_by" + " HAVING count(*) >= %s AND" + " coalesce(sum(CASE WHEN tool_calls_emitted THEN 1 ELSE 0 END), 0) = 0", + (cutoff, min_samples)) + return {f"{p}|{f}|{s}" for p, f, s in cur.fetchall()} + except Exception as exc: # noqa: BLE001 + _log.warning("host_store tool_incapable_routes failed: %s", exc) + return set() + + +# ---- per-session views, derived on the fly from `calls` (#4b) ------------------- +# `calls` is per-request and carries session_id / status / provider_id / +# model_family / served_by / tokens / cost / caller — everything the in-process +# cache-affinity + session-meter folds held, now derived (fleet-consistent). + +_SESSION_TOTALS_ZERO = {"calls": 0, "tokens_in": 0, "tokens_out": 0, + "tokens_cached": 0, "cost_usd": 0.0} + + +def hot_route(session: "str | None") -> "str | None": + """The route (provider|family|served_by) that most recently served this + session SUCCESSFULLY — its prompt-cache prefix is hot there. None when unknown. + Matches route_reliability.route_key; resolved per request into the cache_hot + field. Fail-soft.""" + if not session: + return None + try: + with _get_pool().connection() as conn: + row = conn.execute( + "SELECT provider_id, model_family, served_by FROM calls" + " WHERE session_id = %s AND status < 400 AND served_by IS NOT NULL" + " ORDER BY ts DESC, id DESC LIMIT 1", (session,)).fetchone() + return f"{row[0]}|{row[1]}|{row[2]}" if row else None + except Exception as exc: # noqa: BLE001 + _log.warning("host_store hot_route failed: %s", exc) + return None + + +def session_totals(session: "str | None") -> dict[str, Any]: + """The session's running totals (calls, tokens_in/out/cached, cost_usd) summed + over its committed `calls`. The caller adds the in-flight call on top. Fail-soft + -> zeros.""" + if not session: + return dict(_SESSION_TOTALS_ZERO) + try: + with _get_pool().connection() as conn: + r = conn.execute( + "SELECT count(*), coalesce(sum(tokens_in),0), coalesce(sum(tokens_out),0)," + " coalesce(sum(tokens_cached),0), coalesce(sum(cost_usd),0)" + " FROM calls WHERE session_id = %s", (session,)).fetchone() + return {"calls": int(r[0]), "tokens_in": int(r[1]), "tokens_out": int(r[2]), + "tokens_cached": int(r[3]), "cost_usd": round(float(r[4]), 6)} + except Exception as exc: # noqa: BLE001 + _log.warning("host_store session_totals failed: %s", exc) + return dict(_SESSION_TOTALS_ZERO) + + +def session_warm(session: "str | None") -> list[dict[str, Any]]: + """The session's warm routes for DISPLAY: per family, the most recent route + that served it successfully ({family, provider, served_by}). Fail-soft -> [].""" + if not session: + return [] + try: + with _get_pool().connection() as conn: + cur = conn.execute( + "SELECT DISTINCT ON (model_family) model_family, provider_id, served_by" + " FROM calls WHERE session_id = %s AND status < 400" + " AND model_family IS NOT NULL" + " ORDER BY model_family, ts DESC, id DESC", (session,)) + return [{"family": f, "provider": p, "served_by": s or p} + for f, p, s in cur.fetchall()] + except Exception as exc: # noqa: BLE001 + _log.warning("host_store session_warm failed: %s", exc) + return [] + + +def session_owner(session: "str | None") -> "str | None": + """The consumer that owns the session = the caller of its EARLIEST call + (first-writer-wins, for cross-consumer isolation). None when unknown.""" + if not session: + return None + try: + with _get_pool().connection() as conn: + row = conn.execute( + "SELECT caller FROM calls WHERE session_id = %s" + " ORDER BY ts ASC, id ASC LIMIT 1", (session,)).fetchone() + return row[0] if row else None + except Exception as exc: # noqa: BLE001 + _log.warning("host_store session_owner failed: %s", exc) + return None + + +def all_session_totals() -> dict[str, dict[str, Any]]: + """Every session's totals (operator /x/sessions view), derived from calls.""" + try: + with _get_pool().connection() as conn: + cur = conn.execute( + "SELECT session_id, count(*), coalesce(sum(tokens_in),0)," + " coalesce(sum(tokens_out),0), coalesce(sum(tokens_cached),0)," + " coalesce(sum(cost_usd),0) FROM calls" + " WHERE session_id IS NOT NULL GROUP BY session_id") + return {s: {"calls": int(c), "tokens_in": int(ti), "tokens_out": int(to), + "tokens_cached": int(tc), "cost_usd": round(float(cu), 6)} + for s, c, ti, to, tc, cu in cur.fetchall()} + except Exception as exc: # noqa: BLE001 + _log.warning("host_store all_session_totals failed: %s", exc) + return {} + + # ---- peer_offers (antseed marketplace book; written by the sidecar) ------------ # Columns surfaced to the reader, in the row shape sources/antseed expects. The @@ -515,6 +729,9 @@ def reset() -> None: """Test hook: close + forget the pool (recreated against DATABASE_URL next use). For per-test isolation against a shared DB, truncate the tables.""" global _pool, _inserts_since_prune + # Drain pending async writes first: closing the pool while the background + # writer holds a connection races and flakes test isolation. + _write_q.join() with _pool_lock: if _pool is not None: _pool.close() @@ -527,4 +744,4 @@ def truncate_all_for_tests() -> None: """Test helper: wipe every table for isolation against a shared Postgres.""" with _get_pool().connection() as conn: conn.execute("TRUNCATE calls, settings_overrides, provider_overlays," - " consumer_keys, peer_offers, buyer_status") + " consumer_keys, peer_offers, buyer_status, route_observations") diff --git a/llm_router_host.py b/llm_router_host.py index e6bf64f..b8ea939 100644 --- a/llm_router_host.py +++ b/llm_router_host.py @@ -21,11 +21,8 @@ from pathlib import Path from typing import Callable +import host_store import route_reliability as _route_reliability -import route_latency as _route_latency -import route_tool_capability as _route_tool_capability -import route_cache as _route_cache -import route_session_meter as _route_session_meter from provider_adapters.anthropic import make_anthropic_async_call_provider from provider_adapters.common import ( AsyncCallProviderHook, @@ -155,7 +152,7 @@ def _inject_host_fields(self) -> None: peerless routes) has exactly one source and cannot drift across the Python/Lua boundary. It compares that key to the hot route the host resolved into `ctx.request.cache_hot_route` per request (see - `route_cache.hot_route`). The algebra observes only the Bool; the route + `host_store.hot_route`). The algebra observes only the Bool; the route key never enters the signature.""" # One source of truth for the route-key serialization: the getter must # build a candidate's key identically to the fold, or affinity is silently @@ -501,10 +498,10 @@ async def _resolve_call_async(self, request: dict, call_override=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 / 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 + # Record the outcome here (not in the hook) so the streaming/override path — + # all of opencode's traffic, and every flow node — writes a route + # observation too, the host-owned perf the algebra reads (derived) and the + # market view surfaces (#15/#4a). Mocks record as well, so a mocked call # is measured exactly like a live one. _fold_route_outcome(request, result, session=session) return result @@ -649,10 +646,10 @@ def _fold_route_outcome(request: dict, result: dict, reliability (success EMA), latency (EMA), and learned tool capability. Called from _resolve_call_async for BOTH the direct hook and the streaming/override path — the streaming result carries `ok` + `latency_ms` too — so opencode's - streamed traffic and every Σ_flow node feed the same EMAs the algebra reads - (offer.success_rate / offer.latency_ms). Previously this lived inside the - direct hook only, so route_latency/reliability stayed empty for the streaming - traffic that is, in practice, all of it.""" + streamed traffic and every Σ_flow node feed the same route_observations the + algebra reads derived (offer.success_rate / offer.latency_ms). Previously this + lived inside the direct hook only, so reliability/latency stayed empty for the + streaming traffic that is, in practice, all of it.""" # The route's identity is at the request TOP LEVEL (provider_id / model_family # / peer_id) — the core stamps it there for every call. The per-call `offer` # dict is only attached by some sources (antseed marketplace) and is None for @@ -672,21 +669,17 @@ def _fold_route_outcome(request: dict, result: dict, # 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) - # Per-session cache affinity: a successful call makes this route the session's - # hot route (it now holds the prompt-cache prefix), so the next turn's - # cache_hot field marks it and a cache-aware policy keeps it sticky. Same one - # route identity as reliability/latency; no-op when the caller named no session. - _route_cache.observe(session, rkey, ok) - # Record the warm route for the per-session display panel (which family is - # warm, on which provider, via which peer/backend). Success only, like the - # affinity fold. Display-only; the routing decision stays in route_cache. - if ok and session: - _route_session_meter.observe_route(session, pid, fam, peer_id or pid) - # Learned tool capability is a marketplace concern only (static/partner routes - # declare their capabilities in config), so keep it peer-scoped. - if peer_id: - _route_tool_capability.observe( - rkey, bool(request.get("tools")), - bool((result.get("response") or {}).get("tool_calls"))) + # #4a/#4c: record the per-ATTEMPT raw observation (every provider call, + # including failed fallbacks — a grain `calls` lacks). reliability, latency AND + # learned tool capability are derived on the fly from route_observations, not + # folded in-process. served_by = the peer for marketplace routes, else the + # provider — the identity route_stats / tool_incapable_routes key on. The tool + # signals carry only on tools-requests (capability is a marketplace concern). + host_store.observe_route_call_async({ + "ts": int(time.time() * 1000), "provider_id": pid, "model_family": fam, + "served_by": peer_id or pid, "ok": ok, "latency_ms": result.get("latency_ms"), + "tools_requested": bool(request.get("tools")), + "tool_calls_emitted": bool((result.get("response") or {}).get("tool_calls"))}) + # Cache affinity + the per-session meter are DERIVED on the fly from `calls` + # now (#4b): the route_observation above + the ingress's call ledger carry + # everything (session, route, outcome, tokens, cost), so no per-session fold. diff --git a/route_cache.py b/route_cache.py deleted file mode 100644 index 27a9aee..0000000 --- a/route_cache.py +++ /dev/null @@ -1,67 +0,0 @@ -"""Host-side per-session cache affinity (the host half of the cache_hot field). - -Prompt caching is STATE between calls: a provider discounts the reused prefix -only if the SAME peer serves the session again, so the host must remember which -route last served each session and steer the next turn back to it. That memory -cannot live in the algebra (a policy is a pure function of one call) nor in a -source's `offers_sync` (request-blind, shared across sessions) — it is -per-session host state, exactly like `route_latency` is per-route host state. - -The measurement is the simplest honest one: the route that most recently served -a session SUCCESSFULLY is its hot route (it holds the KV prefix). `observe` -folds each call outcome; `hot_route` returns the session's current hot route key -(or None). Only successful calls fold, exactly like `route_latency` — a failure -carries no honest "this peer holds the prefix" signal. - -The algebra never sees the route key: per request the host resolves -`hot_route(session)` into `ctx.request.cache_hot_route`, and the `cache_hot` -field getter (declared host-side in `LLMRouterHost`) reconstructs each -candidate's route key the same way `_fold_route_outcome` does and compares, -exposing only a Bool. Route identity stays 100% host-internal. - -In-process (resets on restart), exactly like `route_latency` / -`route_reliability`; a new or unknown session has no hot route -> no candidate -is cache_hot -> the policy routes purely on its other terms (no phantom -affinity for a fresh session). Reuses `route_reliability.route_key` so a -session's hot route shares the one route identity. Fleet-scale (multi-process) -affinity needs a shared store with TTL + eviction — the same in-process-state -debt the sibling forms already carry, not new debt. -""" -from __future__ import annotations - -import threading - -from route_reliability import route_key # shared route identity # noqa: F401 (re-exported) - -_lock = threading.Lock() -# session -> route_key of the last SUCCESSFUL call on that session. -_hot: dict[str, str] = {} - - -def observe(session: "str | None", key: str, ok: bool) -> None: - """Fold one call outcome. A successful call makes its route the session's hot - route (it now holds the prefix). Failures are ignored (no honest signal), and - a missing session is a no-op — affinity only exists for sessions the caller - names.""" - if not session or not ok: - return - with _lock: - _hot[session] = key - - -def hot_route(session: "str | None") -> "str | None": - """The route key holding this session's prefix hot, or None if unknown.""" - if not session: - return None - return _hot.get(session) - - -def snapshot() -> dict[str, str]: - with _lock: - return dict(_hot) - - -def reset() -> None: - """Test hook.""" - with _lock: - _hot.clear() diff --git a/route_latency.py b/route_latency.py deleted file mode 100644 index 15dc2f9..0000000 --- a/route_latency.py +++ /dev/null @@ -1,76 +0,0 @@ -"""Host-side per-route latency fold (the host half of the latency form). - -The twin of `route_reliability`: MEASURING speed is the host's job, and the -algebra already reads it as a per-candidate field (`offer.latency_ms`, a CORE -field, observed pointwise like price/success_rate). This is the host's -measurement: a per-route EMA of observed end-to-end call latency, stamped onto -offers in `offers_sync` so a policy can gate/score on speed — a slow $0 -marketplace peer (e.g. a 12 s antseed glm-5.2) ranks below a fast partner -(a ~0.9 s gpt-5.5), instead of winning every cost-led policy and stalling the -caller. - -Only SUCCESSFUL calls fold in. A fast failure (an antseed peer returning -"empty assistant content" in 1.9 s) must NOT make a broken route look fast — that -is reliability's concern (`route_reliability`), and counting its latency would -reward exactly the route we want to avoid. - -The measurement is deliberately the simplest honest one — an EMA of total -response latency (TTFT + generation), which is what the non-streaming backend can -observe per call. Richer judgment the form allows (true TTFT from the first -stream chunk, p50/p95 windowing, population-relative) can grow here, host-side, -without touching the algebra. Folding happens in `llm_router_host` on each call -outcome; `offers_sync` reads `latency_ms(...)` to stamp the offer. - -In-process (resets on restart), exactly like `route_reliability`; a missing route -reads None -> the offer is left unstamped -> the algebra falls back to the field -default (an unmeasured route is optimistically routable, then learns down on its -first slow observation — the same optimistic-default discipline as -`route_reliability` / `route_tool_capability`). Reuses `route_reliability.route_key` -so a route's latency and reliability share one identity. -""" -from __future__ import annotations - -import threading - -from route_reliability import route_key # shared route identity # noqa: F401 (re-exported) - -# Latency shifts with load faster than reliability does, so smooth a little less -# heavily (weight a fresh sample more) than route_reliability's 0.2. -_ALPHA = 0.3 - -_lock = threading.Lock() -_ema: dict[str, float] = {} - - -def observe(key: str, latency_ms: "int | float | None", ok: bool) -> None: - """Fold one observed call latency into the route's EMA. Ignored unless the - call SUCCEEDED and the latency is a positive number — a failed/instant - outcome carries no honest speed signal.""" - if not ok or latency_ms is None: - return - try: - v = float(latency_ms) - except (TypeError, ValueError): - return - if v <= 0: - return - with _lock: - cur = _ema.get(key) - _ema[key] = v if cur is None else _ALPHA * v + (1.0 - _ALPHA) * cur - - -def latency_ms(key: str) -> "int | None": - """The route's folded latency in ms (rounded), or None if never observed.""" - v = _ema.get(key) - return None if v is None else round(v) - - -def snapshot() -> dict[str, int]: - with _lock: - return {k: round(v) for k, v in _ema.items()} - - -def reset() -> None: - """Test hook.""" - with _lock: - _ema.clear() diff --git a/route_reliability.py b/route_reliability.py index 0f6c223..11c42a2 100644 --- a/route_reliability.py +++ b/route_reliability.py @@ -1,74 +1,27 @@ -"""Host-side per-route reliability fold (the host half of the reliability form). - -MEASURING reliability is the host's job — llm-router #14 made the algebra read it -as a per-candidate field (`offer.success_rate`), like price, observed pointwise. -This is the host's measurement: a per-route success-rate EMA over observed call -outcomes, stamped onto AntSeed offers so the router prefers the reliable seller -and rotates off a broken one (a different route for the same family). - -The measurement is deliberately the simplest honest one — a binary success EMA. -All of the richer judgment the form allows (weighting by error kind, latency-with- -error, population-relative, windowing) lives here, host-side, and can grow without -touching the algebra. Folding happens in `llm_router_host` on each call outcome; -`offers_sync` reads `success_rate(...)` to stamp the offer. - -In-process (resets on restart), exactly like the engine EMA it replaces for -marketplace routes; a missing route reads None → the offer is left unstamped → the -algebra falls back to its own coarse EMA / the field default. +"""Host-side route identity. + +MEASURING per-route reliability/latency is the host's job (llm-router #14/#15: the +algebra reads them as per-candidate fields, like price). The measurement used to +be an in-process EMA folded here; since #4a it is DERIVED on the fly from the raw +per-attempt `route_observations` ledger (`host_store.route_stats`) — fleet- +consistent and surviving restarts, instead of a per-pod dict. + +What stays here is the one thing that is pure identity, not measurement: the +`route_key` that names a route (a specific seller peer serving a family, or the +provider itself for a peerless gateway route). It is reused by host_store route/session derivations, +the offer-stamp sites, and matches the key +`route_stats` aggregates on. It stays entirely host-internal; the algebra never +sees it. """ from __future__ import annotations -import threading - -# Same smoothing the engine EMA used (ema_alpha). First observation seeds the -# rate directly; subsequent ones decay geometrically toward it. -_ALPHA = 0.2 - -_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: - """Identity of a route = a specific seller peer serving a specific family. - Stays entirely host-internal; the algebra never sees it.""" + """Identity of a route = provider|family|served_by (the peer for a marketplace + route, else the provider). Host-internal; never enters the signature.""" return f"{provider_id}|{model_family}|{peer_id}" -def observe(key: str, ok: bool) -> None: - s = 1.0 if ok else 0.0 - 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() + """Test hook. Reliability/latency state now lives in route_observations (reset + by truncating the host store); kept as a no-op so callers need no change.""" diff --git a/route_session_meter.py b/route_session_meter.py deleted file mode 100644 index 63fd5cd..0000000 --- a/route_session_meter.py +++ /dev/null @@ -1,110 +0,0 @@ -"""Per-session usage meter: per-call is already on x_router; this accumulates the -RUNNING TOTAL per session (the sid the caller sends), so a conversation/agent has -both numbers — this call AND everything it has spent so far. Pairs with -route_cache (same session key): route_cache says which peer is hot, this says what -the session has cost and how much of it was served from cache. - -Tracks per session: calls, tokens_in, tokens_out, tokens_cached, cost_usd. The -cache hit ratio (tokens_cached / tokens_in) and the realized cost are then exact -per session, using the provider's reported cost — no model-price guessing. - -In-process (resets on restart), like route_cache / route_latency; fleet scale -needs a shared store with TTL — the same debt the sibling forms carry. -""" -from __future__ import annotations - -import threading - -_lock = threading.Lock() -_acc: dict[str, dict] = {} # session -> running totals -# session -> {model_family -> {family, provider, served_by}}: the routes that -# successfully served the session, i.e. which models/peers are WARM (hold the -# session's prompt-cache prefix). For DISPLAY (the warm panel); the affinity -# decision stays in route_cache. Per family, so a flow's glm AND gpt both show. -_warm: dict[str, dict[str, dict]] = {} -# session -> owning consumer key (the authed caller that FIRST wrote this sid). -# A session's economics (cost/tokens/cache + warm peers) belong to one consumer; -# this binding lets the consumer-facing view refuse to disclose another -# consumer's session (cross-consumer isolation). First-writer-wins: a different -# consumer reusing someone else's opaque sid must NOT steal or overwrite it. -_owner: dict[str, str] = {} - - -def observe(session: "str | None", *, tokens_in=0, tokens_out=0, - tokens_cached=0, cost_usd=0.0, owner: "str | None" = None) -> "dict | None": - """Fold one call's usage into the session's running total; return the new - accumulated totals (so the caller can put per-call AND acc on the response). - No-op (returns None) when the caller named no session. When `owner` (the - authed consumer key) is given, bind sid->owner first-writer-wins so the - consumer-facing view can scope reads to the owning consumer.""" - if not session: - return None - with _lock: - if owner: - _owner.setdefault(session, owner) - a = _acc.get(session) - if a is None: - a = {"calls": 0, "tokens_in": 0, "tokens_out": 0, - "tokens_cached": 0, "cost_usd": 0.0} - _acc[session] = a - a["calls"] += 1 - a["tokens_in"] += int(tokens_in or 0) - a["tokens_out"] += int(tokens_out or 0) - a["tokens_cached"] += int(tokens_cached or 0) - a["cost_usd"] = round(a["cost_usd"] + float(cost_usd or 0.0), 6) - return dict(a) - - -def observe_route(session: "str | None", provider: "str | None", - family: "str | None", served_by: "str | None") -> None: - """Record (for DISPLAY) that `family` was served warm for this session by - `provider` via `served_by` (the peer / real backend behind a marketplace, or - the provider itself for direct routes). Keyed per family so a multi-family - flow shows all warm models. No-op without a session/family.""" - if not session or not family: - return - with _lock: - w = _warm.get(session) - if w is None: - w = {} - _warm[session] = w - w[family] = {"family": family, "provider": provider, - "served_by": served_by or provider} - - -def warm(session: "str | None") -> list[dict]: - """The session's warm routes: [{family, provider, served_by}], one per family.""" - if not session: - return [] - with _lock: - return list((_warm.get(session) or {}).values()) - - -def owner(session: "str | None") -> "str | None": - """The consumer key that owns this session (first writer), or None for an - unknown / unnamed session. Used to scope the consumer-facing session view.""" - if not session: - return None - with _lock: - return _owner.get(session) - - -def get(session: "str | None") -> "dict | None": - if not session: - return None - with _lock: - a = _acc.get(session) - return dict(a) if a else None - - -def snapshot() -> dict[str, dict]: - with _lock: - return {s: dict(a) for s, a in _acc.items()} - - -def reset() -> None: - """Test hook.""" - with _lock: - _acc.clear() - _warm.clear() - _owner.clear() diff --git a/route_tool_capability.py b/route_tool_capability.py deleted file mode 100644 index 4ab5525..0000000 --- a/route_tool_capability.py +++ /dev/null @@ -1,87 +0,0 @@ -"""Host-side per-route LEARNED tool capability — the sound close to the AntSeed -optimistic `supports_tools=true` default (sources/antseed.py). - -AntSeed market rows carry no capability data, so offers_sync stamps -`supports_tools=true` for every peer. KNOWN HOLE that this module closes: a peer -that ACCEPTS the `tools` field but whose model does not actually function-call -returns plain text — no tool_calls, no error — a silent tools-less answer that no -error-path retry catches. - -We learn it, by the same shape as route_reliability: on each call whose REQUEST -carried tools, observe whether the RESPONSE carried tool_calls. A route -(provider|family|peer) that, over `_MIN_SAMPLES` tools-requests, emitted tool_calls -in NONE is marked tool-incapable; `offers_sync` then stops stamping supports_tools -for it, so the core's meets_req filters it out of tool requests (it still serves -non-tool requests). - -Two honesty guards on the inherent ambiguity (a single response can't tell -"ignored the tools" from "the model chose not to call one"): -- the signal is statistical — `_MIN_SAMPLES` high, and ANY observed tool_call - proves capability permanently (a capable route is never demoted); -- a tool-incapable mark EXPIRES after `_EXPIRY_MS`, so a false positive only - sidelines a route briefly before it is re-tested. - -In-process (resets on restart), like route_reliability. Route keys are built by -route_reliability.route_key — one source for the route identity. -""" -from __future__ import annotations - -import threading -import time - -# Mark a route tool-incapable after this many tools-requests with zero tool_calls. -_MIN_SAMPLES = 20 -# A tool-incapable mark expires this long after its window started, so the route -# is re-tested (a false positive only sidelines it briefly). -_EXPIRY_MS = 30 * 60 * 1000 - -_lock = threading.Lock() -# route_key -> {"reqs": tools-requests seen, "calls": of which emitted tool_calls, -# "since": now_ms when the current window started} -_state: dict[str, dict] = {} - - -def _now_ms() -> int: - """Wall clock in ms. Indirected so tests can monkeypatch it.""" - return int(time.time() * 1000) - - -def observe(key: str, tools_requested: bool, tool_calls_emitted: bool) -> None: - """Record one call outcome for a route. Only tools-requests carry signal.""" - if not tools_requested: - return - with _lock: - st = _state.get(key) - if st is None: - st = {"reqs": 0, "calls": 0, "since": _now_ms()} - _state[key] = st - st["reqs"] += 1 - if tool_calls_emitted: - st["calls"] += 1 - - -def is_capable(key: str) -> bool: - """True unless the route has proven it ignores tools (>= _MIN_SAMPLES - tools-requests, zero tool_calls). Optimistic by default; an expired - incapable mark is reset so the route is re-tested.""" - with _lock: - st = _state.get(key) - if st is None: - return True # optimistic default — unobserved routes are assumed capable - if not (st["reqs"] >= _MIN_SAMPLES and st["calls"] == 0): - return True # capable (or not enough evidence yet) - if _now_ms() - st["since"] >= _EXPIRY_MS: - _state[key] = {"reqs": 0, "calls": 0, "since": _now_ms()} # re-test - return True - return False - - -def snapshot() -> dict[str, dict]: - with _lock: - return {k: dict(v) for k, v in _state.items()} - - -def reset() -> None: - """Test hook.""" - with _lock: - _state.clear() diff --git a/shim.py b/shim.py index e6cc2c5..996bfdc 100644 --- a/shim.py +++ b/shim.py @@ -37,8 +37,7 @@ from fastapi.responses import JSONResponse, StreamingResponse from pydantic import BaseModel, ConfigDict -import route_cache -import route_session_meter +import host_store _log = logging.getLogger("unhardcoded.shim") @@ -282,20 +281,18 @@ def session_meter(sid: str, request: Request): consistent either way.""" caller = request.headers.get("x-llm-router-caller") if caller: - owner = route_session_meter.owner(sid) + owner = host_store.session_owner(sid) if owner is None or owner != caller: return JSONResponse(status_code=404, content={"error": { "message": "session not found", "type": "not_found", "code": "session_not_found"}}) - acc = route_session_meter.get(sid) or { - "calls": 0, "tokens_in": 0, "tokens_out": 0, - "tokens_cached": 0, "cost_usd": 0.0} - return {**acc, "warm": route_session_meter.warm(sid)} + acc = host_store.session_totals(sid) + return {**acc, "warm": host_store.session_warm(sid)} @app.get("/x/sessions") def session_meters(): """All session meters (operator view of per-session spend/cache).""" - return {"sessions": route_session_meter.snapshot()} + return {"sessions": host_store.all_session_totals()} @app.get("/x/calls") def recent_calls(limit: int = 100): @@ -446,8 +443,7 @@ 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 + import host_store catalog = host.catalog() or {} models = catalog.get("models") or {} state = host.dump_state() or {} @@ -457,24 +453,24 @@ def market_view(): 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() + # Live perf is host-owned now (#15): the engine folds no EMA, so build it + # from the host's per-route measurements — DERIVED on the fly from + # route_observations (#4a), aggregated across the peers/route ids that + # serve a given provider|family. None until the router has called it. + _stats = host_store.route_stats() # {route_key: {success_rate, latency_ms, count}} def _perf(provider, family): prefix = f"{provider}|{family}|" - keys = [k for k in _counts if k.startswith(prefix)] - total = sum(_counts[k] for k in keys) + rows = [v for k, v in _stats.items() if k.startswith(prefix)] + total = sum(r["count"] for r in rows) if not total: return None - 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) + sr_rows = [r for r in rows if r.get("success_rate") is not None] + lt_rows = [r for r in rows if r.get("latency_ms") is not None] + sr_calls = sum(r["count"] for r in sr_rows) + lt_calls = sum(r["count"] for r in lt_rows) + sr = sum(r["success_rate"] * r["count"] for r in sr_rows) + lt = sum(r["latency_ms"] * r["count"] for r in lt_rows) return {"success_rate": (sr / sr_calls) if sr_calls else None, "latency_ms": round(lt / lt_calls) if lt_calls else None, "calls": total} @@ -1253,7 +1249,7 @@ def _request_to_contract( # reads off ctx.request. A brand-new session has no hot route -> the key # is simply absent -> cache_hot is false for everyone (no phantom pin). contract["session"] = req.session - hot = route_cache.hot_route(req.session) + hot = host_store.hot_route(req.session) if hot is not None: contract["cache_hot_route"] = hot @@ -1427,13 +1423,17 @@ def _build_x_router(result: dict, subscription_providers=frozenset(), "compact": _compact_suggested(resp), } if session: - x_router["session_acc"] = route_session_meter.observe( - session, - tokens_in=resp.get("tokens_in") or 0, - tokens_out=resp.get("tokens_out") or 0, - tokens_cached=resp.get("tokens_cached") or 0, - cost_usd=x_router["cost_usd"] or 0.0, - owner=owner) + # #4b: derive the running total from the committed `calls` and add THIS + # in-flight call (not yet in the ledger). Owner is derived from the + # session's earliest call, so no explicit binding is needed here. + prior = host_store.session_totals(session) + x_router["session_acc"] = { + "calls": prior["calls"] + 1, + "tokens_in": prior["tokens_in"] + (resp.get("tokens_in") or 0), + "tokens_out": prior["tokens_out"] + (resp.get("tokens_out") or 0), + "tokens_cached": prior["tokens_cached"] + (resp.get("tokens_cached") or 0), + "cost_usd": round(prior["cost_usd"] + (x_router["cost_usd"] or 0.0), 6), + } return x_router diff --git a/sources/antseed.py b/sources/antseed.py index 0366057..4f41cee 100644 --- a/sources/antseed.py +++ b/sources/antseed.py @@ -17,8 +17,6 @@ import host_store import route_reliability as _route_reliability -import route_latency as _route_latency -import route_tool_capability as _route_tool_capability import settings from sources import Balance, Price @@ -206,23 +204,29 @@ def offers_sync(self, provider_id: str) -> list[dict]: self._stats["rejected_by_reputation"] = rejected_by_reputation self._stats["denied"] = denied self._stats["offers"] = len(kept_rows) + # #4a/#4c: reliability + latency + learned tool-incapability are derived on + # the fly from route_observations (one query each per offers_sync, not per + # candidate), keyed by route identity. + stats = host_store.route_stats() + incapable = host_store.tool_incapable_routes() offers = [] for row in kept_rows: family = row["family"] model = self._models.get(family) or {} rkey = _route_reliability.route_key(provider_id, family, row["peer_id"]) + rstat = stats.get(rkey) or {} # AntSeed rows carry no capability data, so supports_tools defaults to # true (else meets_req filters the whole peer market out of any tools # request). The default-true HOLE — a peer that accepts `tools` but # never function-calls returns a SILENT tools-less answer (no error, # no retry) — is closed by the LEARNED per-route signal: a route - # observed to ignore tools (route_tool_capability) is dropped from - # supports_tools, so meets_req filters it for tool requests while it - # still serves non-tool requests. The learned-incapable verdict - # overrides even a curated claim (the peer is the ground truth); - # everything else (json_mode, curated caps) is unchanged. + # observed to ignore tools (host_store.tool_incapable_routes) is dropped + # from supports_tools, so meets_req filters it for tool requests while it + # still serves non-tool requests. The learned-incapable verdict overrides + # even a curated claim (the peer is the ground truth); everything else + # (json_mode, curated caps) is unchanged. caps = {"supports_json_mode": True, **(model.get("capabilities") or {})} - if _route_tool_capability.is_capable(rkey): + if rkey not in incapable: caps.setdefault("supports_tools", True) else: caps.pop("supports_tools", None) @@ -247,12 +251,12 @@ def offers_sync(self, provider_id: str) -> list[dict]: # host-measured reliability for THIS route, stamped like price and # read pointwise by the algebra (offer.success_rate, llm-router # #14). None until observed -> algebra default/engine fallback. - "success_rate": _route_reliability.success_rate(rkey), + "success_rate": rstat.get("success_rate"), # host-measured latency for THIS route, stamped like success_rate # and read pointwise by the algebra (offer.latency_ms). None until # observed -> field default (optimistically routable, learns down # on its first slow call). Lets a policy route by speed. - "latency_ms": _route_latency.latency_ms(rkey), + "latency_ms": rstat.get("latency_ms"), }) return offers diff --git a/sources/openrouter.py b/sources/openrouter.py index 0ec9275..1b67c3d 100644 --- a/sources/openrouter.py +++ b/sources/openrouter.py @@ -8,8 +8,8 @@ import time from typing import Any +import host_store import route_reliability as _route_reliability -import route_latency as _route_latency from sources import Balance, Price BASE_URL = "https://openrouter.ai/api/v1" @@ -185,10 +185,13 @@ def offers_sync(self, provider_id: str) -> list[dict]: latency is keyed on the provider itself (peer == provider_id) — the same key the latency fold uses for peerless routes — making OpenRouter speed directly comparable to a marketplace peer's.""" + # #4a: latency derived on the fly from route_observations (one query), + # keyed on the provider itself for these peerless gateway routes. + stats = host_store.route_stats() out = [] for o in self._offers: lkey = _route_reliability.route_key(provider_id, o["model_family"], provider_id) - out.append({**o, "latency_ms": _route_latency.latency_ms(lkey)}) + out.append({**o, "latency_ms": (stats.get(lkey) or {}).get("latency_ms")}) return out def market_book(self) -> dict: diff --git a/tests/conftest.py b/tests/conftest.py index a4a275c..e18d9ee 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -32,7 +32,10 @@ def host_store_clean(): """Truncate the operational store before the test (isolation against the shared Postgres). Skips the test if Postgres is unreachable.""" try: - host_store.reset() + # Drain pending async writes from a prior test FIRST so they can't land + # after this truncate; don't close the pool (the background writer shares + # it — closing it mid-write races and flakes isolation). + host_store._write_q.join() host_store.truncate_all_for_tests() except Exception as exc: # noqa: BLE001 pytest.skip(f"host store Postgres unavailable: {exc}") @@ -78,6 +81,37 @@ def seed_peer_offers(peers, observed_at=None): rows) +def seed_call(session=None, provider=None, family=None, served_by=None, status=200, + caller=None, tokens_in=0, tokens_out=0, tokens_cached=0, cost_usd=0.0, + ts=None): + """Insert one per-request `calls` row (as the ingress would), the raw the + per-session views (hot_route / session_*) derive from. ts in SECONDS.""" + import time + host_store.insert_call({ + "ts": int(time.time()) if ts is None else ts, + "session": session, "provider": provider, "model_family": family, + "served_by": served_by, "served_model_id": served_by, "status": status, + "caller": caller, "key_sha256": None, "tokens_in": tokens_in, + "tokens_out": tokens_out, "tokens_total": tokens_in + tokens_out, + "tokens_cached": tokens_cached, "cost_usd": cost_usd}) + + +def seed_route_obs(provider, family, served_by, ok, latency_ms=None, n=1, ts=None, + tools_requested=False, tool_calls_emitted=False): + """Seed n per-attempt route_observations for a route (the raw from which + route_stats / tool_incapable_routes derive on the fly).""" + import time + t = int(time.time() * 1000) if ts is None else ts + rows = [(t, provider, family, served_by, ok, latency_ms, + tools_requested, tool_calls_emitted) for _ in range(n)] + with host_store._get_pool().connection() as conn: + conn.cursor().executemany( + "INSERT INTO route_observations" + " (ts, provider_id, model_family, served_by, ok, latency_ms," + " tools_requested, tool_calls_emitted)" + " VALUES (%s,%s,%s,%s,%s,%s,%s,%s)", rows) + + def seed_buyer_status(pid, pinned_peer_id=None, deposits_available=None, deposits_reserved=None, wallet_address=None, connection_state=None): diff --git a/tests/test_antseed_offers.py b/tests/test_antseed_offers.py index c627362..a16928f 100644 --- a/tests/test_antseed_offers.py +++ b/tests/test_antseed_offers.py @@ -16,9 +16,8 @@ sys.path.insert(0, str(ROOT)) import route_reliability as rr # noqa: E402 -import route_latency as rl # noqa: E402 from sources.antseed import AntSeedSource # noqa: E402 -from conftest import seed_peer_offers as _seed_market # noqa: E402 +from conftest import seed_peer_offers as _seed_market, seed_route_obs # noqa: E402 @pytest.fixture(autouse=True) @@ -65,10 +64,9 @@ def test_offers_sync_surfaces_top_n_distinct_peers(tmp_path): def test_offers_sync_stamps_host_measured_reliability(tmp_path): - rr.reset() _seed_market([_peer("peerA", 0.5), _peer("peerB", 1.0)]) # peerA observed failing -> demoted; peerB never observed -> unstamped - rr.observe(rr.route_key("antseed", FAMILY, "peerA"), False) + seed_route_obs("antseed", FAMILY, "peerA", ok=False) offers = AntSeedSource(CATALOG).offers_sync("antseed") by_peer = {o["peer_id"]: o for o in offers} assert by_peer["peerA"]["success_rate"] == 0.0 @@ -79,10 +77,8 @@ def test_offers_sync_stamps_host_measured_latency(tmp_path): # The latency twin of the reliability stamp: a peer observed slow carries its # measured latency_ms so a policy can route by speed; an unobserved peer is # left unstamped (None -> field default, optimistically routable). - rr.reset() - rl.reset() _seed_market([_peer("peerA", 0.5), _peer("peerB", 1.0)]) - rl.observe(rl.route_key("antseed", FAMILY, "peerA"), 12000, ok=True) + seed_route_obs("antseed", FAMILY, "peerA", ok=True, latency_ms=12000) offers = AntSeedSource(CATALOG).offers_sync("antseed") by_peer = {o["peer_id"]: o for o in offers} assert by_peer["peerA"]["latency_ms"] == 12000 @@ -122,13 +118,10 @@ def test_offers_sync_drops_supports_tools_for_learned_incapable_route(tmp_path): # the AntSeed default-true hole is closed by the learned signal: a route # observed to ignore tools is filtered from tool requests (no supports_tools), # while other caps and non-tool routing are unaffected. - import route_tool_capability as tc - rr.reset() - tc.reset() _seed_market([_peer("peerA", 0.5)]) - rkey = rr.route_key("antseed", FAMILY, "peerA") - for _ in range(tc._MIN_SAMPLES): # peerA never emits tool_calls on tool reqs - tc.observe(rkey, True, False) + # peerA emits no tool_calls on _MIN_SAMPLES (20) tools-requests -> incapable + seed_route_obs("antseed", FAMILY, "peerA", ok=True, n=20, + tools_requested=True, tool_calls_emitted=False) caps = AntSeedSource(CATALOG).offers_sync("antseed")[0]["capabilities"] assert "supports_tools" not in caps # learned-incapable -> filtered for tools assert caps.get("supports_json_mode") is True # other caps unaffected diff --git a/tests/test_async_concurrency.py b/tests/test_async_concurrency.py index 214bed6..799410f 100644 --- a/tests/test_async_concurrency.py +++ b/tests/test_async_concurrency.py @@ -69,13 +69,11 @@ async def hook(req): @pytest.mark.asyncio -async def test_shared_state_is_coherent_under_concurrency(): - """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 test_shared_state_is_coherent_under_concurrency(host_store_clean): + """Every coroutine records a route observation; the per-route observation + count must equal the number of calls with no lost updates. Reliability is + host-owned and derived from route_observations now (#15/#4a).""" + import host_store async def hook(req): await asyncio.sleep(0.01) @@ -85,5 +83,7 @@ async def hook(req): contract = {"profile": "default", "messages": [{"role": "user", "content": "hi"}]} await asyncio.gather(*(host.execute_async(dict(contract)) for _ in range(N))) - total_n = sum(rr.snapshot_counts().values()) + host_store._write_q.join() # drain the async route-obs writes + with host_store._get_pool().connection() as c: + total_n = c.execute("SELECT count(*) FROM route_observations").fetchone()[0] 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 095271e..e5669be 100644 --- a/tests/test_host.py +++ b/tests/test_host.py @@ -274,15 +274,14 @@ async def override(request): assert res2["response"]["text"] == "via-mock" -def test_streaming_override_path_folds_route_metrics(host): - # The fix: the override (streaming) path must feed the per-route EMAs too. - # Before, the fold lived only inside the direct hook, so route_latency / - # reliability stayed empty for opencode's all-streaming traffic and flow nodes - # — which is exactly why the flow couldn't rank by latency. +def test_streaming_override_path_folds_route_metrics(host, host_store_clean): + # The fix: the override (streaming) path must record a route observation too. + # Before, the fold lived only inside the direct hook, so reliability/latency + # stayed empty for opencode's all-streaming traffic and flow nodes — which is + # exactly why the flow couldn't rank by latency. Derived now (#4a) from + # route_observations, written async, so drain the queue before reading. import asyncio - import route_reliability as rr - import route_latency as rl - rr.reset(); rl.reset() + import host_store req = {"provider_id": "antseed", "model_family": "glm-5.2", "offer": {"model_family": "glm-5.2", "peer_id": "peerX"}} @@ -291,32 +290,35 @@ async def override(r): # a streamed result: ok + latency_ms return {"ok": True, "latency_ms": 12000, "response": {"tool_calls": None}} asyncio.run(host._resolve_call_async(req, call_override=override)) - k = rr.route_key("antseed", "glm-5.2", "peerX") - assert rl.latency_ms(k) == 12000 # latency folded from the streamed call - assert rr.success_rate(k) == 1.0 + host_store._write_q.join() # drain the async route-obs write + k = "antseed|glm-5.2|peerX" + st = host_store.route_stats() + assert st[k]["latency_ms"] == 12000 # latency derived from the streamed call + assert st[k]["success_rate"] == 1.0 - # A mock folds too now (#15: the host owns perf, so a mocked call is measured + # A mock records 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_store.truncate_all_for_tests() 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) == 5 + host_store._write_q.join() + assert host_store.route_stats()[k]["latency_ms"] == 5 -def test_fold_uses_top_level_route_identity(host): +def test_fold_uses_top_level_route_identity(host, host_store_clean): # The route identity is at the request TOP LEVEL (provider_id/model_family); # the per-call `offer` is None for openrouter/static/partner routes. The fold # read family from `offer` only, so those routes NEVER folded — latency_ms # stayed empty and a policy couldn't rank them by speed (the z.ai/glm-5.2 case). import asyncio - import route_reliability as rr - import route_latency as rl - rr.reset(); rl.reset() + import host_store req = {"provider_id": "openrouter", "model_family": "z-ai/glm-5.2", "served_model_id": "z-ai/glm-5.2"} # no offer, no peer_id async def override(r): return {"ok": True, "latency_ms": 1500, "response": {}} asyncio.run(host._resolve_call_async(req, call_override=override)) - k = rr.route_key("openrouter", "z-ai/glm-5.2", "openrouter") - assert rl.latency_ms(k) == 1500 # folded from top-level identity + host_store._write_q.join() + # peerless route -> served_by is the provider itself + k = "openrouter|z-ai/glm-5.2|openrouter" + assert host_store.route_stats()[k]["latency_ms"] == 1500 # from top-level identity diff --git a/tests/test_host_store.py b/tests/test_host_store.py index 90297d7..45cde74 100644 --- a/tests/test_host_store.py +++ b/tests/test_host_store.py @@ -249,3 +249,30 @@ def test_served_by_and_tokens_cached_recorded(store): store.insert_call(_row(usage_event_id="ev2")) r2 = [c for c in store.recent_calls() if c["usage_event_id"] == "ev2"][0] assert r2["served_by"] is None and r2["tokens_cached"] is None + + +# ---- route_observations -> route_stats (reliability/latency derived on the fly) - + +def test_route_stats_derives_reliability_and_latency(store): + from conftest import seed_route_obs + # peerA: 4 ok + 1 fail -> success 0.8; latency avg over OK only + seed_route_obs("antseed", "m", "peerA", ok=True, latency_ms=100, n=3) + seed_route_obs("antseed", "m", "peerA", ok=True, latency_ms=300) # ok=4, lat avg=150 + seed_route_obs("antseed", "m", "peerA", ok=False, latency_ms=9999) # failure: latency ignored + seed_route_obs("antseed", "m", "peerB", ok=True, latency_ms=50) + st = store.route_stats() + assert st["antseed|m|peerA"]["success_rate"] == 0.8 + assert st["antseed|m|peerA"]["latency_ms"] == 150 # avg(100,100,100,300), failure excluded + assert st["antseed|m|peerA"]["count"] == 5 + assert st["antseed|m|peerB"]["success_rate"] == 1.0 + assert "antseed|m|missing" not in st + + +def test_route_stats_window_excludes_old_observations(store): + from conftest import seed_route_obs + import time + now = int(time.time() * 1000) + seed_route_obs("p", "m", "fresh", ok=True, ts=now) + seed_route_obs("p", "m", "stale", ok=True, ts=now - 20 * 60 * 1000) # 20 min ago + assert set(store.route_stats(window_ms=15 * 60 * 1000)) == {"p|m|fresh"} + assert set(store.route_stats(window_ms=30 * 60 * 1000)) == {"p|m|fresh", "p|m|stale"} diff --git a/tests/test_metering.py b/tests/test_metering.py index 35201ad..c7c5326 100644 --- a/tests/test_metering.py +++ b/tests/test_metering.py @@ -17,7 +17,6 @@ from shim import _executed_cost_usd, _CACHE_READ_FACTOR # noqa: E402 from llm_router_host import _cached_tokens # noqa: E402 -import route_session_meter # noqa: E402 def _result(resp, chosen): @@ -70,22 +69,17 @@ def test_cached_tokens_across_usage_shapes(): assert _cached_tokens(None) is None -def test_session_owner_binds_first_writer_wins(): - # Cross-consumer isolation: the FIRST consumer to meter a sid owns it. A - # second consumer reusing the same opaque sid must NOT steal or overwrite the - # ownership — otherwise B could hijack A's session economics / warm peers. - route_session_meter.reset() - assert route_session_meter.owner("sidA") is None # unknown -> None - route_session_meter.observe("sidA", owner="keyA", tokens_in=10, cost_usd=0.001) - assert route_session_meter.owner("sidA") == "keyA" - # B reuses the same sid: data folds in (it's the same session key) but the - # owner is sticky. - route_session_meter.observe("sidA", owner="keyB") - assert route_session_meter.owner("sidA") == "keyA" - # observe without an owner never clears or changes the binding either. - route_session_meter.observe("sidA", tokens_in=5) - assert route_session_meter.owner("sidA") == "keyA" - # no session / no owner -> None, never raises. - assert route_session_meter.owner(None) is None - route_session_meter.reset() - assert route_session_meter.owner("sidA") is None # reset clears owners +def test_session_owner_binds_first_writer_wins(host_store_clean): + # Cross-consumer isolation: the FIRST consumer to use a sid owns it (#4b: the + # owner is the caller of the session's EARLIEST call, derived from `calls`). A + # second consumer reusing the same opaque sid must NOT steal ownership. + from conftest import seed_call + import host_store + assert host_store.session_owner("sidA") is None # unknown -> None + seed_call(session="sidA", caller="keyA", tokens_in=10, cost_usd=0.001, ts=100) + assert host_store.session_owner("sidA") == "keyA" + # B reuses the same sid later: data is the same session, but the owner is the + # first writer (earliest call). + seed_call(session="sidA", caller="keyB", ts=200) + assert host_store.session_owner("sidA") == "keyA" + assert host_store.session_owner(None) is None # no session -> None, never raises diff --git a/tests/test_route_cache.py b/tests/test_route_cache.py index 7cee52a..44b767a 100644 --- a/tests/test_route_cache.py +++ b/tests/test_route_cache.py @@ -1,26 +1,18 @@ -"""Essence for the per-session cache-affinity substrate (PR1). +"""Essence for per-session cache affinity (the cache_hot field). Caching is STATE between calls: a provider discounts the reused prefix only if -the SAME peer serves the session again. That memory cannot live in the algebra -(a policy is a pure function of one call) nor in offers_sync (request-blind, -shared across sessions) — it is per-session host state, like route_latency is -per-route host state. This closes three things: - - 1. route_cache folds session -> last successful route (success only, like - route_latency); unknown session / no session -> no hot route. - 2. the central fold hook (_resolve_call_async -> _fold_route_outcome) records - it, with the session threaded host-side from the contract. - 3. the cache_hot field is true for EXACTLY the candidate whose route is the - session's hot route, so a policy scoring it keeps the hot peer sticky. - -Zero engine change: cache_hot is a host-declared extension field (fields.lua -schema{extensions} seam) whose getter reconstructs the route key from candidate -identity exactly as _fold_route_outcome does, and compares it to the hot route -the host resolved into ctx.request.cache_hot_route per request. +the SAME route serves the session again. The host remembers which route last +served a session and steers the next turn back to it. Since #4b that memory is +DERIVED on the fly from the `calls` ledger (host_store.hot_route) — the route of +the session's most recent SUCCESSFUL call — instead of an in-process fold. + +The algebra never sees the route key: per request the host resolves +hot_route(session) into ctx.request.cache_hot_route, and the cache_hot field +getter reconstructs each candidate's route key the same way and compares, +exposing only a Bool. """ from __future__ import annotations -import asyncio import sys from pathlib import Path @@ -29,55 +21,48 @@ ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) -import route_cache as rc # noqa: E402 -import route_reliability as rr # noqa: E402 +import host_store # noqa: E402 +import route_reliability as rr # noqa: E402 from llm_router_host import LLMRouterHost # noqa: E402 +from conftest import seed_call # noqa: E402 # -------------------------------------------------------------------------- -# 1. route_cache module — session -> hot route, success only +# 1. hot_route — derived from the session's most recent SUCCESSFUL call # -------------------------------------------------------------------------- -def test_hot_route_unknown_session_is_none(): - rc.reset() - assert rc.hot_route("s1") is None - assert rc.hot_route(None) is None +def test_hot_route_unknown_session_is_none(host_store_clean): + assert host_store.hot_route("s1") is None + assert host_store.hot_route(None) is None -def test_successful_call_makes_route_hot(): - rc.reset() - k = rr.route_key("antseed", "glm-5.2", "peerX") - rc.observe("s1", k, ok=True) - assert rc.hot_route("s1") == k +def test_successful_call_makes_route_hot(host_store_clean): + seed_call(session="s1", provider="antseed", family="glm-5.2", served_by="peerX") + assert host_store.hot_route("s1") == rr.route_key("antseed", "glm-5.2", "peerX") -def test_failed_call_does_not_set_affinity(): - rc.reset() - k = rr.route_key("antseed", "glm-5.2", "peerX") - rc.observe("s1", k, ok=False) - # only successful calls fold, exactly like route_latency: a failure carries - # no honest "this peer holds the prefix" signal. - assert rc.hot_route("s1") is None +def test_failed_call_does_not_set_affinity(host_store_clean): + # only successful calls (status < 400) hold the prefix; a failure carries no + # honest "this route holds the prefix" signal. + seed_call(session="s1", provider="antseed", family="glm-5.2", + served_by="peerX", status=502) + assert host_store.hot_route("s1") is None -def test_latest_successful_route_wins(): - rc.reset() - a = rr.route_key("antseed", "glm-5.2", "peerA") - b = rr.route_key("openrouter", "minimax-m2.7", "openrouter") - rc.observe("s1", a, ok=True) - rc.observe("s1", b, ok=True) - assert rc.hot_route("s1") == b # the most recent successful peer holds the prefix +def test_latest_successful_route_wins(host_store_clean): + seed_call(session="s1", provider="antseed", family="glm-5.2", served_by="peerA", ts=100) + seed_call(session="s1", provider="openrouter", family="minimax-m2.7", + served_by="openrouter", ts=200) + assert host_store.hot_route("s1") == rr.route_key("openrouter", "minimax-m2.7", "openrouter") -def test_sessions_are_independent(): - rc.reset() - a = rr.route_key("antseed", "glm-5.2", "peerA") - rc.observe("s1", a, ok=True) - assert rc.hot_route("s2") is None +def test_sessions_are_independent(host_store_clean): + seed_call(session="s1", provider="antseed", family="glm-5.2", served_by="peerA") + assert host_store.hot_route("s2") is None # -------------------------------------------------------------------------- -# 2. central fold hook records the session's hot route +# 2. cache_hot field — true for exactly the session's hot route # -------------------------------------------------------------------------- @pytest.fixture @@ -92,46 +77,6 @@ def host(): return h -def test_central_fold_records_session_hot_route(host): - rc.reset() - req = {"provider_id": "antseed", "model_family": "glm-5.2", - "offer": {"model_family": "glm-5.2", "peer_id": "peerX"}} - - async def override(_r): - return {"ok": True, "latency_ms": 5, "response": {"tool_calls": None}} - - asyncio.run(host._resolve_call_async(req, call_override=override, session="s1")) - assert rc.hot_route("s1") == rr.route_key("antseed", "glm-5.2", "peerX") - - -def test_fold_without_session_is_noop(host): - rc.reset() - req = {"provider_id": "antseed", "model_family": "glm-5.2", - "offer": {"model_family": "glm-5.2", "peer_id": "peerX"}} - - async def override(_r): - return {"ok": True, "latency_ms": 5, "response": {}} - - asyncio.run(host._resolve_call_async(req, call_override=override)) - assert rc.snapshot() == {} # no session -> nothing recorded - - -def test_failed_call_through_fold_sets_no_affinity(host): - rc.reset() - req = {"provider_id": "antseed", "model_family": "glm-5.2", - "offer": {"model_family": "glm-5.2", "peer_id": "peerX"}} - - async def override(_r): - return {"ok": False, "error": "boom", "response": {}} - - asyncio.run(host._resolve_call_async(req, call_override=override, session="s1")) - assert rc.hot_route("s1") is None - - -# -------------------------------------------------------------------------- -# 3. cache_hot field — true for exactly the session's hot route -# -------------------------------------------------------------------------- - def test_cache_hot_marks_only_the_session_hot_route(host): # Two marketplace-style candidates, same family, different peers. The host # resolved this session's hot route into the contract; the cache_hot field @@ -153,7 +98,6 @@ def test_cache_hot_marks_only_the_session_hot_route(host): "profile": "default", "policy_ir": policy, "cache_hot_route": hot, - # isolate the pool to our two peers (the static catalog has no glm-5.2) "requirements": {"model_family": "glm-5.2"}, "extra_candidates": [cold_cand, hot_cand], }) @@ -176,10 +120,8 @@ def test_cache_hot_false_when_no_hot_route_in_contract(host): ["always", {"action": "next_candidate"}]] ranked, _ = host.rank({ "prompt": "x", "profile": "default", "policy_ir": policy, - "requirements": {"model_family": "glm-5.2"}, # isolate to our two peers + "requirements": {"model_family": "glm-5.2"}, "extra_candidates": [a, b], }) assert ranked - # cache_hot is false for everyone (no hot route in the contract): both peers - # survive, the affinity scorer adds nothing, no phantom stickiness. assert {r["candidate"]["served_model_id"] for r in ranked} == {"A", "B"} diff --git a/tests/test_route_latency.py b/tests/test_route_latency.py deleted file mode 100644 index fafbfba..0000000 --- a/tests/test_route_latency.py +++ /dev/null @@ -1,70 +0,0 @@ -"""Host-side per-route latency fold (twin of route_reliability): an EMA of -observed end-to-end call latency, folded only on SUCCESS, stamped onto offers as -offer.latency_ms so a policy can route by speed.""" -from __future__ import annotations - -import sys -from pathlib import Path - -ROOT = Path(__file__).resolve().parents[1] -sys.path.insert(0, str(ROOT)) - -import route_latency as rl # noqa: E402 - -K = rl.route_key("antseed", "glm-5.2", "peerSlow") - - -def test_unobserved_route_is_none(): - rl.reset() - assert rl.latency_ms(K) is None - - -def test_first_observation_seeds_directly(): - rl.reset() - rl.observe(K, 12000, ok=True) - assert rl.latency_ms(K) == 12000 - - -def test_ema_decays_toward_new_samples(): - rl.reset() - rl.observe(K, 12000, ok=True) # seed - rl.observe(K, 2000, ok=True) # a fast call pulls it down - v = rl.latency_ms(K) - assert 2000 < v < 12000 # between, nearer the slow seed (alpha 0.3) - assert v == round(0.3 * 2000 + 0.7 * 12000) - - -def test_failures_do_not_fold(): - # A fast failure (empty content in 1.9s) must NOT make a broken route look - # fast — that is reliability's job, not latency's. - rl.reset() - rl.observe(K, 1900, ok=False) - assert rl.latency_ms(K) is None - rl.observe(K, 12000, ok=True) # only the honest success counts - rl.observe(K, 1900, ok=False) # later failure ignored - assert rl.latency_ms(K) == 12000 - - -def test_nonpositive_or_missing_latency_ignored(): - rl.reset() - rl.observe(K, None, ok=True) - rl.observe(K, 0, ok=True) - rl.observe(K, -5, ok=True) - assert rl.latency_ms(K) is None - - -def test_routes_are_independent(): - rl.reset() - fast = rl.route_key("openai", "gpt-5.5", "openai") - rl.observe(K, 12000, ok=True) - rl.observe(fast, 900, ok=True) - assert rl.latency_ms(K) == 12000 - assert rl.latency_ms(fast) == 900 # a fast partner stays distinct from a slow peer - - -def test_snapshot_and_reset(): - rl.reset() - rl.observe(K, 12000, ok=True) - assert rl.snapshot() == {K: 12000} - rl.reset() - assert rl.snapshot() == {} diff --git a/tests/test_route_reliability.py b/tests/test_route_reliability.py deleted file mode 100644 index fceb834..0000000 --- a/tests/test_route_reliability.py +++ /dev/null @@ -1,53 +0,0 @@ -""" -Host-side per-route reliability fold: a success-rate EMA the host updates on each -call outcome and stamps onto offers. Proves the EMA math and that the async -backend folds the outcome under the route's (provider, family, peer) key. -""" -from __future__ import annotations - -import sys -from pathlib import Path - -import pytest - -ROOT = Path(__file__).resolve().parents[1] -sys.path.insert(0, str(ROOT)) - -import llm_router_host as H # noqa: E402 -import route_reliability as rr # noqa: E402 - - -def test_ema_seeds_then_decays(): - rr.reset() - k = rr.route_key("antseed", "m", "peerX") - assert rr.success_rate(k) is None - rr.observe(k, False) # first obs seeds directly - assert rr.success_rate(k) == 0.0 - rr.observe(k, True) # 0.2*1 + 0.8*0 - assert rr.success_rate(k) == pytest.approx(0.2) - rr.observe(k, True) # 0.2*1 + 0.8*0.2 - assert rr.success_rate(k) == pytest.approx(0.36) - - -def _req(peer_id, family="m"): - return { - "api_kind": "openai_compatible", "base_url": "http://s/v1", - "served_model_id": "m", "provider_id": "antseed", - "messages": [{"role": "user", "content": "hi"}], - "offer": {"peer_id": peer_id, "model_family": family, "wire_model_id": "m"}, - } - - -# 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._fold_route_outcome(_req("peerGood"), {"ok": True, "latency_ms": 10}) - assert rr.success_rate(rr.route_key("antseed", "m", "peerGood")) == 1.0 - - -def test_fold_route_outcome_on_failure(): - rr.reset() - H._fold_route_outcome(_req("peerBad"), {"ok": False, "error_kind": "bad_response"}) - assert rr.success_rate(rr.route_key("antseed", "m", "peerBad")) == 0.0 diff --git a/tests/test_route_tool_capability.py b/tests/test_route_tool_capability.py index 891beca..5c1c5ed 100644 --- a/tests/test_route_tool_capability.py +++ b/tests/test_route_tool_capability.py @@ -1,57 +1,55 @@ -"""Learned per-route tool capability: a route observed to ignore tools (a +"""Learned per-route tool capability (#4c): a route observed to ignore tools (a tools-request that never returns tool_calls) is marked incapable so offers_sync stops stamping supports_tools for it. Optimistic by default; any real tool_call -proves capability; an incapable mark expires so a false positive is re-tested. +clears it; the mark is windowed, so a false positive ages out and is re-tested. +Derived on the fly from route_observations (host_store.tool_incapable_routes). """ from __future__ import annotations import sys +import time from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) -import route_tool_capability as tc # noqa: E402 +import host_store # noqa: E402 +from conftest import seed_route_obs # noqa: E402 K = "antseed|glm-5.2|peerX" +N = 20 # _MIN_SAMPLES -def test_unobserved_route_is_capable(): - tc.reset() - assert tc.is_capable(K) is True +def _obs(n, tool_calls_emitted=False, ts=None): + seed_route_obs("antseed", "glm-5.2", "peerX", ok=True, n=n, ts=ts, + tools_requested=True, tool_calls_emitted=tool_calls_emitted) -def test_non_tool_requests_carry_no_signal(): - tc.reset() - for _ in range(50): - tc.observe(K, tools_requested=False, tool_calls_emitted=False) - assert tc.is_capable(K) is True - assert tc.snapshot() == {} # nothing recorded for non-tool requests +def test_unobserved_route_is_capable(host_store_clean): + assert K not in host_store.tool_incapable_routes() -def test_route_that_never_emits_tool_calls_is_marked_incapable(): - tc.reset() - for _ in range(tc._MIN_SAMPLES - 1): - tc.observe(K, True, False) - assert tc.is_capable(K) is True # not enough evidence yet - tc.observe(K, True, False) # crosses _MIN_SAMPLES with zero tool_calls - assert tc.is_capable(K) is False +def test_non_tool_requests_carry_no_signal(host_store_clean): + seed_route_obs("antseed", "glm-5.2", "peerX", ok=True, n=50) # tools_requested=False + assert K not in host_store.tool_incapable_routes() -def test_any_tool_call_proves_capability_permanently(): - tc.reset() - tc.observe(K, True, True) # one real tool_call - for _ in range(tc._MIN_SAMPLES * 2): - tc.observe(K, True, False) # then many without - assert tc.is_capable(K) is True # proven capable, never demoted +def test_route_that_never_emits_tool_calls_is_marked_incapable(host_store_clean): + _obs(N - 1) + assert K not in host_store.tool_incapable_routes() # not enough evidence yet + _obs(1) # crosses _MIN_SAMPLES, zero tool_calls + assert K in host_store.tool_incapable_routes() -def test_incapable_mark_expires_and_route_is_retested(monkeypatch): - tc.reset() - now = [1_000_000] - monkeypatch.setattr(tc, "_now_ms", lambda: now[0]) - for _ in range(tc._MIN_SAMPLES): - tc.observe(K, True, False) - assert tc.is_capable(K) is False - now[0] += tc._EXPIRY_MS # advance past the expiry window - assert tc.is_capable(K) is True # re-tested: window reset +def test_any_tool_call_clears_incapability(host_store_clean): + _obs(1, tool_calls_emitted=True) # one real tool_call + _obs(N * 2) # then many without + assert K not in host_store.tool_incapable_routes() # capable + + +def test_old_observations_age_out_of_the_window(host_store_clean): + now = int(time.time() * 1000) + _obs(N, ts=now - 31 * 60 * 1000) # 31 min ago, past the 30-min window + assert K not in host_store.tool_incapable_routes() # aged out -> re-tested + _obs(N, ts=now) # fresh evidence -> incapable + assert K in host_store.tool_incapable_routes() diff --git a/tests/test_shim.py b/tests/test_shim.py index 1d19eb7..196ea82 100644 --- a/tests/test_shim.py +++ b/tests/test_shim.py @@ -452,14 +452,16 @@ def test_x_market_attaches_model_meta_and_book_last_seen(client, host): src.SOURCE_STATE.clear() -def test_x_market_attaches_live_perf_after_calls(client, host): +def test_x_market_attaches_live_perf_after_calls(client, host, host_store_clean): import sources as src + import host_store src.SOURCE_STATE.clear() for prov, fam in _all_pairs(host): host.set_mock_response(prov, fam, _ok_response("ok")) resp = client.post("/v1/chat/completions", json={ "model": "", "messages": [{"role": "user", "content": "hi"}]}) provider = resp.json()["x_router"]["provider"] + host_store._write_q.join() # the call's route observation is async body = client.get("/x/market").json() perfs = [row["perf"] for f in body["families"] for row in f["rows"] if row["seller"] == provider and row["perf"]] @@ -504,23 +506,16 @@ def test_x_router_carries_executed_cost(client, host): assert xr["cost_usd"] is not None # example metrics price every pair -def test_session_view_scoped_to_owning_consumer(client, host): - """Cross-consumer isolation at the endpoint the consumer-facing /v1/session - proxies to: a chat carrying the proxy-injected x-llm-router-caller binds the - sid to that consumer; only that consumer (forwarded as the same header) may - read the session — anyone else gets 404 (not 403, which would confirm the sid - exists). An operator caller (no header) stays unscoped.""" - import route_session_meter - route_session_meter.reset() - for prov, fam in _all_pairs(host): - host.set_mock_response(prov, fam, _ok_response("ok")) - # consumer A meters sidA (the proxy injects the authed caller as this header). - r = client.post("/v1/chat/completions", - headers={"x-llm-router-caller": "consumerA"}, - json={"model": "", "session": "sidA", - "messages": [{"role": "user", "content": "hi"}]}) - assert r.status_code == 200 - assert route_session_meter.owner("sidA") == "consumerA" +def test_session_view_scoped_to_owning_consumer(client, host, host_store_clean): + """Cross-consumer isolation at the consumer-facing /x/session view. The owner + is the caller of the session's earliest call (#4b: derived from the `calls` + ledger the ingress writes); only that consumer (forwarded as + x-llm-router-caller) may read it — anyone else gets 404 (not 403, which would + confirm the sid exists). An operator caller (no header) stays unscoped.""" + from conftest import seed_call + # the ingress records the session's call with the authed caller. + seed_call(session="sidA", caller="consumerA", provider="p", family="f", + served_by="s", tokens_in=5, status=200) # A reads its own session -> 200 with the accumulated economics. r = client.get("/x/session/sidA", headers={"x-llm-router-caller": "consumerA"}) @@ -540,4 +535,3 @@ def test_session_view_scoped_to_owning_consumer(client, host): r = client.get("/x/session/sidA") assert r.status_code == 200 assert r.json()["calls"] == 1 - route_session_meter.reset()