feat(cache): per-session cache affinity — cache_hot + route_cache + session

features/10_agent_cache.feature

@@ -0,0 +1,28 @@
+Feature: Agent cache affinity — cache_hot session stickiness
+  An agent loop re-sends a large, stable prefix every turn. To make the
+  provider's prompt-cache discount actually land, the router must keep the
+  conversation pinned to the peer that already holds that prefix hot. This proves
+  the session -> route_cache -> cache_hot pipeline end to end over /v1: a turn
+  carrying a session teaches the router which peer served it, and the next turn's
+  ranking gives that peer a decisive affinity bonus. Provider-health independent:
+  the assertions read the RANKING from x_router.decision_trace, which is present
+  even when execution later exhausts.
+
+  Background:
+    Given the stack is healthy
+    And I have a caller token
+
+  @p0 @api @agent @cache
+  Scenario: An agent's session keeps its working route cache-hot across turns
+    When an agent establishes session "agent-cache-1" with a free turn
+    Then the agent's turn routed to a concrete peer
+    When the agent re-ranks its turn with the same session as "hot"
+    And the agent re-ranks the same turn with no session as "cold"
+    Then the agent's route scores higher in "hot" than in "cold"
+
+  @p1 @api @agent @cache
+  Scenario: A brand-new session gets no phantom affinity
+    When an agent establishes session "agent-cache-2" with a free turn
+    And the agent re-ranks the same turn with unknown session "never-seen-zzz" as "fresh"
+    And the agent re-ranks the same turn with no session as "none"
+    Then the rankings "fresh" and "none" are identical

features/steps/steps.py

@@ -214,3 +214,126 @@ def step_every_has(context, path, key):
 @then('the response text contains "{sub}"')
 def step_text_contains(context, sub):
     assert sub in context.resp_text, f'response text missing {sub!r}'
+
+
+# ---- agent cache-affinity steps (10_agent_cache.feature) -------------------
+
+# A simple, stable family the agent loop runs on (codex gpt-5.3-codex-spark,
+# a $0 subscription route). Kept fixed and boring on purpose — the test is about
+# cache stickiness, not model choice.
+AGENT_FAMILY = "gpt-5.3-codex-spark"
+
+# Seed policy: cheapest within the agent's family. Routes to the $0 codex peer
+# when healthy; the override failplan rolls to the next peer of the same family
+# on a transient flake, so the seeding turn can SUCCEED and fold route_cache.
+_SEED_POLICY = [
+    "policy",
+    ["and", ["meets_req"], ["not", ["is", "disabled"]], ["family_eq", AGENT_FAMILY]],
+    ["neg", ["normalize", ["field", "price_in"]]],
+    ["argmax"], ["id"],
+    ["override", ["always", {"action": "next_candidate"}],
+                 "provider_down", {"action": "next_candidate"}],
+]
+
+
+def _cache_aware_policy(family):
+    # cheapest WITHIN the agent's family + a decisive cache_hot affinity bonus.
+    # family_eq (a filter predicate) keeps the scorer running on the survivors,
+    # so each candidate's row carries a real score (unlike a requirements pin).
+    return ["policy",
+            ["and", ["meets_req"], ["not", ["is", "disabled"]],
+                    ["family_eq", family]],
+            ["add", ["neg", ["normalize", ["field", "price_in"]]],
+                    ["scale", 10, ["gate", ["is", "cache_hot"], ["lit", 1]]]],
+            ["argmax"], ["id"], ["always", {"action": "next_candidate"}]]
+
+
+@when('an agent establishes session "{sid}" with a free turn')
+def step_agent_seed(context, sid):
+    # A real agent turn on the session, on the fixed AGENT_FAMILY. Retry a
+    # transient provider flake (a 429) until the call SUCCEEDS — that success is
+    # what folds the session's hot route into route_cache.
+    body = {"model": "", "max_tokens": 8, "session": sid,
+            "messages": [{"role": "user", "content": "hi"}],
+            "policy_ir": _SEED_POLICY}
+    chosen = None
+    for _ in range(6):
+        _do(context, "POST", "/v1/chat/completions", auth="consumer", body=body)
+        xr = (context.json or {}).get("x_router") or {}
+        if context.resp.status_code == 200 and xr.get("served_model_id"):
+            chosen = xr
+            break
+    assert chosen, (f"the {AGENT_FAMILY} route did not succeed to seed the session "
+                    f"in 6 tries; last status {context.resp.status_code}: "
+                    f"{context.resp_text[:200]}")
+    context.agent = {"sid": sid, "provider": chosen.get("provider"),
+                     "family": AGENT_FAMILY,
+                     "served": chosen.get("served_model_id")}
+    context.ranks = {}
+
+
+@then('the agent\'s turn routed to a concrete peer')
+def step_agent_routed(context):
+    a = getattr(context, "agent", None)
+    assert a and a.get("served") and a.get("family"), f"no route captured: {a}"
+
+
+def _rerank(context, sid):
+    # Re-issue the agent's turn with a cache-aware policy pinned to its family.
+    # We read the RANKING from x_router.decision_trace (present even if execution
+    # later exhausts), so the assertion is independent of provider health.
+    body = {"model": "", "max_tokens": 8,
+            "messages": [{"role": "user", "content": "hi"}],
+            "policy_ir": _cache_aware_policy(context.agent["family"])}
+    if sid is not None:
+        body["session"] = sid
+    _do(context, "POST", "/v1/chat/completions", auth="consumer", body=body)
+    tr = jpath(context.json or {}, "x_router.decision_trace")
+    ranked = tr.get("ranked") if isinstance(tr, dict) else None
+    assert ranked, f"no ranked candidates in trace (status {context.resp.status_code}): {context.resp_text[:200]}"
+    return ranked
+
+
+@when('the agent re-ranks its turn with the same session as "{label}"')
+def step_rerank_session(context, label):
+    context.ranks[label] = _rerank(context, context.agent["sid"])
+
+
+@when('the agent re-ranks the same turn with no session as "{label}"')
+def step_rerank_nosession(context, label):
+    context.ranks[label] = _rerank(context, None)
+
+
+@when('the agent re-ranks the same turn with unknown session "{sid}" as "{label}"')
+def step_rerank_unknown(context, sid, label):
+    context.ranks[label] = _rerank(context, sid)
+
+
+def _max_score(ranked):
+    # The /v1 decision-trace rows carry model_family + score (the candidate's
+    # provider/served_model_id are not echoed there), and family_eq has isolated
+    # the agent's family — so the family's top score is the right discriminator:
+    # without affinity every row is a price score in [0,1]; the cache_hot bonus
+    # lifts the session's route well above that, so only a real bonus makes the
+    # "hot" top exceed the "cold" top.
+    scores = [r.get("score") for r in ranked if r.get("score") is not None]
+    return max(scores) if scores else None
+
+
+@then('the agent\'s route scores higher in "{hot}" than in "{cold}"')
+def step_scores_higher(context, hot, cold):
+    sh = _max_score(context.ranks[hot])
+    sc = _max_score(context.ranks[cold])
+    assert sh is not None and sc is not None, \
+        f"missing scores: {hot}={sh} {cold}={sc}"
+    assert sh > sc, (f"cache affinity did not lift the agent's route: "
+                     f"top score {hot}={sh} is not > {cold}={sc} "
+                     f"(family {context.agent['family']!r})")
+
+
+@then('the rankings "{a}" and "{b}" are identical')
+def step_rankings_identical(context, a, b):
+    def key(rows):
+        return [(r.get("served_model_id"), r.get("score")) for r in rows]
+    ka, kb = key(context.ranks[a]), key(context.ranks[b])
+    assert ka == kb, f"rankings differ:\n  {a}={ka}\n  {b}={kb}"

llm_router_host.py

@@ -24,6 +24,7 @@
 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 lupa
 from lupa import LuaRuntime
@@ -89,6 +90,46 @@ def __init__(
         self.router = self._dofile(Path(router_path))
         self.config = self._dofile(Path(config_path))
         self.metrics = self._dofile(Path(metrics_path)) if metrics_path else None
+        self._inject_host_fields()
+
+    def _inject_host_fields(self) -> None:
+        """Declare host-universal observation fields that every catalog gets for
+        free because they denote a HOST measurement, not catalog data — currently
+        the per-session cache-affinity Bool `cache_hot`. Done here (once, after the
+        config loads, before router.init and the flow schema read cfg.fields) so no
+        catalog .lua repeats the getter and the field exists for example/live/any
+        config alike.
+
+        Zero engine change: this is the fields.lua extension seam
+        (`schema{ extensions }`). The getter builds each candidate's route key
+        through the SAME `route_reliability.route_key` that `_fold_route_outcome`
+        uses — bridged into Lua as the `host_route_key` global, so the
+        serialization (`provider|family|peer`, peer falling back to provider for
+        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
+        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
+        # lost. Bridge the host's route_key into Lua instead of re-serializing.
+        self.lua.globals().host_route_key = _route_reliability.route_key
+        self.lua.eval("""
+function(cfg)
+    cfg.fields = cfg.fields or {}
+    cfg.fields.cache_hot = {
+        sort = "Bool", default = false, group = "route",
+        get = function(c, ctx)
+            local hot = ctx and ctx.request and ctx.request.cache_hot_route
+            if hot == nil then return false end
+            local pid, fam = c.provider_id, c.model_family
+            if pid == nil or fam == nil then return false end
+            local peer = (c.offer and c.offer.peer_id) or pid
+            return host_route_key(pid, fam, peer) == hot
+        end,
+    }
+end
+""")(self.config)
 
     # ---- public API -----------------------------------------------------
 
@@ -195,7 +236,8 @@ async def execute_flow_async(self, flow_ir, base_contract,
         input_text = _last_user_text(base_contract.get("messages") or [])
         carry = {k: base_contract[k] for k in
                  ("max_tokens", "tools", "tool_choice", "response_format",
-                  "temperature", "seed") if k in base_contract}
+                  "temperature", "seed", "session", "cache_hot_route")
+                 if k in base_contract}
 
         async def run_node(nid, node, prompt):
             # Give the node the FULL conversation (system, history, tool results)
@@ -350,6 +392,10 @@ async def execute_async(self, contract: dict, call_override=None) -> dict:
         (the streaming path uses it to thread a per-request delta channel);
         mock responses still take precedence per (provider, family) pair.
         """
+        # Session id (if the caller named one) rides host-side from here to the
+        # fold so route_cache learns which peer served this conversation. It is a
+        # local of this coroutine, so concurrent executes never share it.
+        session = contract.get("session")
         step = self.router.execute_step(None, _to_lua(self.lua, contract), None)
         while True:
             status = step["status"]
@@ -359,7 +405,7 @@ async def execute_async(self, contract: dict, call_override=None) -> dict:
             handle = step["state_handle"]
             if status == "call":
                 req = _to_py(step["request"]) or {}
-                resp = await self._resolve_call_async(req, call_override)
+                resp = await self._resolve_call_async(req, call_override, session=session)
                 step = self.router.execute_step(handle, None, _to_lua(self.lua, resp))
             elif status == "wait":
                 until_ms = step["until_ms"] or 0
@@ -370,7 +416,8 @@ async def execute_async(self, contract: dict, call_override=None) -> dict:
             else:
                 return {"ok": False, "error": f"internal: bad step status {status}", "trace": {}}
 
-    async def _resolve_call_async(self, request: dict, call_override=None) -> dict:
+    async def _resolve_call_async(self, request: dict, call_override=None,
+                                  session: "str | None" = None) -> dict:
         """Resolve one provider call for the async driver: mock first (so the
         same set_mock_response works for sync and async), then a per-run
         override, then the async hook, then the sync hook as a last resort."""
@@ -388,7 +435,7 @@ async def _resolve_call_async(self, request: dict, call_override=None) -> dict:
         # 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)
+        _fold_route_outcome(request, result, session=session)
         return result
 
     def dump_state(self) -> dict:
@@ -750,7 +797,8 @@ def _peer_gate(peer_id: str, cap: int) -> asyncio.Semaphore:
     return sem
 
 
-def _fold_route_outcome(request: dict, result: dict) -> None:
+def _fold_route_outcome(request: dict, result: dict,
+                        session: "str | None" = None) -> None:
     """Fold ONE call outcome into the host-side per-route measurements:
     reliability (success EMA), latency (EMA), and learned tool capability. Called
     from _resolve_call_async for BOTH the direct hook and the streaming/override
@@ -780,6 +828,11 @@ def _fold_route_outcome(request: dict, result: dict) -> None:
     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)
     # 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_cache.py

@@ -0,0 +1,67 @@
+"""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()

shim.py

@@ -36,6 +36,8 @@
 from fastapi.responses import JSONResponse, StreamingResponse
 from pydantic import BaseModel, ConfigDict
 
+import route_cache
+
 
 # Profile name used when nothing else can be inferred. Replaced via
 # create_app(default_profile=...).
@@ -79,6 +81,12 @@ class ChatRequest(BaseModel):
     # policy). When present it takes precedence over policy_ir/model. Admission
     # failure -> 400 invalid_flow.
     flow_ir: list | None = None
+    # Conversation/session id (optional). When present the host learns which peer
+    # served this session (route_cache) and, next turn, marks that peer's offer
+    # cache_hot so a cache-aware policy keeps the prompt-cache-hot peer sticky.
+    # Pure host state — never enters the algebra's signature; clients without a
+    # session simply get no affinity. Additive to OpenAI-compat.
+    session: str | None = None
 
 
 class PolicyRankRequest(BaseModel):
@@ -941,6 +949,17 @@ def _request_to_contract(
         # one place); it only translates the core's refusal to a 400.
         contract["policy_ir"] = req.policy_ir
 
+    if req.session:
+        # The session rides into the contract so the fold can attribute the call
+        # to it; and the host resolves the session's hot route NOW (snapshot,
+        # before eval) into cache_hot_route, which the cache_hot field getter
+        # 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)
+        if hot is not None:
+            contract["cache_hot_route"] = hot
+
     return contract