Monitoring, Online Evals & Feedback Loops

# script_id: day_086_monitoring_feedback/first_event_log
# One structured event per request, written by FastAPI middleware into SQLite.
# Install: uv add fastapi httpx   Env: none   Services: none (TestClient runs the app in-process)
import sqlite3
import time
import uuid

from fastapi import FastAPI, Request
from fastapi.testclient import TestClient

db = sqlite3.connect("events_demo.db", check_same_thread=False)
db.execute("CREATE TABLE IF NOT EXISTS events (request_id, ts, route, status, latency_ms)")
app = FastAPI()

@app.middleware("http")
async def record_event(request: Request, call_next):
    rid, t0 = request.headers.get("x-request-id") or uuid.uuid4().hex[:12], time.perf_counter()
    response = await call_next(request)
    latency_ms = round((time.perf_counter() - t0) * 1000, 1)
    db.execute("INSERT INTO events VALUES (?, ?, ?, ?, ?)",
               (rid, time.time(), request.url.path, response.status_code, latency_ms))
    db.commit()
    response.headers["x-request-id"] = rid      # the client quotes it when it sends feedback
    return response

@app.post("/v1/ask")
def ask(body: dict) -> dict:
    return {"answer": "Keep [worker] prefetch at 100 or below.", "citations": ["troubleshooting-memory"]}

TestClient(app).post("/v1/ask", json={"question": "Safe prefetch on 3.2.1?"})
print(db.execute("SELECT * FROM events ORDER BY ts DESC LIMIT 1").fetchone())

('5f0c2e9a41b7', 1762514410.38, '/v1/ask', 200, 2.4)     # your numbers will differ

# script_id: day_086_monitoring_feedback/atlas_monitoring
# atlas/monitoring.py: one event per request, scores, metrics, online checks, sampling, alerts.
# Install: uv add pydantic   Env: ATLAS_LOG_DIR (default logs), ORBIT_DATA   Services: none (SQLite)
# Builds on Day 42's latency summary. At scale, the same two tables live in Postgres.
from __future__ import annotations

import functools
import hashlib
import math
import os
import re
import sqlite3
import threading
import time
from dataclasses import asdict, dataclass, fields
from pathlib import Path
from typing import Callable, Literal

LOG_DIR = Path(os.environ.get("ATLAS_LOG_DIR", "logs"))
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
Source = Literal["code", "judge", "user"]


@dataclass(kw_only=True)
class RequestEvent:
    request_id: str
    ts: float
    tenant_id: str
    route: str
    model: str
    variant: str = "control"        # Day 87 sets this during a canary
    status: int
    latency_ms: float
    ttft_ms: float | None = None
    input_tokens: int = 0
    output_tokens: int = 0
    cost_usd: float = 0.0
    cached: bool = False            # Day 85 semantic-cache hit
    fallback: bool = False          # Day 84 CallMeta flags
    degraded: bool = False
    citations: int = 0


@dataclass(frozen=True)
class Score:
    request_id: str
    name: str
    passed: bool
    source: Source

# script_id: day_086_monitoring_feedback/atlas_monitoring
COLS = [f.name for f in fields(RequestEvent)]


class EventLog:
    """Append-only SQLite log: one row per request, plus scores that arrive later."""

    def __init__(self, path: Path = LOG_DIR / "atlas_events.db") -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        self.db = sqlite3.connect(path, check_same_thread=False, timeout=5)
        self.db.execute("PRAGMA journal_mode=WAL")       # readers don't block the writer
        self.db.execute("PRAGMA synchronous=NORMAL")     # safe with WAL, far fewer fsyncs
        self.lock = threading.Lock()                     # one writer at a time
        self.write(f"CREATE TABLE IF NOT EXISTS events ({', '.join(COLS)}, PRIMARY KEY (request_id))")
        self.write("CREATE TABLE IF NOT EXISTS scores (request_id, name, passed, source, ts)")
        self.write("CREATE INDEX IF NOT EXISTS events_ts ON events (ts)")

    def write(self, sql: str, params: tuple | list = ()) -> None:
        with self.lock, self.db:                         # `with db` commits, or rolls back
            self.db.execute(sql, params)

    def record(self, e: RequestEvent) -> None:
        self.write(f"INSERT OR REPLACE INTO events VALUES ({', '.join('?' * len(COLS))})",
                   list(asdict(e).values()))

    def add_score(self, request_id: str, name: str, passed: bool, source: Source) -> None:
        self.write("INSERT INTO scores VALUES (?, ?, ?, ?, ?)",
                   (request_id, name, int(passed), source, time.time()))

    def get(self, request_id: str) -> RequestEvent | None:
        row = self.db.execute("SELECT * FROM events WHERE request_id = ?", (request_id,)).fetchone()
        return RequestEvent(**dict(zip(COLS, row))) if row else None

    def window(self, minutes: int, now: float | None = None, **filters: str) -> list[RequestEvent]:
        """Events in (now - minutes, now], optionally filtered, e.g. tenant_id="acme"."""
        assert set(filters) <= set(COLS), f"unknown filter: {set(filters) - set(COLS)}"
        now = time.time() if now is None else now
        sql = "SELECT * FROM events WHERE ts > ? AND ts <= ?" + "".join(f" AND {k} = ?" for k in filters)
        rows = self.db.execute(sql, [now - minutes * 60, now, *filters.values()])
        return [RequestEvent(**dict(zip(COLS, r))) for r in rows]

    def scores(self, events: list[RequestEvent]) -> list[Score]:
        ids, out = [e.request_id for e in events], []
        for i in range(0, len(ids), 900):                # stay under SQLite's variable limit
            chunk = ids[i:i + 900]
            sql = f"SELECT request_id, name, passed, source FROM scores WHERE request_id IN ({','.join('?' * len(chunk))})"
            out += [Score(r, n, bool(p), s) for r, n, p, s in self.db.execute(sql, chunk)]
        return out

# script_id: day_086_monitoring_feedback/atlas_monitoring
def pct(values: list[float], q: float) -> float:
    """Nearest-rank percentile: q% of values are at or below the result."""
    if not values:
        return math.nan
    s = sorted(values)
    return s[max(0, math.ceil(q / 100 * len(s)) - 1)]


def rate(flags: list[bool]) -> float:
    return sum(flags) / len(flags) if flags else math.nan


def metrics(events: list[RequestEvent], scores: list[Score] = ()) -> dict[str, float]:
    """RED metrics, tokens and dollars, and quality, for one window of events."""
    ok = [e for e in events if e.status < 500]          # 4xx is the caller's problem, not an outage
    lat = [e.latency_ms for e in ok]
    m = {"n": len(events), "error_rate": rate([e.status >= 500 for e in events]),
         "p50_ms": pct(lat, 50), "p95_ms": pct(lat, 95), "mean_ms": sum(lat) / len(lat) if lat else math.nan,
         "ttft_p95_ms": pct([e.ttft_ms for e in ok if e.ttft_ms is not None], 95),
         "usd_total": sum(e.cost_usd for e in events),
         "usd_per_req": sum(e.cost_usd for e in events) / len(events) if events else math.nan,
         "cache_hit": rate([bool(e.cached) for e in ok]), "fallback": rate([bool(e.fallback) for e in ok]),
         "degraded": rate([bool(e.degraded) for e in ok]), "citation_rate": rate([e.citations > 0 for e in ok])}
    code: dict[str, bool] = {}                          # a request passes only if every check passed
    for s in scores:
        if s.source == "code":
            code[s.request_id] = code.get(s.request_id, True) and s.passed
    judge = [s.passed for s in scores if s.source == "judge"]
    thumbs_down = [not s.passed for s in scores if s.source == "user" and s.name == "thumbs"]
    return m | {"n_checked": len(code), "online_pass_rate": rate(list(code.values())),
                "n_judged": len(judge), "judge_pass_rate": rate(judge),
                "n_rated": len(thumbs_down), "thumbs_down_rate": rate(thumbs_down)}

# script_id: day_086_monitoring_feedback/atlas_monitoring
CURRENT_MAJOR = "3"                                   # Orbit 3.2.1 is current
EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+")
SECRET = re.compile(r"\b(?:sk|atk)[-_][A-Za-z0-9_-]{12,}")   # OpenAI-style and Atlas (Day 83) keys


@dataclass
class Answer:                                         # compact Day 28 Answer
    text: str
    citations: list[str]


@dataclass(frozen=True)
class CheckResult:
    name: str
    passed: bool


@functools.cache
def corpus_ids() -> frozenset[str]:
    return frozenset(p.stem for p in (DATA_DIR / "docs").glob("*.md"))


def majors(text: str) -> set[str]:
    """Major versions named like "2.x" or "3.2.1" (not "0.5 s", not the "2.x" inside "3.2.x")."""
    return set(re.findall(r"(?<![\d.])([1-9])\.(?:x\b|\d+\.(?:\d+|x\b))", text))


def scrub(text: str) -> str:                          # compact Day 73 regex scrub
    return SECRET.sub("[SECRET]", EMAIL.sub("[EMAIL]", text))


def run_online_checks(event: RequestEvent, answer: Answer, hits: list[str],
                      question: str = "") -> list[CheckResult]:
    """Reference-free: no gold answer needed, so these run on 100% of traffic.
    `event` is there so you can vary checks by route or tenant."""
    return [
        CheckResult("cites_existing_doc", set(answer.citations) <= corpus_ids()),
        CheckResult("cites_retrieved", set(answer.citations) <= set(hits)),
        CheckResult("version_matches", majors(answer.text) <= majors(question) | {CURRENT_MAJOR}),
        CheckResult("answered", "i don't know" not in answer.text.lower()),
        CheckResult("no_pii", scrub(answer.text) == answer.text),
    ]

# script_id: day_086_monitoring_feedback/atlas_monitoring
Judge = Callable[[str, Answer], "bool | None"]


def in_sample(request_id: str, rate: float = 0.1) -> bool:
    """Deterministic: the same request id is in or out on every run, in every process."""
    h = int.from_bytes(hashlib.sha256(request_id.encode()).digest()[:8], "big")
    return h < rate * 2**64


def score_request(log: EventLog, event: RequestEvent, answer: Answer, hits: list[str],
                  question: str = "", judge: Judge | None = None, rate: float = 0.1) -> None:
    """Runs after the response is sent: FastAPI BackgroundTasks here, an RQ worker at scale."""
    for c in run_online_checks(event, answer, hits, question):
        log.add_score(event.request_id, c.name, c.passed, "code")
    if judge is not None and in_sample(event.request_id, rate):
        verdict = judge(question, answer)             # None: the judge refused or broke, so no score
        if verdict is not None:
            log.add_score(event.request_id, "judge_grounded", verdict, "judge")

# script_id: day_086_monitoring_feedback/online_judge
# atlas/jobs.py addition: a sampled, validated judge that runs in an RQ worker, never inline.
# Install: uv add "openai>=3.46,<4" "rq>=3.0" redis pydantic
# Env: OPENAI_API_KEY, ATLAS_JUDGE_MODEL, REDIS_URL, ORBIT_DATA
# Services: Redis + `rq worker judge --url $REDIS_URL --with-scheduler` (Day 78). ≈ $0.0003 per judgment.
import os
from typing import Literal

from openai import OpenAI
from pydantic import BaseModel
from redis import Redis
from rq import Queue, Retry

from atlas_monitoring import DATA_DIR, Answer, EventLog, in_sample   # atlas.monitoring in R5

JUDGE_MODEL = os.environ.get("ATLAS_JUDGE_MODEL", "gpt-6-luna")
JUDGE_PROMPT = """You grade one answer from Atlas, an assistant for Orbit (current version 3.2.1).
Fail it if it relies on Orbit 2.x behaviour for a 3.x question, if the passages don't support
its claims, or if it answers a different question. Otherwise pass. Reason briefly first."""
client = OpenAI()


class Verdict(BaseModel):
    reasoning: str
    verdict: Literal["pass", "fail"]


def judge_job(request_id: str, question: str, text: str, citations: list[str]) -> None:
    """Runs in the worker process and writes its score back to the event log."""
    docs = [DATA_DIR / "docs" / f"{d}.md" for d in citations]
    passages = "\n\n".join(p.read_text()[:3000] for p in docs if p.exists()) or "(none cited)"
    r = client.responses.parse(model=JUDGE_MODEL, instructions=JUDGE_PROMPT, text_format=Verdict,
                               input=f"Question: {question}\n\nAnswer: {text}\n\nPassages:\n{passages}")
    if r.output_parsed is not None:                   # None = refusal: record nothing
        EventLog().add_score(request_id, "judge_grounded", r.output_parsed.verdict == "pass", "judge")


def maybe_enqueue_judge(q: Queue, request_id: str, question: str, answer: Answer, rate: float = 0.1) -> bool:
    if not in_sample(request_id, rate):
        return False
    q.enqueue(judge_job, request_id, question, answer.text, answer.citations,
              job_timeout=60, result_ttl=600, retry=Retry(max=2, interval=[10, 30]))
    return True


if __name__ == "__main__":
    q = Queue("judge", connection=Redis.from_url(os.environ.get("REDIS_URL", "redis://localhost:6379/0")))
    stale = Answer("On Orbit 2.x, set ORBIT_PREFETCH in the environment.", ["environment-variables-2x"])
    print("enqueued:", maybe_enqueue_judge(q, "demo-1", "What prefetch value is safe on Orbit 3.2.x?", stale, rate=1.0))

# script_id: day_086_monitoring_feedback/feedback
# atlas/feedback.py: thumbs tied to a request id, owner-checked, plus policy-controlled content capture.
# Install: uv add fastapi pydantic   Env: ATLAS_LOG_DIR   Services: none
from __future__ import annotations

import json
import time
from dataclasses import dataclass
from typing import Callable, Literal

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field

from atlas_monitoring import Answer, EventLog, scrub    # atlas.monitoring in R5


class Feedback(BaseModel):
    request_id: str = Field(max_length=64)
    rating: Literal["up", "down"]
    reason: Literal["wrong", "unhelpful", "unsafe", "other"] | None = None
    comment: str | None = Field(default=None, max_length=1000)


@dataclass(frozen=True)
class Principal:                  # compact Day 83 Principal: only the field used here
    tenant_id: str


def assert_owner(p: Principal, owner_tenant: str) -> None:         # Day 83
    if owner_tenant != p.tenant_id:
        raise HTTPException(404, "not found")   # 404, not 403: a 403 confirms the id exists


def feedback_router(log: EventLog, current: Callable[..., Principal]) -> APIRouter:
    log.write("CREATE TABLE IF NOT EXISTS feedback (request_id, ts, rating, reason, comment)")
    log.write("CREATE TABLE IF NOT EXISTS content (request_id PRIMARY KEY, user_id, question, answer, citations)")
    router = APIRouter()

    @router.post("/v1/feedback", status_code=204)
    def give_feedback(fb: Feedback, p: Principal = Depends(current)) -> None:
        event = log.get(fb.request_id)
        if event is None:
            raise HTTPException(404, "not found")
        assert_owner(p, event.tenant_id)
        comment = scrub(fb.comment) if fb.comment else None          # scrub on write (Day 73)
        log.write("INSERT INTO feedback VALUES (?, ?, ?, ?, ?)",
                  (fb.request_id, time.time(), fb.rating, fb.reason, comment))
        log.add_score(fb.request_id, "thumbs", fb.rating == "up", "user")

    return router

# script_id: day_086_monitoring_feedback/feedback
def store_content(log: EventLog, request_id: str, user_id: str, question: str, answer: Answer,
                  capture: bool = True) -> None:
    """Keep question/answer text only where policy allows (Day 73), scrubbed before it's stored."""
    if capture:
        log.write("INSERT OR REPLACE INTO content VALUES (?, ?, ?, ?, ?)",
                  (request_id, user_id, scrub(question), scrub(answer.text), json.dumps(answer.citations)))


def content_for(log: EventLog, request_id: str) -> dict | None:
    row = log.db.execute("SELECT c.*, e.tenant_id, e.ts FROM content c JOIN events e USING (request_id) "
                         "WHERE c.request_id = ?", (request_id,)).fetchone()
    keys = ["request_id", "user_id", "question", "answer", "citations", "tenant_id", "ts"]
    return dict(zip(keys, row)) if row else None

# script_id: day_086_monitoring_feedback/atlas_monitoring
@dataclass(frozen=True)
class AlertRule:
    name: str
    metric: str
    op: Literal[">", "<"]
    threshold: float                   # absolute, or a multiple of the baseline when baseline_days is set
    window_min: int
    min_samples: int
    severity: Literal["page", "ticket"]
    baseline_days: int | None = None   # compare with the same window N days earlier


@dataclass(frozen=True)
class Alert:
    rule: str
    severity: str
    value: float
    limit: float
    n: int
    ts: float


SAMPLES = {"online_pass_rate": "n_checked", "judge_pass_rate": "n_judged", "thumbs_down_rate": "n_rated"}


def window_metrics(log: EventLog, minutes: int, end: float) -> dict[str, float]:
    ev = log.window(minutes, now=end)
    m = metrics(ev, log.scores(ev))
    return m | {"usd_per_hour": m["usd_total"] * 60 / minutes}


def evaluate_alerts(log: EventLog, rules: list[AlertRule], now: float | None = None) -> list[Alert]:
    now, fired = time.time() if now is None else now, []
    for r in rules:
        cur, count = window_metrics(log, r.window_min, now), SAMPLES.get(r.metric, "n")
        if cur[count] < r.min_samples:
            continue                                   # too little data to judge: stay quiet
        limit = r.threshold
        if r.baseline_days:
            base = window_metrics(log, r.window_min, now - r.baseline_days * 86400)
            if base[count] < r.min_samples:
                continue                               # no trustworthy baseline yet
            limit = r.threshold * base[r.metric]
        value = cur[r.metric]
        if (value > limit) if r.op == ">" else (value < limit):
            fired.append(Alert(r.name, r.severity, round(value, 4), round(limit, 4), int(cur[count]), now))
    return fired

# script_id: day_086_monitoring_feedback/atlas_monitoring
RULES = [
    AlertRule("availability", "error_rate", ">", 0.02, 5, 50, "page"),
    AlertRule("latency_p95", "p95_ms", ">", 1.5, 15, 50, "page", baseline_days=1),
    AlertRule("spend", "usd_per_hour", ">", 1.00, 60, 20, "ticket"),
    AlertRule("online_quality", "online_pass_rate", "<", 0.95, 30, 100, "page", baseline_days=1),
    AlertRule("judge_quality", "judge_pass_rate", "<", 0.9, 60, 20, "ticket", baseline_days=1),
    AlertRule("thumbs_down", "thumbs_down_rate", ">", 2.0, 60, 30, "ticket", baseline_days=1),
]

if __name__ == "__main__":                    # the by-hand example: 20 requests, 3 bad, 2 thumbs-down
    import tempfile
    log, now = EventLog(Path(tempfile.mkdtemp()) / "demo.db"), time.time()
    q = "What prefetch value is safe on Orbit 3.2.x?"
    for i in range(20):
        e = RequestEvent(request_id=f"r{i:02}", ts=now - 60 + i, tenant_id="acme", route="/v1/ask",
                         model="gpt-6-luna", status=200, latency_ms=900 + 30 * i, citations=1)
        log.record(e)
        stale = i in (4, 11, 17)
        text = "On Orbit 2.x, set ORBIT_PREFETCH." if stale else "On 3.2.1 keep prefetch at 100 or below."
        score_request(log, e, Answer(text, ["troubleshooting-memory"]), ["troubleshooting-memory"], q)
        if i in (0, 4, 7, 9, 13, 15, 17, 19):
            log.add_score(e.request_id, "thumbs", i not in (4, 17), "user")
    m = window_metrics(log, 5, now)
    print({k: round(m[k], 3) for k in ("n", "online_pass_rate", "n_rated", "thumbs_down_rate")})
    slow = [800.0] * 90 + [6000.0] * 10
    print(f"mean {sum(slow) / len(slow):.0f} ms vs p95 {pct(slow, 95):.0f} ms")
    print("sample:", sum(in_sample(f"req-{i}") for i in range(1000)), "of 1000; stable:", in_sample("r04") == in_sample("r04"))
    print("alerts:", evaluate_alerts(log, RULES, now) or "none (n=20 is below every min_samples)")

{'n': 20, 'online_pass_rate': 0.85, 'n_rated': 8, 'thumbs_down_rate': 0.25}
mean 1320 ms vs p95 6000 ms
sample: 93 of 1000; stable: True
alerts: none (n=20 is below every min_samples)
# your numbers will differ

# script_id: day_086_monitoring_feedback/stub_api
# An offline stand-in for the Day 77 Atlas API: fake retrieval and answers, real logging and scoring.
# Install: uv add fastapi httpx uvicorn pydantic   Env: ORBIT_DATA, ATLAS_LOG_DIR, ATLAS_SCENARIO
# Serve it over HTTP: ATLAS_SCENARIO=stale_index python stub_api.py --serve   (port 8000)
from __future__ import annotations

import hashlib
import json
import os
import random
import sys
import time
from typing import Callable

from fastapi import BackgroundTasks, Depends, FastAPI, Header, HTTPException

from atlas_monitoring import DATA_DIR, Answer, EventLog, RequestEvent, corpus_ids, score_request
from feedback import Principal, feedback_router, store_content

STALE_DOCS = {"configuration-reference", "concepts-workers", "troubleshooting-memory"}
KEYS = {"key-acme": "acme", "key-zephyr": "zephyr", "key-demo": "demo"}   # toy keys; Day 83 stores hashes
PRICE_IN, PRICE_OUT = 0.10, 0.50    # gpt-6-luna USD per 1M tokens, as of 2026-11; verify at the pricing page


def current_principal(authorization: str = Header("")) -> Principal:
    tenant = KEYS.get(authorization.removeprefix("Bearer "))
    if tenant is None:
        raise HTTPException(401, "invalid API key")
    return Principal(tenant)


def _load_gold() -> dict[str, list[str]]:
    gold = {}
    for name, q, docs in (("qa_gold", "question", "citations"), ("retrieval_queries", "query", "relevant_doc_ids")):
        for line in (DATA_DIR / f"eval/{name}.jsonl").read_text().splitlines():
            if line.strip():
                row = json.loads(line)
                gold[row[q]] = row[docs]
    return gold


GOLD = _load_gold()


def fake_retrieve(question: str) -> list[str]:
    """Gold docs for known questions, two hash-picked docs for anything else."""
    if question in GOLD:
        return list(GOLD[question])
    docs, h = sorted(corpus_ids()), int(hashlib.sha256(question.encode()).hexdigest(), 16)
    return sorted({docs[h % len(docs)], docs[(h // 7) % len(docs)]})

# script_id: day_086_monitoring_feedback/stub_api
def fake_answer(question: str, needed: list[str], scenario: str, rng: random.Random) -> tuple[Answer, list[str]]:
    hits = [d for d in needed if not (scenario == "stale_index" and d in STALE_DOCS)]
    if hits == needed:
        return Answer(f"Per the {needed[0]} page (Orbit 3.2.1): ...", needed), hits
    roll = rng.random()                         # three ways a stale index shows up
    if roll < 0.5:                              # answers from the old 2.x page
        old = "environment-variables-2x"
        return Answer("On Orbit 2.x, set ORBIT_PREFETCH in the environment.", [old]), hits + [old]
    if roll < 0.8:
        return Answer("I don't know: the docs I found don't cover this.", []), hits
    return Answer("Keep [worker] prefetch at 100 or below.", needed), hits    # cites pages it never retrieved


def fake_judge(question: str, answer: Answer) -> bool:
    """Stand-in for online_judge: fails 2.x answers, plus ~3% seeded disagreement."""
    noise = int(hashlib.sha256((question + answer.text).encode()).hexdigest(), 16) % 100 < 3
    return ("2.x" not in answer.text) != noise

# script_id: day_086_monitoring_feedback/stub_api
def create_stub_app(log: EventLog, scenario: str = "baseline", clock: Callable[[], float] = time.time,
                    seed: int = 0, judge: Callable = fake_judge) -> FastAPI:
    app, rng = FastAPI(title=f"Atlas stub ({scenario})"), random.Random(seed)

    @app.post("/v1/ask")
    def ask(body: dict, bg: BackgroundTasks, p: Principal = Depends(current_principal),
            x_user_id: str = Header("anon")) -> dict:
        question, rid = str(body.get("question", "")), f"{rng.getrandbits(64):016x}"
        answer, hits = fake_answer(question, fake_retrieve(question), scenario, rng)
        latency = rng.lognormvariate(7.0, 0.6)                 # median ~1.1 s, p95 ~2.9 s
        tin, tout = rng.randint(3500, 5500), rng.randint(150, 450)
        e = RequestEvent(request_id=rid, ts=clock(), tenant_id=p.tenant_id, route="/v1/ask",
                         model="gpt-6-luna", status=500 if rng.random() < 0.005 else 200,
                         latency_ms=round(latency, 1), ttft_ms=round(latency * 0.3, 1),
                         input_tokens=tin, output_tokens=tout, citations=len(answer.citations),
                         cost_usd=(tin * PRICE_IN + tout * PRICE_OUT) / 1e6)
        log.record(e)       # the real app records in middleware; the stub logs its simulated latency
        if e.status == 500:
            raise HTTPException(500, "upstream error")
        store_content(log, rid, x_user_id, question, answer)
        bg.add_task(score_request, log, e, answer, hits, question, judge)    # after the response
        return {"request_id": rid, "answer": answer.text, "citations": answer.citations}

    app.include_router(feedback_router(log, current_principal))
    return app


if __name__ == "__main__":
    app = create_stub_app(EventLog(), os.environ.get("ATLAS_SCENARIO", "baseline"))
    if "--serve" in sys.argv:
        import uvicorn
        uvicorn.run(app, port=8000)
    else:
        from fastapi.testclient import TestClient
        tc, acme, zephyr = TestClient(app), {"authorization": "Bearer key-acme"}, {"authorization": "Bearer key-zephyr"}
        r = tc.post("/v1/ask", json={"question": "What prefetch value is safe on Orbit 3.2.x?"}, headers=acme).json()
        print(r)
        down = {"request_id": r["request_id"], "rating": "down", "reason": "wrong"}
        print("zephyr rates acme's answer:", tc.post("/v1/feedback", json=down, headers=zephyr).status_code)
        print("acme rates its own answer:", tc.post("/v1/feedback", json=down, headers=acme).status_code)

{'request_id': '629f6fbed82c07cd', 'answer': 'Per the troubleshooting-memory page (Orbit 3.2.1): ...', 'citations': ['troubleshooting-memory', 'concepts-workers', 'configuration-reference']}
zephyr rates acme's answer: 404
acme rates its own answer: 204
# your numbers will differ

# script_id: day_086_monitoring_feedback/replay_traffic
# scripts/replay_traffic.py: replay a seeded mix of Orbit questions against an Atlas API.
# Install: none for live runs (stdlib); offline mode needs fastapi httpx   Env: ORBIT_DATA
# Live:    ATLAS_REPLAY_KEYS='{"acme": "...", "zephyr": "...", "demo": "..."}' python replay_traffic.py --url http://localhost:8000
# Offline: python replay_traffic.py   (in-process stub API; no key, no services). Reused on Days 87, 88, 90.
from __future__ import annotations

import json
import math
import os
import random
import sys
import time
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from pathlib import Path
from typing import Callable

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
Send = Callable[[str, dict, dict], tuple[int, dict]]        # (path, json body, headers) -> (status, body)


@dataclass
class ReplayReport:
    sent: int
    ok: int
    errors: int
    p50_ms: float
    p95_ms: float


def _jsonl(path: Path) -> list[dict]:
    return [json.loads(line) for line in path.read_text().splitlines() if line.strip()]


def build_workload(n: int, seed: int = 0) -> list[str]:
    rng = random.Random(seed)
    pools = [[r["question"] for r in _jsonl(DATA_DIR / "eval/qa_gold.jsonl")],
             [r["query"] for r in _jsonl(DATA_DIR / "eval/retrieval_queries.jsonl")],
             [r["title"] for r in _jsonl(DATA_DIR / "issues/issues.jsonl")]]
    return [rng.choice(rng.choices(pools, weights=[0.5, 0.3, 0.2])[0]) for _ in range(n)]

# script_id: day_086_monitoring_feedback/replay_traffic
def http_send(base_url: str, timeout: float = 30) -> Send:
    def send(path: str, body: dict, headers: dict) -> tuple[int, dict]:
        req = urllib.request.Request(base_url.rstrip("/") + path, json.dumps(body).encode(),
                                     {"content-type": "application/json", **headers})
        try:
            with urllib.request.urlopen(req, timeout=timeout) as r:
                raw = r.read()
                return r.status, json.loads(raw) if raw else {}
        except urllib.error.HTTPError as e:
            return e.code, {}
        except OSError:                                   # refused, reset, timed out
            return 599, {}
    return send


def replay_traffic(base_url: str, *, keys: dict[str, str], n: int, rps: float, seed: int = 0,
                   send: Send | None = None, sleep: Callable[[float], None] = time.sleep, workers: int = 8,
                   on_result: Callable[[str, str, str, int, dict], None] | None = None) -> ReplayReport:
    """keys maps tenant -> API key. Requests go out at `rps`, as one of 20 users per tenant."""
    send, rng = send or http_send(base_url), random.Random(seed)
    plan = [(q, rng.choice(sorted(keys)), f"user-{rng.randrange(20)}") for q in build_workload(n, seed)]
    lat: list[float] = []
    statuses: list[int] = []

    def one(q: str, tenant: str, user: str) -> None:
        t0 = time.perf_counter()
        status, body = send("/v1/ask", {"question": q}, {"authorization": f"Bearer {keys[tenant]}", "x-user-id": user})
        lat.append((time.perf_counter() - t0) * 1000)
        statuses.append(status)
        if on_result:
            on_result(q, tenant, user, status, body)

    with ThreadPoolExecutor(workers) as pool:
        futures = []
        for job in plan:
            futures.append(pool.submit(one, *job))
            if workers == 1:
                futures[-1].result()              # serial mode keeps a simulated clock in step
            sleep(1 / rps)
        for f in futures:
            f.result()                            # surface exceptions from worker threads
    ok, s = sum(200 <= c < 300 for c in statuses), sorted(lat)
    p = lambda q: round(s[max(0, math.ceil(q / 100 * len(s)) - 1)], 1) if s else math.nan
    return ReplayReport(len(plan), ok, len(plan) - ok, p(50), p(95))

# script_id: day_086_monitoring_feedback/replay_traffic
class VirtualClock:
    def __init__(self, t: float) -> None:
        self.t = t

    def now(self) -> float:
        return self.t

    def sleep(self, dt: float) -> None:
        self.t += dt


def offline_demo(path: Path, *, n: int = 300, rps: float = 0.1, seed: int = 0,
                 on_tick: Callable | None = None) -> tuple:
    """Returns (log, {scenario: (end_ts, ReplayReport)}). on_tick(log, now) runs after each stale request."""
    from fastapi.testclient import TestClient
    from atlas_monitoring import EventLog
    from stub_api import KEYS, create_stub_app
    for suffix in ("", "-wal", "-shm"):
        Path(f"{path}{suffix}").unlink(missing_ok=True)
    log, rng, end, phases = EventLog(path), random.Random(seed), time.time(), {}
    keys = {tenant: key for key, tenant in KEYS.items()}
    for i, (scenario, offset) in enumerate((("baseline", 86400), ("stale_index", 0))):
        clock = VirtualClock(end - offset - n / rps)
        tc = TestClient(create_stub_app(log, scenario, clock.now, seed=seed + i))
        send = lambda p, b, h: (lambda r: (r.status_code, r.json() if r.content else {}))(tc.post(p, json=b, headers=h))

        def people(q: str, tenant: str, user: str, status: int, body: dict) -> None:
            if status != 200:
                return
            h, bad = {"authorization": f"Bearer {keys[tenant]}", "x-user-id": user}, any(
                s in body["answer"] for s in ("2.x", "don't know"))
            if rng.random() < 0.3:
                down = rng.random() < (0.6 if bad else 0.04)
                send("/v1/feedback", {"request_id": body["request_id"], "rating": "down" if down else "up"}, h)
            if bad and rng.random() < 0.3:        # implicit signal: ask again 40 s later
                clock.sleep(40)
                send("/v1/ask", {"question": "In other words: " + q}, h)

        def tick(dt: float) -> None:
            clock.sleep(dt)
            if on_tick and scenario == "stale_index":
                on_tick(log, clock.now())

        report = replay_traffic("in-process", keys=keys, n=n, rps=rps, seed=seed + i, send=send,
                                sleep=tick, workers=1, on_result=people)
        phases[scenario] = (clock.now(), report)
    return log, phases

# script_id: day_086_monitoring_feedback/replay_traffic
if __name__ == "__main__":
    if "--url" in sys.argv:
        url = sys.argv[sys.argv.index("--url") + 1]
        print(replay_traffic(url, keys=json.loads(os.environ["ATLAS_REPLAY_KEYS"]), n=200, rps=2.0))
        sys.exit()
    from atlas_monitoring import LOG_DIR, RULES, evaluate_alerts, window_metrics
    log, phases = offline_demo(LOG_DIR / "atlas_events.db")
    rows = {name: window_metrics(log, 90, end) for name, (end, _) in phases.items()}
    print(f"{'metric':<18}{'baseline':>10}{'stale_index':>13}")
    for k in ("n", "p50_ms", "p95_ms", "error_rate", "usd_per_req", "citation_rate",
              "online_pass_rate", "judge_pass_rate", "thumbs_down_rate"):
        print(f"{k:<18}{rows['baseline'][k]:>10.4g}{rows['stale_index'][k]:>13.4g}")
    print("replay:", phases["stale_index"][1])
    for a in evaluate_alerts(log, RULES, now=phases["stale_index"][0]):
        print(f"ALERT [{a.severity}] {a.rule}: {a.value} vs limit {a.limit} (n={a.n})")

metric              baseline  stale_index
n                        300          312
p50_ms                  1119         1163
p95_ms                  3224         2561
error_rate                 0     0.003205
usd_per_req        0.0005999    0.0005998
citation_rate              1       0.9678
online_pass_rate           1       0.8778
judge_pass_rate            1       0.9706
thumbs_down_rate     0.04819      0.08696
replay: ReplayReport(sent=300, ok=300, errors=0, p50_ms=3.3, p95_ms=4.7)
ALERT [page] online_quality: 0.8938 vs limit 0.95 (n=160)
# your numbers will differ

# script_id: day_086_monitoring_feedback/feedback_to_dataset
# atlas/feedback.py addition: failures -> deduplicated draft eval cases -> human review -> dataset v3.
# Install: uv add pydantic fastapi httpx   Env: ORBIT_DATA, ATLAS_LOG_DIR   Services: none
from __future__ import annotations

import hashlib
import json
import re
import time
from pathlib import Path
from typing import Literal

from pydantic import BaseModel, Field

from atlas_monitoring import LOG_DIR, EventLog
from feedback import content_for

SHARED_TENANTS = {"demo", "orbit-maintainers"}       # everyone else's cases stay tenant-scoped


class EvalCase(BaseModel):                           # compact Day 43 schema
    id: str
    input: str
    must_cite: list[str] = Field(default_factory=list)
    tags: list[str] = Field(default_factory=list)
    tenant: str = "demo"
    source: Literal["seed", "synthetic", "production"] = "seed"
    status: Literal["draft", "reviewed"] = "draft"


def failures(log: EventLog, since: float) -> dict[str, set[str]]:
    """request_id -> why it was picked: a thumbs-down, a failed code check or a judge fail."""
    rows = log.db.execute("SELECT s.request_id, s.source || ':' || s.name FROM scores s JOIN events e "
                          "USING (request_id) WHERE e.ts >= ? AND s.passed = 0", (since,))
    out: dict[str, set[str]] = {}
    for rid, why in rows:
        out.setdefault(rid, set()).add(why)
    return out


def dedup_key(question: str) -> str:                 # compact Day 43 normalisation
    return " ".join(re.findall(r"[a-z0-9.\-]+", question.lower().removeprefix("in other words: ")))


def feedback_to_cases(log: EventLog, since: float) -> list[EvalCase]:
    cases: dict[tuple[str, str], EvalCase] = {}
    for rid, why in failures(log, since).items():
        c = content_for(log, rid)
        if c is None:                                # content capture was off for this request
            continue
        key = (c["tenant_id"], dedup_key(c["question"]))
        digest = hashlib.sha256("|".join(key).encode()).hexdigest()[:8]
        case = cases.setdefault(key, EvalCase(id=f"prod-{digest}", input=c["question"],
                                              tenant=c["tenant_id"], source="production"))
        case.tags = sorted(set(case.tags) | why)
    return list(cases.values())

# script_id: day_086_monitoring_feedback/feedback_to_dataset
def save_dataset(name: str, version: str, cases: list[EvalCase], parent: str | None,
                 root: Path = Path("evals")) -> dict:
    """Compact Day 43 save: parent cases + reviewed new ones, with a manifest naming the parent."""
    def read(v: str) -> list[EvalCase]:
        f = root / name / v / "cases.jsonl"
        return [EvalCase.model_validate_json(x) for x in f.read_text().splitlines()] if f.exists() else []
    merged = {c.id: c for c in (read(parent) if parent else []) + [c for c in cases if c.status == "reviewed"]}
    body = "".join(c.model_dump_json() + "\n" for c in merged.values())
    out = root / name / version
    out.mkdir(parents=True, exist_ok=True)
    (out / "cases.jsonl").write_text(body)
    manifest = {"name": name, "version": version, "parent": parent, "count": len(merged),
                "sha256": hashlib.sha256(body.encode()).hexdigest()[:12], "created": time.strftime("%Y-%m-%d")}
    (out / "manifest.json").write_text(json.dumps(manifest, indent=2))
    return manifest


if __name__ == "__main__":
    from replay_traffic import offline_demo
    from stub_api import fake_retrieve
    log, phases = offline_demo(LOG_DIR / "atlas_events.db")
    drafts = feedback_to_cases(log, since=phases["stale_index"][0] - 3 * 3600)
    review = Path("review/ask_v3_candidates.jsonl")
    review.parent.mkdir(exist_ok=True)
    review.write_text("".join(c.model_dump_json() + "\n" for c in drafts))
    shared = [c for c in drafts if c.tenant in SHARED_TENANTS]
    print(f"{len(drafts)} drafts ({len(shared)} shareable) -> {review}")
    print(next((c for c in shared if "code:version_matches" in c.tags), shared[0]).model_dump_json())
    # SIMULATED REVIEW. A person does this: keep real failures, fill must_cite, flip the status.
    for c in shared[:10]:
        if any(t.startswith(("code:", "judge:")) for t in c.tags):
            c.must_cite, c.status = fake_retrieve(c.input.removeprefix("In other words: ")), "reviewed"
    print(save_dataset("ask", "v3", shared, parent="v2"))

30 drafts (10 shareable) -> review/ask_v3_candidates.jsonl
{"id":"prod-59f20f33","input":"How do I stop an Orbit worker on Windows?","must_cite":[],"tags":["code:answered","code:cites_retrieved","code:version_matches"],"tenant":"demo","source":"production","status":"draft"}
{'name': 'ask', 'version': 'v3', 'parent': 'v2', 'count': 9, 'sha256': '40c6187eae8e', 'created': '2026-11-07'}
# your numbers will differ

# script_id: day_086_monitoring_feedback/dashboard
# atlas/ui/dashboard.py: a Gradio Blocks dashboard over the event log, plus a like-button handler.
# Install: uv add "gradio>=6.31,<7" pandas   Env: ATLAS_LOG_DIR   Services: none
# Run: python dashboard.py --serve   (without --serve: builds the UI, prints a text report, exits)
from __future__ import annotations

import json
import math
import sys
import urllib.request
from pathlib import Path

import gradio as gr
import pandas as pd

from atlas_monitoring import LOG_DIR, EventLog, RequestEvent, Score, metrics

TILES = ["n", "p95_ms", "error_rate", "usd_per_req", "online_pass_rate", "thumbs_down_rate"]


def trend_frame(ev: list[RequestEvent], sc: list[Score]) -> pd.DataFrame:
    by_id: dict[str, list[Score]] = {}
    for s in sc:
        by_id.setdefault(s.request_id, []).append(s)
    t0, buckets = min((e.ts for e in ev), default=0.0), {}
    for e in ev:
        buckets.setdefault(int((e.ts - t0) // 300) * 5, []).append(e)
    rows = []
    for minute, es in sorted(buckets.items()):
        m = metrics(es, [s for e in es for s in by_id.get(e.request_id, [])])
        rows += [{"minute": minute, "series": k, "value": m[k]} for k in ("online_pass_rate", "citation_rate", "error_rate")]
    return pd.DataFrame(rows, columns=["minute", "series", "value"]).dropna()


def worst_frame(ev: list[RequestEvent], sc: list[Score], k: int = 10) -> pd.DataFrame:
    failed: dict[str, list[str]] = {}
    for s in sc:
        if not s.passed:
            failed.setdefault(s.request_id, []).append(f"{s.source}:{s.name}")
    worst = sorted(ev, key=lambda e: (-len(failed.get(e.request_id, [])), -e.latency_ms))[:k]
    return pd.DataFrame([{"request_id": e.request_id, "tenant": e.tenant_id, "latency_ms": e.latency_ms,
                          "failed": ", ".join(failed.get(e.request_id, []))} for e in worst])

# script_id: day_086_monitoring_feedback/dashboard
def snapshot(log: EventLog, minutes: float, tenant: str, now: float | None = None) -> list:
    ev = log.window(int(minutes), now, **({} if tenant == "all" else {"tenant_id": tenant}))
    sc = log.scores(ev)
    m = metrics(ev, sc)
    tiles = [None if math.isnan(m.get(k, math.nan)) else round(m[k], 4) for k in TILES]
    return [*tiles, trend_frame(ev, sc), worst_frame(ev, sc)]


def build_dashboard(log: EventLog, now: float | None = None) -> gr.Blocks:
    with gr.Blocks(title="Atlas monitoring") as demo:
        with gr.Row():
            tenant = gr.Dropdown(["all", "acme", "zephyr", "demo"], value="all", label="Tenant")
            minutes = gr.Slider(15, 1440, value=60, step=15, label="Window (minutes)")
        with gr.Row():
            tiles = [gr.Number(label=k) for k in TILES]
        trend = gr.LinePlot(x="minute", y="value", color="series", title="Rates per 5-minute bucket")
        worst = gr.Dataframe(label="Worst 10 requests: search the request id in your traces (Day 42)")
        for event in (demo.load, gr.Timer(30).tick, tenant.change, minutes.change):
            event(lambda mins, t: snapshot(log, mins, t, now), [minutes, tenant], [*tiles, trend, worst])
    return demo


def make_like_handler(api_url: str, api_key: str):
    """For the Day 79 chat UI: chatbot.like(handler, [request_ids]), where request_ids is a
    gr.State dict {message index: request id} that your chat function fills in."""
    def handler(like: gr.LikeData, request_ids: dict) -> None:
        idx = like.index if isinstance(like.index, int) else like.index[0]
        if request_ids.get(idx) is None or not isinstance(like.liked, bool):
            return                                     # unknown message, or a non-thumbs reaction
        body = {"request_id": request_ids[idx], "rating": "up" if like.liked else "down"}
        req = urllib.request.Request(f"{api_url}/v1/feedback", json.dumps(body).encode(),
                                     {"content-type": "application/json", "authorization": f"Bearer {api_key}"})
        urllib.request.urlopen(req, timeout=10).close()
    return handler


if __name__ == "__main__":
    path, now = LOG_DIR / "atlas_events.db", None
    if "--demo" in sys.argv or not path.exists():      # no real log yet: fill one with the replay
        from replay_traffic import offline_demo
        _, phases = offline_demo(path)
        now = phases["stale_index"][0]
    log = EventLog(path)
    demo = build_dashboard(log, now)
    if "--serve" in sys.argv:
        demo.launch()
    else:
        *tiles, trend, worst = snapshot(log, 60, "all", now)
        print(dict(zip(TILES, tiles)))
        print(worst.head(5).to_string(index=False))

{'n': 268, 'p95_ms': 2457.5, 'error_rate': 0.0037, 'usd_per_req': 0.0006, 'online_pass_rate': 0.8727, 'thumbs_down_rate': 0.1081}
      request_id tenant  latency_ms                            failed
eed175405631ca9f   demo      1267.1 code:version_matches, user:thumbs
68d52eb618ede6c3   demo      1188.7 code:version_matches, user:thumbs
0d589a58c842c19a   acme       811.1 code:version_matches, user:thumbs
e0463f9f83a81a4e zephyr       471.2 code:version_matches, user:thumbs
2af346e9a3706225   demo      2624.0                     code:answered
# your numbers will differ