Caching & Cost Optimization at Scale

{"tenant": "acme", "route": "ask", "model": "gpt-6-luna", "input_tokens": 2400, "cached_tokens": 1100, "output_tokens": 300}
{"tenant": "acme", "route": "ask", "model": "gpt-6-luna", "input_tokens": 2350, "cached_tokens": 1100, "output_tokens": 280}
{"tenant": "zephyr", "route": "investigate", "model": "gpt-6.1-terra", "input_tokens": 9000, "cached_tokens": 1100, "output_tokens": 1200}

# script_id: day_085_caching_cost_scale/first_semantic_cache
# A semantic cache in ~20 lines. Install: uv add "openai>=3.46,<4" numpy
# Env: OPENAI_API_KEY, ATLAS_EMBED_MODEL (default text-embedding-3-small). Cost: < $0.0001
import os

import numpy as np
from openai import OpenAI

EMBED_MODEL = os.environ.get("ATLAS_EMBED_MODEL", "text-embedding-3-small")
client = OpenAI()


def openai_embed(texts: list[str]) -> np.ndarray:
    resp = client.embeddings.create(model=EMBED_MODEL, input=texts)
    return np.array([d.embedding for d in resp.data], dtype=np.float32)   # already unit length


if __name__ == "__main__":
    cache: list[tuple[np.ndarray, str]] = []                 # (question vector, answer)
    for q in ["How do I fix ORB-1101 errors?", "What's the fix for ORB-1101 errors?",
              "How do I fix ORB-1102 errors?"]:
        v = openai_embed([q])[0]
        sims = [float(vec @ v) for vec, _ in cache]
        if sims and max(sims) >= 0.90:
            print(f"{q:38} hit {max(sims):.2f} -> {cache[sims.index(max(sims))][1]}")
        else:
            cache.append((v, f"answer to {q!r}"))           # a real app calls the model here
            print(f"{q:38} miss (answered and stored)")

How do I fix ORB-1101 errors?          miss (answered and stored)
What's the fix for ORB-1101 errors?    hit 0.xx -> answer to 'How do I fix ORB-1101 errors?'
How do I fix ORB-1102 errors?          hit 0.xx -> answer to 'How do I fix ORB-1101 errors?'

# script_id: day_085_caching_cost_scale/atlas_cache
# atlas/cache.py: a tenant-scoped semantic cache with an exact-term guard.
# Install: uv add numpy pydantic
# Env: none. Runs offline with hashed_embed (a stand-in); pass a real embedder in production.
from __future__ import annotations

import hashlib
import re
import time
from dataclasses import dataclass
from typing import Callable, Literal

import numpy as np
from pydantic import BaseModel, Field

Embedder = Callable[[list[str]], np.ndarray]        # texts -> (n, dim) unit vectors


class Citation(BaseModel):                          # compact copies of Day 15 / Day 28
    doc_id: str
    quote: str


class Answer(BaseModel):
    text: str
    citations: list[Citation] = Field(default_factory=list)


@dataclass
class CacheHit:
    answer: Answer
    similarity: float
    source_query: str
    cached_at: float


@dataclass
class _Entry:
    scope: str
    query: str
    terms: frozenset[str]
    vec: np.ndarray
    answer: Answer
    cached_at: float

# script_id: day_085_caching_cost_scale/atlas_cache
TERM_RE = re.compile(
    r"ORB-\d{4}"                          # Orbit error codes
    r"|\b\d+\.(?:\d+|x)(?:\.\d+)?\b"      # versions: 2.x, 3.2.1
    r"|\b[a-z]+(?:_[a-z]+)+\b",           # config keys: lease_seconds, ORBIT_LEASE
    re.IGNORECASE)


def key_terms(text: str) -> frozenset[str]:
    """The exact tokens that must match before two questions may share an answer."""
    return frozenset(t.lower() for t in TERM_RE.findall(text))


def detect_version(question: str) -> str:          # compact copy of Day 44
    m = re.search(r"\b([23])\.(?:\d+|x)\b", question)
    return f"{m.group(1)}.x" if m else "3.x"       # Orbit 3.2.1 is current


def hashed_embed(texts: list[str], dim: int = 512) -> np.ndarray:
    """Offline stand-in for an embedding model: hashed words + character trigrams.
    It measures spelling overlap, not meaning. Use it for demos and tests only."""
    out = np.zeros((len(texts), dim), dtype=np.float32)
    for i, text in enumerate(texts):
        t = text.lower()
        feats = re.findall(r"[a-z0-9_.-]+", t) + [t[j:j + 3] for j in range(len(t) - 2)]
        for f in feats:
            h = hashlib.blake2b(f.encode(), digest_size=4).digest()
            out[i, int.from_bytes(h, "little") % dim] += 1.0
    return out / np.maximum(np.linalg.norm(out, axis=1, keepdims=True), 1e-9)

# script_id: day_085_caching_cost_scale/atlas_cache
class SemanticCache:
    """In-memory cache-aside store for one process, tests and demos (Redis: see Go Deeper)."""

    def __init__(self, embed: Embedder, *, threshold: float = 0.95, ttl_s: int = 86_400,
                 terms: Callable[[str], frozenset[str]] | None = key_terms) -> None:
        self.embed, self.threshold, self.ttl_s, self.terms = embed, threshold, ttl_s, terms
        self.clock: Callable[[], float] = time.time       # swap in tests to fast-forward
        self._entries: list[_Entry] = []

    def lookup(self, scope: str, query: str) -> CacheHit | None:
        now = self.clock()
        want = self.terms(query) if self.terms else None
        cands = [e for e in self._entries if e.scope == scope and now - e.cached_at < self.ttl_s
                 and (want is None or e.terms == want)]               # exact-term guard
        if not cands:
            return None
        sims = np.stack([e.vec for e in cands]) @ self.embed([query])[0]   # dot = cosine
        best = int(np.argmax(sims))
        if sims[best] < self.threshold:
            return None
        e = cands[best]
        return CacheHit(e.answer, float(sims[best]), e.query, e.cached_at)

    def store(self, scope: str, query: str, answer: Answer) -> None:
        terms = self.terms(query) if self.terms else frozenset()
        vec = self.embed([query])[0]
        self._entries.append(_Entry(scope, query, terms, vec, answer, self.clock()))

    def invalidate(self, prefix: str) -> int:
        """Drop every entry whose scope starts with prefix. Returns how many were dropped."""
        keep = [e for e in self._entries if not e.scope.startswith(prefix)]
        dropped, self._entries = len(self._entries) - len(keep), keep
        return dropped

# script_id: day_085_caching_cost_scale/atlas_cache
if __name__ == "__main__":
    fix = Answer(text="ORB-1101 LeaseExpired: let heartbeats renew the lease, or raise lease_seconds.",
                 citations=[Citation(doc_id="error-codes", quote="A job outlived its lease")])
    for guard in (None, key_terms):
        cache = SemanticCache(hashed_embed, threshold=0.70, terms=guard)   # demo value
        cache.store("demo", "How do I fix ORB-1101 errors?", fix)
        cache.store("demo", "How do I configure retries in Orbit 3.x?", fix)
        print("guard on" if guard else "guard off")
        for q in ["How can I fix ORB-1101 errors?", "How do I fix ORB-1102 errors?",
                  "How do I configure retries in Orbit 2.x?"]:
            hit = cache.lookup("demo", q)
            print(f"  {q:42} " + (f"hit {hit.similarity:.2f} <- {hit.source_query}" if hit else "miss"))

guard off
  How can I fix ORB-1101 errors?             hit 0.84 <- How do I fix ORB-1101 errors?
  How do I fix ORB-1102 errors?              hit 0.90 <- How do I fix ORB-1101 errors?
  How do I configure retries in Orbit 2.x?   hit 0.89 <- How do I configure retries in Orbit 3.x?
guard on
  How can I fix ORB-1101 errors?             hit 0.84 <- How do I fix ORB-1101 errors?
  How do I fix ORB-1102 errors?              miss
  How do I configure retries in Orbit 2.x?   miss

# script_id: day_085_caching_cost_scale/threshold_sweep
# Hit rate vs false-hit rate across thresholds, on 30 labelled Orbit pairs.
# Install: uv add numpy pydantic   Env: ORBIT_DATA (default datasets/orbit)
# Offline by default; with OPENAI_API_KEY set it uses real embeddings (first_semantic_cache.py).
import json
import os
from dataclasses import dataclass
from pathlib import Path

from atlas_cache import SemanticCache, hashed_embed, key_terms

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
GOLD = {r["id"]: r["question"] for r in map(json.loads, (DATA_DIR / "eval/qa_gold.jsonl").open())}


@dataclass(frozen=True)
class CachePair:
    a: str
    b: str
    should_hit: bool


PARAPHRASES = {   # qa_gold id -> how a user might ask the same thing an hour later
    "qa-001": "What do lease_seconds and heartbeat_seconds default to in orbit.toml?",
    "qa-002": "Which release fixed the Redis reconnect storm after failover?",
    "qa-004": "ORB-1202: what does it mean and how do I fix it?",
    "qa-005": "What delays come before the first three retries with exponential backoff?",
    "qa-011": "Postgres broker pickup latency is stuck near 5 seconds, why?",
    "qa-012": "Is there a safe prefetch value for Orbit 3.2.x?",
    "qa-013": "Do Flows work with the SQS broker?",
    "qa-015": "What time zone are Orbit cron expressions evaluated in?",
    "qa-016": "How long are dead-lettered jobs kept by Orbit?",
    "qa-018": "Is there a Prometheus /metrics endpoint in Orbit?",
    "qa-021": "For how long does Orbit remember an idempotency key?",
    "qa-023": "Why not use orbit doctor as a Kubernetes liveness probe?",
    "qa-025": "What's the way to stop an Orbit worker on Windows?",
    "qa-026": "Can the dashboard run several replicas in token mode?",
    "qa-029": "How often did workers send heartbeats in Orbit 2.x?",
}

# script_id: day_085_caching_cost_scale/threshold_sweep
NEAR_MISSES = [   # look alike, need different answers; the last four have no exact term
    ("How do I fix ORB-1101?", "How do I fix ORB-1102?"),
    ("What does ORB-1001 mean?", "What does ORB-1002 mean?"),
    ("What does ORB-1201 mean?", "What does ORB-1202 mean?"),
    ("Why am I seeing ORB-1701?", "Why am I seeing ORB-1702?"),
    ("How do I configure retries in Orbit 2.x?", "How do I configure retries in Orbit 3.x?"),
    ("How often do workers send heartbeats in 2.x?", "How often do workers send heartbeats in 3.x?"),
    ("What is the default lease_seconds?", "What is the default heartbeat_seconds?"),
    ("What is the default max_attempts?", "What is the default max_delay_seconds?"),
    ("Is prefetch above 100 safe on 3.2.x?", "Is prefetch above 100 safe on 3.1.x?"),
    ("What does ttl_days control?", "What does max_payload_kb control?"),
    ("How do I set oidc_issuer?", "How do I set oidc_client_id?"),
    ("How do I stop an Orbit worker on Windows?", "How do I stop an Orbit worker on Kubernetes?"),
    ("Does the Redis broker support priorities?", "Does the SQS broker support priorities?"),
    ("How long does Orbit keep dead-lettered jobs?", "How long does Orbit keep job results?"),
    ("How do I enable tracing?", "How do I enable metrics?"),
]
CACHE_PAIRS = ([CachePair(GOLD[i], p, True) for i, p in PARAPHRASES.items()]
               + [CachePair(a, b, False) for a, b in NEAR_MISSES])


@dataclass(frozen=True)
class CacheReport:
    threshold: float
    hit_rate: float          # paraphrases that hit: higher saves more
    false_hit_rate: float    # near-misses that hit: each one is a wrong answer


def sweep_thresholds(cache: SemanticCache, pairs: list[CachePair],
                     thresholds: list[float]) -> list[CacheReport]:
    sims = (cache.embed([p.a for p in pairs]) * cache.embed([p.b for p in pairs])).sum(axis=1)
    same = [cache.terms is None or cache.terms(p.a) == cache.terms(p.b) for p in pairs]
    pos = [i for i, p in enumerate(pairs) if p.should_hit]
    neg = [i for i, p in enumerate(pairs) if not p.should_hit]
    reports = []
    for t in thresholds:                       # embed once, reuse the similarities per threshold
        hit = [bool(sims[i] >= t and same[i]) for i in range(len(pairs))]
        reports.append(CacheReport(t, sum(hit[i] for i in pos) / len(pos),
                                   sum(hit[i] for i in neg) / len(neg)))
    return reports


if __name__ == "__main__":
    embed = hashed_embed
    if os.environ.get("OPENAI_API_KEY"):
        from first_semantic_cache import openai_embed as embed
    grid = [x / 100 for x in range(50, 100, 5)]
    for terms in (None, key_terms):
        print("guard on" if terms else "guard off", "\nthreshold  hit_rate  false_hit_rate")
        for r in sweep_thresholds(SemanticCache(embed, terms=terms), CACHE_PAIRS, grid):
            print(f"  {r.threshold:.2f}     {r.hit_rate:6.0%}    {r.false_hit_rate:8.0%}")

guard off
threshold  hit_rate  false_hit_rate
  0.50       100%        100%
  0.55       100%         93%
  0.60       100%         93%
  0.65        93%         93%
  0.70        87%         93%
  0.75        73%         67%
  0.80        47%         60%
  0.85        40%         47%
  0.90        27%         20%
  0.95         7%          0%
guard on
threshold  hit_rate  false_hit_rate
  0.50       100%         27%
  0.55       100%         27%
  0.60       100%         27%
  0.65        93%         27%
  0.70        87%         27%
  0.75        73%         13%
  0.80        47%         13%
  0.85        40%          7%
  0.90        27%          0%
  0.95         7%          0%

# script_id: day_085_caching_cost_scale/atlas_cache
@dataclass(frozen=True)
class Principal:                                   # compact copy of Day 83
    tenant_id: str
    key_id: str
    plan: Literal["demo", "free", "team"]
    scopes: frozenset[str]


def tenant_key(p: Principal, *parts: str) -> str:  # Day 83: every key starts with the tenant
    return ":".join(["t", p.tenant_id, *parts])


def cache_scope(p: Principal, *, model: str, prompt_version: str, index_version: str) -> str:
    """Everything that changes the right answer goes here, and must match exactly."""
    return tenant_key(p, "cache", f"idx={index_version}", f"prompt={prompt_version}", f"model={model}")


def cached_ask(cache: SemanticCache, scope: str, question: str, ask: Callable[[str], Answer],
               *, history_len: int = 0) -> tuple[Answer, CacheHit | None]:
    if history_len:                                # follow-ups depend on the conversation
        return ask(question), None
    scope = f"{scope}:orbit={detect_version(question)}"
    if hit := cache.lookup(scope, question):
        return hit.answer, hit
    answer = ask(question)
    if answer.citations:                           # never cache "I don't know" or uncited text
        cache.store(scope, question, answer)
    return answer, None

# script_id: day_085_caching_cost_scale/atlas_cache
if __name__ == "__main__":
    calls: list[str] = []
    def ask(q: str) -> Answer:                     # stand-in for Day 28's ask(): counts real calls
        calls.append(q)
        return Answer(text=f"answer v{len(calls)}", citations=[Citation(doc_id="error-codes", quote="...")])
    acme = Principal("acme", "k1", "team", frozenset({"ask"}))
    zephyr = Principal("zephyr", "k2", "team", frozenset({"ask"}))
    cache = SemanticCache(hashed_embed, threshold=0.70)     # demo value; yours comes from the sweep
    v1 = dict(model="gpt-6-luna", prompt_version="ask-v4", index_version="2026-11-01")
    for who, q, idx in [(acme, "How do I fix ORB-1101 errors?", v1),
                        (acme, "How can I fix ORB-1101 errors?", v1),
                        (zephyr, "How can I fix ORB-1101 errors?", v1),
                        (acme, "How can I fix ORB-1101 errors?", {**v1, "index_version": "2026-11-08"})]:
        answer, hit = cached_ask(cache, cache_scope(who, **idx), q, ask)
        print(f"{who.tenant_id:6} idx={idx['index_version']}  {'hit ' if hit else 'miss'}  {answer.text}")
    print("erasure for acme dropped", cache.invalidate("t:acme:"), "entries")

acme   idx=2026-11-01  miss  answer v1
acme   idx=2026-11-01  hit   answer v1
zephyr idx=2026-11-01  miss  answer v2
acme   idx=2026-11-08  miss  answer v3
erasure for acme dropped 2 entries

# script_id: day_085_caching_cost_scale/atlas_router
# atlas/model_router.py: cheapest adequate model first, escalate when a check fails.
# Install: uv add numpy pydantic   Env: ATLAS_MODEL, ATLAS_STRONG_MODEL, ORBIT_DATA
from __future__ import annotations

import hashlib
import os
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable

from atlas_cache import Answer, Citation, key_terms

CHEAP = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
STRONG = os.environ.get("ATLAS_STRONG_MODEL", "gpt-6.1-terra")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
# USD per 1M tokens: (input, cached input, output). As of 2026-11; verify at the pricing page.
PRICES = {"gpt-6-luna": (0.10, 0.01, 0.50), "gpt-6.1-terra": (1.00, 0.05, 5.00)}
HARD = re.compile(r"\b(why|investigate|debug|root cause|compare|migrat\w*|upgrad\w*|in what order)\b", re.I)


@dataclass(frozen=True)
class RouteDecision:
    model: str
    effort: str           # reasoning effort (Day 6): a second, finer cost knob
    reason: str


def choose_model(question: str, *, plan: str = "free") -> RouteDecision:
    if plan == "demo":                              # public demo tenant: cheap only
        return RouteDecision(CHEAP, "low", "demo plan")
    if len(question) > 1_500 or HARD.search(question) or question.count("?") > 1:
        return RouteDecision(STRONG, "medium", "investigation / multi-part")
    if key_terms(question):
        return RouteDecision(CHEAP, "low", "exact-term lookup")
    return RouteDecision(CHEAP, "medium", "default")

# script_id: day_085_caching_cost_scale/atlas_router
KNOWN_DOCS = {p.stem for p in (DATA_DIR / "docs").glob("*.md")}
UNSURE = re.compile(r"\b(i don't know|not sure|cannot find|no information)\b", re.I)


def needs_escalation(answer: Answer) -> str | None:
    """Day 44-style checks. Returns a reason to escalate, or None if the answer may ship."""
    if UNSURE.search(answer.text):
        return "model unsure"
    if not answer.citations:
        return "no citation"
    unknown = [c.doc_id for c in answer.citations if c.doc_id not in KNOWN_DOCS]
    return f"cites unknown doc {unknown[0]!r}" if unknown else None


@dataclass
class CallResult:
    answer: Answer
    input_tokens: int
    output_tokens: int
    cached_tokens: int = 0
    seconds: float = 0.0


ModelCall = Callable[[str, RouteDecision], CallResult]     # plug in any provider


def cost_usd(model: str, input_tokens: int, output_tokens: int, cached_tokens: int = 0) -> float:
    if model not in PRICES:
        raise KeyError(f"no price for {model!r}: add it to PRICES from the pricing page")
    p_in, p_cached, p_out = PRICES[model]
    return ((input_tokens - cached_tokens) * p_in + cached_tokens * p_cached + output_tokens * p_out) / 1e6

# script_id: day_085_caching_cost_scale/atlas_router
@dataclass
class RouteTrace:
    decisions: list[RouteDecision] = field(default_factory=list)
    escalated: bool = False
    cost_usd: float = 0.0
    latency_s: float = 0.0


def run_once(call: ModelCall, question: str, d: RouteDecision, trace: RouteTrace) -> Answer:
    r = call(question, d)
    trace.decisions.append(d)
    trace.cost_usd += cost_usd(d.model, r.input_tokens, r.output_tokens, r.cached_tokens)
    trace.latency_s += r.seconds                    # escalation is sequential: the user waits for both
    return r.answer


def answer_routed(question: str, *, plan: str = "free", call: ModelCall) -> tuple[Answer, RouteTrace]:
    trace = RouteTrace()
    first = choose_model(question, plan=plan)
    answer = run_once(call, question, first, trace)
    reason = needs_escalation(answer)
    if reason and first.model != STRONG and plan != "demo":
        trace.escalated = True
        answer = run_once(call, question, RouteDecision(STRONG, "medium", f"escalated: {reason}"), trace)
    return answer, trace

# script_id: day_085_caching_cost_scale/atlas_router
class SimulatedModels:
    """Offline ModelCall with canned behaviour (NOT real model quality). The cheap model
    fails multi-doc questions about half the time; the strong model rarely fails."""

    def __init__(self, gold: dict[str, list[str]]) -> None:
        self.gold = gold                            # question -> cited doc ids from qa_gold

    def __call__(self, question: str, d: RouteDecision) -> CallResult:
        docs = self.gold.get(question, ["faq"])
        roll = int(hashlib.sha256(f"{d.model}|{question}".encode()).hexdigest(), 16) % 100
        unsure = (55 if len(docs) > 1 else 15) if d.model == CHEAP else 4
        if roll < unsure:
            answer = Answer(text="I don't know.")
        elif roll < unsure + 8:                     # confidently wrong: real doc, wrong one
            answer = Answer(text="(simulated)", citations=[Citation(doc_id="glossary", quote="...")])
        else:
            answer = Answer(text="(simulated)", citations=[Citation(doc_id=docs[0], quote="...")])
        out = 250 if d.effort == "low" else 450     # reasoning tokens bill as output
        secs = (0.8 if d.model == CHEAP else 2.2) + out / 300
        return CallResult(answer, input_tokens=2_400, output_tokens=out, cached_tokens=1_100, seconds=secs)

# script_id: day_085_caching_cost_scale/openai_answerer
# Real ModelCall for the router. Install: uv add "openai>=3.46,<4" pydantic numpy
# Env: OPENAI_API_KEY, ORBIT_DATA. Cost: one Responses call per answer.
import re
import time

from openai import OpenAI

from atlas_cache import Answer
from atlas_router import DATA_DIR, CallResult, RouteDecision

client = OpenAI()
DOCS = {p.stem: p.read_text(encoding="utf-8") for p in (DATA_DIR / "docs").glob("*.md")}
WORDS = {d: set(re.findall(r"[a-z0-9_]+", t.lower())) for d, t in DOCS.items()}
INSTRUCTIONS = ("Answer the question about Orbit using only the docs provided. Cite each doc id "
                "you used with a short exact quote. If the docs don't answer it, say \"I don't know\" "
                "and cite nothing.")


def retrieve(question: str, k: int = 3) -> list[str]:      # stand-in for Day 28's retriever
    q = set(re.findall(r"[a-z0-9_]+", question.lower()))
    return sorted(DOCS, key=lambda d: -len(q & WORDS[d]))[:k]


def openai_call(question: str, d: RouteDecision) -> CallResult:
    context = "\n\n".join(f'<doc id="{i}">\n{DOCS[i][:6000]}\n</doc>' for i in retrieve(question))
    start = time.monotonic()
    r = client.responses.parse(model=d.model, reasoning={"effort": d.effort}, instructions=INSTRUCTIONS,
                               input=f"{context}\n\nQuestion: {question}", text_format=Answer,
                               max_output_tokens=25_000)
    u = r.usage
    return CallResult(answer=r.output_parsed or Answer(text="I don't know"),   # None on refusal
                      input_tokens=u.input_tokens, output_tokens=u.output_tokens,
                      cached_tokens=u.input_tokens_details.cached_tokens,
                      seconds=time.monotonic() - start)

# script_id: day_085_caching_cost_scale/router_eval
# all-cheap vs all-strong vs router on eval/qa_gold.jsonl. Install: uv add numpy pydantic
# Offline: simulated models (plumbing check). With OPENAI_API_KEY: real calls (≈ $0.30, as of 2026-11).
import json
import os

import numpy as np

from atlas_router import (CHEAP, DATA_DIR, STRONG, RouteDecision, RouteTrace, SimulatedModels,
                          answer_routed, run_once)

cases = [json.loads(line) for line in (DATA_DIR / "eval/qa_gold.jsonl").open()]
if os.environ.get("OPENAI_API_KEY"):
    from openai_answerer import openai_call as call
else:
    call = SimulatedModels({c["question"]: c["citations"] for c in cases})


def passed(case: dict, answer) -> bool:     # code check; add a Day 45 judge for live runs
    return any(c.doc_id in case["citations"] for c in answer.citations)


def run(strategy: str) -> dict:
    results = []
    for case in cases:
        if strategy == "router":
            answer, trace = answer_routed(case["question"], call=call)
        else:
            trace, model = RouteTrace(), CHEAP if strategy == "all-cheap" else STRONG
            answer = run_once(call, case["question"], RouteDecision(model, "medium", strategy), trace)
        results.append((passed(case, answer), trace))
    traces = [t for _, t in results]
    return {"strategy": strategy, "pass": np.mean([p for p, _ in results]),
            "usd": np.mean([t.cost_usd for t in traces]),
            "p95": np.percentile([t.latency_s for t in traces], 95),
            "esc": np.mean([t.escalated for t in traces])}


if __name__ == "__main__":
    print(f"{'strategy':10} {'pass':>5} {'$/request':>10} {'p95 s':>6} {'escalated':>9}")
    for r in map(run, ["all-cheap", "all-strong", "router"]):
        print(f"{r['strategy']:10} {r['pass']:5.0%} {r['usd']:10.5f} {r['p95']:6.1f} {r['esc']:9.0%}")

strategy    pass  $/request  p95 s escalated
all-cheap    50%    0.00037    2.3        0%
all-strong   93%    0.00360    3.7        0%
router       90%    0.00198    6.0       33%

# script_id: day_085_caching_cost_scale/cost_report
# Per-tenant / route / model cost from a JSONL call log, savings per lever, one budget guard.
# Install: uv add numpy pydantic   Run: python cost_report.py [calls.jsonl]
import json
import random
import sys
from collections import defaultdict
from pathlib import Path

from atlas_router import CHEAP, PRICES, STRONG, cost_usd


def synthetic_log(path: Path, n: int = 4_000, seed: int = 85) -> Path:
    """One FAKE day shaped like Atlas traffic. Point the report at your real log instead."""
    rng = random.Random(seed)
    with path.open("w") as f:
        for _ in range(n):
            tenant = rng.choices(["acme", "zephyr", "demo"], [5, 3, 2])[0]
            route = "ask" if tenant == "demo" else rng.choices(["ask", "investigate"], [9, 1])[0]
            model = STRONG if route == "investigate" or rng.random() < 0.12 else CHEAP
            hit = route == "ask" and rng.random() < 0.3              # semantic-cache hit: no call
            tin = 0 if hit else rng.randint(1_500, 4_000) * (3 if route == "investigate" else 1)
            f.write(json.dumps({"tenant": tenant, "route": route, "model": model, "semantic_hit": hit,
                                "input_tokens": tin, "cached_tokens": int(tin * rng.uniform(0.3, 0.6)),
                                "output_tokens": 0 if hit else rng.randint(150, 900)}) + "\n")
    return path


def row_cost(r: dict) -> float:
    return cost_usd(r["model"], r["input_tokens"], r["output_tokens"], r["cached_tokens"])

# script_id: day_085_caching_cost_scale/cost_report
def cost_report(log_path: Path, by: tuple[str, ...] = ("tenant", "route", "model")) -> list[dict]:
    groups: dict[tuple, dict] = defaultdict(lambda: {"requests": 0, "semantic_hits": 0, "cost_usd": 0.0})
    for r in map(json.loads, log_path.open()):
        g = groups[tuple(r[k] for k in by)]
        g["requests"] += 1
        g["semantic_hits"] += r["semantic_hit"]
        g["cost_usd"] += row_cost(r)
    return sorted(({**dict(zip(by, k)), **v} for k, v in groups.items()), key=lambda x: -x["cost_usd"])


def savings_by_lever(log_path: Path) -> dict[str, float]:
    rows = [json.loads(line) for line in log_path.open()]
    misses = [r for r in rows if r["route"] == "ask" and not r["semantic_hit"]]
    avg_ask = sum(map(row_cost, misses)) / len(misses)               # what a hit would have cost
    return {"provider prompt caching": sum(r["cached_tokens"] * (PRICES[r["model"]][0] - PRICES[r["model"]][1])
                                           for r in rows) / 1e6,
            "semantic cache": sum(r["semantic_hit"] for r in rows) * avg_ask,
            "routing (vs all-strong)": sum(cost_usd(STRONG, r["input_tokens"], r["output_tokens"],
                                                    r["cached_tokens"]) - row_cost(r)
                                           for r in rows if r["model"] == CHEAP)}


DAILY_BUDGET_USD = {"acme": 3.20, "zephyr": 2.50, "demo": 0.80}      # caps you set per tenant


def budget_alerts(log_path: Path, warn_at: float = 0.8) -> list[str]:
    alerts = []
    for r in cost_report(log_path, by=("tenant",)):
        cap = DAILY_BUDGET_USD.get(r["tenant"])
        if cap and r["cost_usd"] >= warn_at * cap:
            alerts.append(f"ALERT {r['tenant']}: ${r['cost_usd']:.2f} spent = {r['cost_usd'] / cap:.0%} of ${cap:.2f}/day")
    return alerts


if __name__ == "__main__":
    log = Path(sys.argv[1]) if len(sys.argv) > 1 else synthetic_log(Path("calls_synthetic.jsonl"))
    for r in cost_report(log, by=("tenant", "route")):
        print(f"{r['tenant']:7} {r['route']:12} {r['requests']:5} req  {r['semantic_hits']:4} hits  ${r['cost_usd']:.3f}")
    for lever, usd in savings_by_lever(log).items():
        print(f"saved by {lever:24} ${usd:.3f}")
    print(*budget_alerts(log) or ["no tenant over 80% of budget"], sep="\n")

acme    investigate    191 req     0 hits  $1.459
acme    ask           1859 req   545 hits  $1.172
zephyr  investigate    136 req     0 hits  $1.008
zephyr  ask           1014 req   322 hits  $0.635
demo    ask            800 req   237 hits  $0.522
saved by provider prompt caching  $1.773
saved by semantic cache           $1.001
saved by routing (vs all-strong)  $8.390
ALERT acme: $2.63 spent = 82% of $3.20/day