Shipping Changes Safely

# file: release.toml
[prompts]
ask = "v3"                 # what everyone runs (the pin)

[models]
ask = "gpt-6-luna"

[[canaries]]
flag = "prompt.ask"        # "<prompt|model>.<name>"
control = "v3"             # must equal the pin above
candidate = "v4"
pct = 10                   # share of units that get the candidate
unit = "user"              # bucket by user: sticky for a whole chat

# script_id: day_087_shipping_changes_safely/first_flag
# Sticky bucketing + a TOML release config. Stdlib only (Python 3.11+ for tomllib).
import hashlib
import tomllib

release = tomllib.loads(
    '[prompts]\nask = "v3"\n'
    '[[canaries]]\nflag = "prompt.ask"\ncontrol = "v3"\ncandidate = "v4"\npct = 10\n'
)


def bucket(unit_id: str, flag: str) -> int:
    digest = hashlib.sha256(f"{flag}:{unit_id}".encode()).hexdigest()
    return int(digest[:8], 16) % 100          # 0-99, identical in every process


canary = release["canaries"][0]
for user in ["acme-u1", "acme-u3", "zephyr-u1", "zephyr-u9", "demo-u2", "orbit-maintainers-u2"]:
    b = bucket(user, canary["flag"])
    version = canary["candidate"] if b < canary["pct"] else canary["control"]
    print(f"{user:22} bucket={b:2}  ask -> {version}")

acme-u1                bucket=60  ask -> v3
acme-u3                bucket= 4  ask -> v4
zephyr-u1              bucket=44  ask -> v3
zephyr-u9              bucket= 8  ask -> v4
demo-u2                bucket=49  ask -> v3
orbit-maintainers-u2   bucket=27  ask -> v3

# script_id: day_087_shipping_changes_safely/bucketing_mistakes
# Three ways to break bucketing, printed as tables. Stdlib only.
import hashlib
import random
import subprocess
import sys


def bucket(unit_id: str, flag: str = "") -> int:
    return int(hashlib.sha256(f"{flag}:{unit_id}".encode()).hexdigest()[:8], 16) % 100


# 1. Random per request: one user's chat flips between prompts mid-conversation.
rng = random.Random(7)
print("1. random per request, acme-u1's 6 turns:",
      ["v4" if rng.random() < 0.5 else "v3" for _ in range(6)])

# 2. Bucket by tenant: at 10% you get all of one tenant or none of it.
mix = {"acme": 0.40, "zephyr": 0.25, "demo": 0.20, "orbit-maintainers": 0.15}
for pct in (10, 30):
    inside = [t for t in mix if bucket(t, "prompt.ask") < pct]
    share = sum(mix[t] for t in inside)
    print(f"2. by tenant at {pct}%: canary = {inside}, real traffic share = {share:.0%}")

# 3. Same hash for every flag: the same users land in every experiment.
users = [f"u{i}" for i in range(10_000)]
a = {u for u in users if bucket(u) < 10}
b_same = {u for u in users if bucket(u) < 10}
b_salted = {u for u in users if bucket(u, "model.investigate") < 10}
print(f"3. overlap with an unsalted second flag: {len(a & b_same) / len(a):.0%}, "
      f"salted: {len(a & b_salted) / len(a):.0%}")

# Bonus: built-in hash() is randomized per process (PYTHONHASHSEED), so it is not sticky.
for _ in range(2):
    out = subprocess.run([sys.executable, "-c", "print(hash('acme-u1') % 100)"],
                         capture_output=True, text=True).stdout.strip()
    print("hash('acme-u1') % 100 in a fresh process:", out)

1. random per request, acme-u1's 6 turns: ['v4', 'v4', 'v3', 'v4', 'v3', 'v4']
2. by tenant at 10%: canary = ['orbit-maintainers'], real traffic share = 15%
2. by tenant at 30%: canary = ['acme', 'demo', 'orbit-maintainers'], real traffic share = 75%
3. overlap with an unsalted second flag: 100%, salted: 9%
hash('acme-u1') % 100 in a fresh process: 26
hash('acme-u1') % 100 in a fresh process: 73

# script_id: day_087_shipping_changes_safely/atlas_release
# atlas/flags.py plus compact stand-ins for Days 21, 83 and 86, so this file runs on its own.
# Install: uv add "pydantic>=2"    Env: ORBIT_DATA (default datasets/orbit)
# Default: a seeded stub model, no key needed. ATLAS_LIVE=1 uses live_answer.py + OPENAI_API_KEY.
from __future__ import annotations

import hashlib
import importlib
import json
import math
import os
import random
import time
import tomllib
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, Literal

from pydantic import BaseModel, Field, model_validator

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))


class Canary(BaseModel):
    flag: str                                # "<prompt|model>.<name>", e.g. "prompt.ask"
    control: str
    candidate: str
    pct: int = Field(ge=0, le=100)
    unit: Literal["tenant", "user", "request"] = "user"


class ReleaseConfig(BaseModel):
    prompts: dict[str, str]
    models: dict[str, str]
    canaries: list[Canary] = Field(default_factory=list)

    @model_validator(mode="after")
    def control_is_the_pin(self) -> ReleaseConfig:
        for c in self.canaries:              # a canary compares against what everyone else runs
            kind, name = c.flag.split(".", 1)
            pinned = {"prompt": self.prompts, "model": self.models}.get(kind, {}).get(name)
            if pinned != c.control:
                raise ValueError(f"{c.flag}: control {c.control!r} is not the pinned {pinned!r}")
        return self


def load_release(path: Path = Path("release.toml")) -> ReleaseConfig:
    with path.open("rb") as f:               # tomllib only reads binary files
        return ReleaseConfig.model_validate(tomllib.load(f))

# script_id: day_087_shipping_changes_safely/atlas_release
class ReleaseWatcher:
    """Re-reads release.toml when its bytes change; a bad edit keeps the last good config."""

    def __init__(self, path: Path, min_interval_s: float = 2.0) -> None:
        self.path, self.min_interval_s = path, min_interval_s
        self.cfg = load_release(path)        # fail fast at boot: there is no last-good yet
        self._digest = hashlib.sha256(path.read_bytes()).hexdigest()
        self._checked = time.monotonic()
        self.last_error: str | None = None

    def get(self) -> ReleaseConfig:
        if time.monotonic() - self._checked < self.min_interval_s:
            return self.cfg                  # cheap path for most requests
        self._checked = time.monotonic()
        raw = self.path.read_bytes()
        digest = hashlib.sha256(raw).hexdigest()
        if digest != self._digest:
            try:
                self.cfg = ReleaseConfig.model_validate(tomllib.loads(raw.decode()))
                self._digest, self.last_error = digest, None
            except ValueError as e:          # TOMLDecodeError and ValidationError are ValueErrors
                self.last_error = f"{type(e).__name__}: {e}".splitlines()[0]
        return self.cfg


# Compact Day 21 registry: in Atlas, load_prompt() reads versioned files under atlas/prompts/.
PROMPTS = {
    ("ask", "v3"): "Answer from the Orbit docs. Cite every doc you used by its slug.",
    ("ask", "v4"): "Be concise. Answer from the Orbit docs and cite at most 2 docs by slug.",
}


def load_prompt(name: str, version: str) -> str:
    if (name, version) not in PROMPTS:
        raise KeyError(f"prompt {name}@{version} is not in the registry")
    return PROMPTS[(name, version)]

# script_id: day_087_shipping_changes_safely/atlas_release
@dataclass(frozen=True)
class Principal:                             # compact Day 83 Principal
    tenant_id: str
    key_id: str
    plan: Literal["demo", "free", "team"] = "team"
    scopes: frozenset[str] = frozenset({"ask"})


def bucket(unit_id: str, flag: str) -> int:
    """0-99, stable across processes and machines (built-in hash() is not)."""
    return int(hashlib.sha256(f"{flag}:{unit_id}".encode()).hexdigest()[:8], 16) % 100


def variant_for(p: Principal, canary: Canary, *,
                user_id: str | None = None) -> Literal["control", "canary"]:
    unit_id = {"tenant": p.tenant_id, "user": user_id or p.key_id,
               "request": uuid.uuid4().hex}[canary.unit]
    return "canary" if bucket(unit_id, canary.flag) < canary.pct else "control"


def resolve(kind: Literal["prompt", "model"], name: str, p: Principal, cfg: ReleaseConfig,
            *, user_id: str | None = None) -> tuple[str, str]:
    for c in cfg.canaries:
        if c.flag == f"{kind}.{name}":
            v = variant_for(p, c, user_id=user_id)
            return (c.candidate if v == "canary" else c.control), v
    return {"prompt": cfg.prompts, "model": cfg.models}[kind][name], "control"


def resolve_prompt(name: str, p: Principal, cfg: ReleaseConfig, *,
                   user_id: str | None = None) -> tuple[str, str]:
    return resolve("prompt", name, p, cfg, user_id=user_id)

# script_id: day_087_shipping_changes_safely/atlas_release
# $ per 1M tokens (input, output), as of 2026-11-06; verify at https://developers.openai.com/api/docs/pricing
PRICES = {"gpt-6-luna": (0.10, 0.50), "gpt-6.1-terra": (1.00, 5.00)}
DOC_SLUGS = {p.stem for p in (DATA_DIR / "docs").glob("*.md")}
GOLD = {c["question"]: c for c in map(json.loads, (DATA_DIR / "eval/qa_gold.jsonl").open())}
_rng = random.Random(87)


def retrieve(question: str) -> str:
    """Stand-in for Day 28 retrieval: the gold docs, so the canary measures the prompt only."""
    slugs = GOLD.get(question, {}).get("citations", [])
    return "\n\n".join(f"[{s}]\n{(DATA_DIR / 'docs' / f'{s}.md').read_text()[:3000]}" for s in slugs)


def stub_answer(question: str, prompt_text: str, model: str, context: str) -> dict:
    cites = list(GOLD[question]["citations"]) if question in GOLD else ["faq"]
    concise = "at most 2" in prompt_text
    if concise:
        cites = cites[:2]                    # the bug: 3-doc answers lose one
        if len(cites) == 2 and _rng.random() < 0.3:
            cites = cites[:1]                # "be concise" trims a needed citation
    elif len(cites) > 1 and model == "gpt-6-luna" and _rng.random() < 0.04:
        cites = cites[:-1]                   # v3 is good, not perfect
    if _rng.random() < 0.01:
        raise TimeoutError("stub provider timeout")
    return {"answer": "(stub)", "citations": cites, "input_tokens": 1800,
            "output_tokens": 180 if concise else 320,
            "latency_ms": max(50.0, _rng.gauss(650 if concise else 900, 120))}


def make_answer_fn() -> Callable[[str, str, str, str], dict]:
    if os.environ.get("ATLAS_LIVE") == "1":
        # Imported by name so the stub path never loads the OpenAI client.
        return importlib.import_module("live_answer").live_answer
    return stub_answer

# script_id: day_087_shipping_changes_safely/live_answer
# Real model call, used by atlas_release when ATLAS_LIVE=1.
# Install: uv add openai pydantic    Env: OPENAI_API_KEY
import time

from openai import OpenAI
from pydantic import BaseModel

client = OpenAI()          # created on first import, which only happens with ATLAS_LIVE=1


class CitedAnswer(BaseModel):
    answer: str
    citations: list[str]   # doc slugs, e.g. "configuration-reference"


def live_answer(question: str, prompt_text: str, model: str, context: str) -> dict:
    t0 = time.perf_counter()
    resp = client.responses.parse(
        model=model, instructions=prompt_text,
        input=f"Docs:\n{context}\n\nQuestion: {question}", text_format=CitedAnswer)
    if resp.output_parsed is None:       # refusal or no final text: count it as an error
        raise RuntimeError(f"no parsed answer (status={resp.status})")
    return {"answer": resp.output_parsed.answer, "citations": resp.output_parsed.citations,
            "input_tokens": resp.usage.input_tokens, "output_tokens": resp.usage.output_tokens,
            "latency_ms": (time.perf_counter() - t0) * 1000}

# script_id: day_087_shipping_changes_safely/atlas_release
@dataclass
class RequestEvent:                          # compact Day 86 RequestEvent + release fields
    request_id: str
    ts: float
    tenant_id: str
    flag: str
    variant: str
    prompt_version: str
    model: str
    latency_ms: float
    cost_usd: float
    error: bool
    passed: bool                             # online check: cites at least one real doc
    citations_ok: bool | None = None         # replay only: cited every gold doc


class EventLog:                              # in-memory stand-in for Day 86's SQLite EventLog
    def __init__(self) -> None:
        self.events: list[RequestEvent] = []

    def record(self, e: RequestEvent) -> None:
        self.events.append(e)

    def window(self, since: float, **filters: str) -> list[RequestEvent]:
        return [e for e in self.events
                if e.ts >= since and all(getattr(e, k) == v for k, v in filters.items())]

# script_id: day_087_shipping_changes_safely/atlas_release
def handle_ask(question: str, p: Principal, cfg: ReleaseConfig, log: EventLog,
               answer_fn: Callable[..., dict], *, user_id: str | None = None) -> dict:
    version, variant = resolve_prompt("ask", p, cfg, user_id=user_id)
    model, t0 = cfg.models["ask"], time.perf_counter()
    try:
        res, err = answer_fn(question, load_prompt("ask", version), model, retrieve(question)), False
    except Exception:                        # a canary that errors must show up, not vanish
        res, err = {"answer": "", "citations": [], "input_tokens": 0, "output_tokens": 0}, True
    cited, gold = set(res["citations"]), GOLD.get(question)
    price_in, price_out = PRICES.get(model, (0.0, 0.0))
    rid = uuid.uuid4().hex[:12]
    log.record(RequestEvent(
        request_id=rid, ts=time.time(), tenant_id=p.tenant_id, flag="prompt.ask",
        variant=variant, prompt_version=version, model=model,
        latency_ms=res.get("latency_ms") or (time.perf_counter() - t0) * 1000,
        cost_usd=(res["input_tokens"] * price_in + res["output_tokens"] * price_out) / 1e6,
        error=err, passed=not err and bool(cited & DOC_SLUGS),
        citations_ok=None if gold is None else set(gold["citations"]) <= cited))
    return {**res, "prompt_version": version, "variant": variant, "request_id": rid}


TENANT_MIX = {"acme": 0.40, "zephyr": 0.25, "demo": 0.20, "orbit-maintainers": 0.15}


def replay_traffic(handler: Callable[[str, Principal, str], dict], n: int = 200,
                   users_per_tenant: int = 40, seed: int = 0) -> int:
    """Compact Day 86 replay: qa_gold questions from simulated users across tenants."""
    rng, questions = random.Random(seed), list(GOLD)
    for _ in range(n):
        tenant = rng.choices(list(TENANT_MIX), weights=list(TENANT_MIX.values()))[0]
        handler(rng.choice(questions), Principal(tenant, key_id=f"key-{tenant}"),
                f"{tenant}-u{rng.randint(1, users_per_tenant)}")
    return n

# script_id: day_087_shipping_changes_safely/atlas_release
@dataclass
class VariantStats:
    n: int
    pass_rate: float
    citation_rate: float
    p95_ms: float
    cost_per_request: float
    error_rate: float


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


def _p95(xs: list[float]) -> float:
    xs = sorted(xs)
    return xs[max(0, math.ceil(0.95 * len(xs)) - 1)] if xs else math.nan


def compare_variants(log: EventLog, flag: str, since: float) -> dict[str, VariantStats]:
    out = {}
    for variant in ("control", "canary"):
        ev = log.window(since, flag=flag, variant=variant)
        out[variant] = VariantStats(
            n=len(ev),
            pass_rate=_rate([e.passed for e in ev]),
            citation_rate=_rate([e.citations_ok for e in ev if e.citations_ok is not None]),
            p95_ms=_p95([e.latency_ms for e in ev if not e.error]),
            cost_per_request=sum(e.cost_usd for e in ev) / len(ev) if ev else math.nan,
            error_rate=_rate([e.error for e in ev]))
    return out

# script_id: day_087_shipping_changes_safely/atlas_release
LOWER_IS_BETTER = {"p95_ms", "cost_per_request", "error_rate"}


@dataclass
class Guard:
    metric: str
    max_drop: float          # rates: points (0.05 = 5 pts); p95/cost: relative (0.2 = +20%)
    min_samples: int = 50


@dataclass
class GateDecision:
    action: Literal["promote", "hold", "rollback"]
    reasons: list[str] = field(default_factory=list)

    def __str__(self) -> str:
        return f"{self.action.upper()}: " + "; ".join(self.reasons or ["all guards within limits"])


def canary_gate(stats: dict[str, VariantStats], guards: list[Guard]) -> GateDecision:
    ctl, can = stats["control"], stats["canary"]
    n = min(ctl.n, can.n)                    # at 100% there is no control left to compare with
    breached, waiting = [], []
    for g in guards:
        a, b = getattr(ctl, g.metric), getattr(can, g.metric)
        if g.metric.endswith("_rate"):       # absolute points
            worse = (b - a) if g.metric in LOWER_IS_BETTER else (a - b)
            label = f"{g.metric} {(b - a) * 100:+.1f} pts"
        else:                                # relative change vs control
            worse = (b - a) / a if a else 0.0
            label = f"{g.metric} {worse * 100:+.0f}%"
        if math.isnan(worse) or n < g.min_samples:
            if not math.isnan(worse) and worse > g.max_drop:
                waiting.append(f"{label} (n={n} < {g.min_samples})")
        elif worse > g.max_drop:
            breached.append(f"{label} (limit {g.max_drop:g}, n={n})")
    if breached:
        return GateDecision("rollback", breached)
    if n < max(g.min_samples for g in guards):
        return GateDecision("hold", waiting or [f"not enough samples (n={n})"])
    return GateDecision("promote")

# script_id: day_087_shipping_changes_safely/atlas_release
GUARDS = [Guard("pass_rate", 0.03), Guard("citation_rate", 0.05), Guard("p95_ms", 0.20),
          Guard("cost_per_request", 0.25), Guard("error_rate", 0.02)]
RELEASE_TOML = """[prompts]
ask = "v3"
[models]
ask = "gpt-6-luna"
[[canaries]]
flag = "prompt.ask"
control = "v3"
candidate = "v4"
pct = 10
unit = "user"
"""


def print_stats(stats: dict[str, VariantStats]) -> None:
    print(f"{'variant':8} {'n':>5} {'pass':>6} {'cites':>6} {'p95 ms':>7} {'$/req':>9} {'errors':>6}")
    for name, s in stats.items():
        print(f"{name:8} {s.n:5} {s.pass_rate:6.1%} {s.citation_rate:6.1%} {s.p95_ms:7.0f} "
              f"{s.cost_per_request:9.6f} {s.error_rate:6.1%}")


if __name__ == "__main__":
    path = Path("demo_release.toml")         # a demo file, so a real release.toml is never clobbered
    path.write_text(RELEASE_TOML)
    watcher, log, answer_fn = ReleaseWatcher(path), EventLog(), make_answer_fn()
    start = time.time()
    for label, n in [("after 200 requests", 200), ("after 1,200 requests", 1000)]:
        replay_traffic(lambda q, p, u: handle_ask(q, p, watcher.get(), log, answer_fn, user_id=u),
                       n=n, seed=len(log.events))
        print(f"--- {label}")
        stats = compare_variants(log, "prompt.ask", since=start)
        print_stats(stats)
        print(canary_gate(stats, GUARDS))

--- after 200 requests
variant      n   pass  cites  p95 ms     $/req errors
control    178  99.4%  95.5%    1094  0.000338   0.6%
canary      22  95.5%  63.6%     731  0.000258   4.5%
HOLD: pass_rate -4.0 pts (n=22 < 50); citation_rate -31.9 pts (n=22 < 50); error_rate +4.0 pts (n=22 < 50)
--- after 1,200 requests
variant      n   pass  cites  p95 ms     $/req errors
control   1079  98.7%  96.7%    1103  0.000336   1.3%
canary     121  98.3%  78.5%     817  0.000266   1.7%
ROLLBACK: citation_rate -18.2 pts (limit 0.05, n=121)

# script_id: day_087_shipping_changes_safely/release_api
# The Day 77 app, reduced to one flag-aware route. Needs atlas_release.py beside it.
# Install: uv add fastapi uvicorn httpx (httpx only for TestClient)
# Run: python release_api.py (in-process demo)  or  uvicorn release_api:app --port 8000
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI, Header
from pydantic import BaseModel

from atlas_release import (GOLD, RELEASE_TOML, EventLog, Principal, ReleaseWatcher,
                           handle_ask, make_answer_fn)


class AskRequest(BaseModel):
    question: str


class AskResponse(BaseModel):
    answer: str
    citations: list[str]
    prompt_version: str
    variant: str


def create_app(release_path: Path = Path("release.toml")) -> FastAPI:
    @asynccontextmanager
    async def lifespan(app: FastAPI) -> AsyncIterator[None]:
        app.state.release = ReleaseWatcher(release_path)   # a bad file fails the boot
        app.state.log, app.state.answer_fn = EventLog(), make_answer_fn()
        yield

    app = FastAPI(lifespan=lifespan)

    @app.post("/v1/ask", response_model=AskResponse)
    def ask(req: AskRequest, x_tenant: str = Header(), x_user: str = Header()) -> dict:
        # Stand-in for Day 83's require_principal: never trust tenant headers in production.
        p, s = Principal(tenant_id=x_tenant, key_id=f"key-{x_tenant}"), app.state
        return handle_ask(req.question, p, s.release.get(), s.log, s.answer_fn, user_id=x_user)

    return app


app = create_app()

# script_id: day_087_shipping_changes_safely/release_api
if __name__ == "__main__":
    from fastapi.testclient import TestClient

    demo = Path("demo_release.toml")
    demo.write_text(RELEASE_TOML)
    question = next(q for q, c in GOLD.items() if c["id"] == "qa-012")   # needs 3 docs
    with TestClient(create_app(demo)) as http:
        for tenant, user in [("acme", "acme-u1"), ("acme", "acme-u3"), ("zephyr", "zephyr-u9"),
                             ("acme", "acme-u3")]:
            body = http.post("/v1/ask", json={"question": question},
                             headers={"x-tenant": tenant, "x-user": user}).json()
            print(f"{user:10} {body['variant']:8} {body['prompt_version']}  cites={body['citations']}")

acme-u1    control  v3  cites=['troubleshooting-memory', 'concepts-workers', 'configuration-reference']
acme-u3    canary   v4  cites=['troubleshooting-memory']
zephyr-u9  canary   v4  cites=['troubleshooting-memory', 'concepts-workers']
acme-u3    canary   v4  cites=['troubleshooting-memory', 'concepts-workers']

# script_id: day_087_shipping_changes_safely/rollback
# Roll back by flipping the flag (no redeploy), log the decision, and survive a bad edit.
# Needs atlas_release.py beside it. Stdlib + pydantic.
import hashlib
import json
import re
import tempfile
import time
from pathlib import Path

from atlas_release import RELEASE_TOML, EventLog, Principal, ReleaseWatcher, handle_ask, stub_answer


def set_canary_pct(path: Path, pct: int | str) -> None:
    """Demo edit (one canary in the file). In a team: a reviewed PR or Day 88's ops CLI."""
    path.write_text(re.sub(r"(?m)^pct = .*$", f"pct = {pct}", path.read_text(), count=1))


def log_release(log_path: Path, action: str, flag: str, reason: str, actor: str) -> None:
    entry = {"ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "action": action,
             "flag": flag, "reason": reason, "actor": actor}
    with log_path.open("a") as f:            # append-only: the audit trail for every flip
        f.write(json.dumps(entry) + "\n")


def cache_key(tenant_id: str, prompt_version: str, model: str, question: str) -> str:
    """Day 85's cache scope plus release fields, so a rollback never serves v4's answers."""
    raw = f"{tenant_id}|ask|{prompt_version}|{model}|{question.strip().lower()}"
    return hashlib.sha256(raw.encode()).hexdigest()[:12]

# script_id: day_087_shipping_changes_safely/rollback
def canary_share(watcher: ReleaseWatcher, n: int = 200) -> float:
    log, p = EventLog(), Principal("acme", "key-acme")
    for i in range(n):
        handle_ask("How long does Orbit keep dead-lettered jobs?", p, watcher.get(), log,
                   stub_answer, user_id=f"acme-u{i}")
    return sum(e.variant == "canary" for e in log.events) / n


if __name__ == "__main__":
    tmp = Path(tempfile.mkdtemp())
    path, release_log = tmp / "release.toml", tmp / "release_log.jsonl"
    path.write_text(RELEASE_TOML)
    watcher = ReleaseWatcher(path, min_interval_s=0)   # check on every request for the demo
    print(f"before:         canary share {canary_share(watcher):.0%}")
    t0 = time.perf_counter()
    set_canary_pct(path, 0)
    log_release(release_log, "rollback", "prompt.ask", "citation_rate guard breached", "oncall")
    print(f"after flip:     canary share {canary_share(watcher):.0%} "
          f"({(time.perf_counter() - t0) * 1000:.0f} ms, no redeploy)")
    set_canary_pct(path, "ten")                       # a fat-fingered hand edit
    print(f"after bad edit: canary share {canary_share(watcher):.0%} | {watcher.last_error}")
    print("release log:", release_log.read_text().strip())
    k3, k4 = (cache_key("acme", v, "gpt-6-luna", "DLQ retention?") for v in ("v3", "v4"))
    print(f"cache keys: v3={k3} v4={k4} (a rollback can't hit v4's entries)")

before:         canary share 10%
after flip:     canary share 0% (49 ms, no redeploy)
after bad edit: canary share 0% | TOMLDecodeError: Invalid value (at line 9, column 7)
release log: {"ts": "2026-11-07T04:38:05Z", "action": "rollback", "flag": "prompt.ask", "reason": "citation_rate guard breached", "actor": "oncall"}
cache keys: v3=7e96deac31b9 v4=e235ac306fa3 (a rollback can't hit v4's entries)

# script_id: day_087_shipping_changes_safely/shadow_compare
# Shadow traffic: ~10% of requests are also answered by the candidate in the background.
# Install: uv add "rq>=3.0" redis   Env: REDIS_URL (optional; unset = run shadow jobs inline)
# With Redis: run `rq worker shadow --url $REDIS_URL` from this file's folder (it imports
# shadow_compare and atlas_release by name); the jobs run there instead.
import os
import uuid
from collections import Counter

from atlas_release import GOLD, bucket, load_prompt, retrieve, stub_answer

READ_ONLY_TOOLS = {"search_docs", "get_issue"}     # never label_issue, post_comment, ...


def pairwise(question: str, a: list[str], b: list[str]) -> str:
    """Code stand-in for Day 45's compare(): the answer covering more gold docs wins."""
    gold = set(GOLD[question]["citations"])
    sa, sb = len(gold & set(a)), len(gold & set(b))
    return "tie" if sa == sb else ("control" if sa > sb else "candidate")


def shadow_job(question: str, candidate: str, control_cites: list[str], tools: list[str]) -> str:
    if set(tools) - READ_ONLY_TOOLS:
        raise PermissionError(f"shadow runs may not use {sorted(set(tools) - READ_ONLY_TOOLS)}")
    try:
        cand = stub_answer(question, load_prompt("ask", candidate), "gpt-6-luna", retrieve(question))
    except TimeoutError:
        return "error"                              # shadow failures never reach the user
    return pairwise(question, control_cites, cand["citations"])


def get_queue():
    if not os.environ.get("REDIS_URL"):
        return None
    from redis import Redis
    from rq import Queue
    return Queue("shadow", connection=Redis.from_url(os.environ["REDIS_URL"]))

# script_id: day_087_shipping_changes_safely/shadow_compare
if __name__ == "__main__":
    queue, tally, shadowed = get_queue(), Counter(), 0
    questions = list(GOLD) * 10                     # 300 requests
    for q in questions:
        try:                                        # the user-facing answer
            control = stub_answer(q, load_prompt("ask", "v3"), "gpt-6-luna", retrieve(q))
        except TimeoutError:
            continue
        if bucket(uuid.uuid4().hex, "shadow.ask") >= 10:
            continue                                # ~90% are never shadowed
        shadowed += 1
        args = (q, "v4", control["citations"], ["search_docs"])
        if queue is not None:
            # By dotted name: RQ refuses functions from __main__, and the worker imports this file.
            queue.enqueue("shadow_compare.shadow_job", *args, job_timeout=60, result_ttl=3600)
        else:
            tally[shadow_job(*args)] += 1
    print(f"shadowed {shadowed}/{len(questions)} requests "
          f"(extra model spend ≈ {shadowed / len(questions):.0%})")
    print("verdicts:", dict(tally) or "enqueued; read results from the worker")
    try:
        shadow_job(questions[0], "v4", [], ["search_docs", "post_comment"])
    except PermissionError as e:
        print("PermissionError:", e)

shadowed 23/300 requests (extra model spend ≈ 8%)
verdicts: {'control': 8, 'tie': 15}
PermissionError: shadow runs may not use ['post_comment']

# script_id: day_087_shipping_changes_safely/migrate_model_gate
# Offline gate for a model swap: run the suite on control and candidate, diff, write a checklist.
# Needs atlas_release.py beside it. ATLAS_LIVE=1 makes ~60 real calls (see Cost).
# The qa_gold suite stands in for investigate's Day 59 trajectory suite here.
import os
from pathlib import Path

from atlas_release import GOLD, PRICES, _p95, load_prompt, make_answer_fn, retrieve


def run_suite(model: str, answer_fn) -> dict[str, float]:
    rows = [answer_fn(q, load_prompt("ask", "v3"), model, retrieve(q)) for q in GOLD]
    p_in, p_out = PRICES[model]
    return {"citation_rate": sum(set(c["citations"]) <= set(r["citations"])
                                 for c, r in zip(GOLD.values(), rows)) / len(rows),
            "p95_ms": _p95([r.get("latency_ms", 0.0) for r in rows]),
            "cost_per_request": sum(r["input_tokens"] * p_in + r["output_tokens"] * p_out
                                    for r in rows) / len(rows) / 1e6,
            "tokens_per_request": sum(r["input_tokens"] + r["output_tokens"] for r in rows) / len(rows)}


if __name__ == "__main__":
    fn = make_answer_fn()
    base, cand = run_suite("gpt-6-luna", fn), run_suite("gpt-6.1-terra", fn)
    print(f"{'metric':20} {'luna':>10} {'terra':>10}")
    for k in base:
        print(f"{k:20} {base[k]:10.4g} {cand[k]:10.4g}")
    peak_rpm = int(os.environ.get("ATLAS_PEAK_RPM", "120"))          # your traffic, not a fact
    tpm_limit = int(os.environ.get("ATLAS_TPM_LIMIT", "2000000"))    # read yours from Day 84
    tpm = cand["tokens_per_request"] * peak_rpm
    checks = {"offline evals: no citation drop": cand["citation_rate"] >= base["citation_rate"],
              f"capacity: {tpm:,.0f} TPM at peak <= 70% of limit": tpm <= 0.7 * tpm_limit,
              f"budget: ${cand['cost_per_request']:.4f}/request signed off": False}  # a human ticks this
    Path("migration_checklist.md").write_text(
        "\n".join(f"- [{'x' if ok else ' '}] {name}" for name, ok in checks.items()) + "\n")
    print(Path("migration_checklist.md").read_text())

metric                     luna      terra
citation_rate            0.9667          1
p95_ms                     1191       1045
cost_per_request        0.00034     0.0034
tokens_per_request         2120       2120
- [x] offline evals: no citation drop
- [x] capacity: 254,400 TPM at peak <= 70% of limit
- [ ] budget: $0.0034/request signed off

<!-- file: MIGRATION_PLAYBOOK.md -->
# Model migration: <route(s)> <old model> -> <new model>
Owner: <name> · Deadline: <provider shutdown date> · Flag: model.<route>

## 1. Inventory (exit: every use listed)
- [ ] grep for ATLAS_*MODEL, hard-coded IDs, judges, fallbacks, routers
- [ ] embeddings listed separately (index migrations: Day 88)

## 2. Offline evals (exit: no guard regression on dataset v3)
- [ ] suite run on old vs new; diff table attached
- [ ] judge model changing? re-validate against human labels (Day 46)

## 3. Prompt adjustments (exit: re-run 2 passes)
## 4. Capacity and cost (exit: TPM at peak <= 70% of limit; $/request signed off)
- [ ] limiter and fallbacks updated (Day 84); load test (Day 90)

## 5. Shadow (exit: no side effects; verdicts reviewed)
## 6. Canary -> ramp -> bake (exit: gate promotes at every stage; bake >= 1 business day)
## 7. Clean up (exit: old ID gone from code, config, judges, fallbacks)
- [ ] release log entry; flag deleted; old prompt versions archived