Reliability

# script_id: day_084_reliability_fallbacks/stacked_retries
# Install: uv add "openai>=3.46,<4"   Env: none (fake key)   Services: none
# Offline: httpx2.MockTransport plays an overloaded API, so nothing leaves your machine.
import httpx2
import openai
from openai import OpenAI

calls = 0


def always_503(request: httpx2.Request) -> httpx2.Response:
    global calls
    calls += 1
    return httpx2.Response(503, headers={"retry-after-ms": "1"},       # tiny hint keeps the demo fast
                           json={"error": {"message": "The requested model is temporarily overloaded.",
                                           "type": "service_unavailable_error", "code": "server_is_overloaded"}})


def my_retry_loop(client: OpenAI, attempts: int = 4) -> openai.APIStatusError:
    for _ in range(attempts):                        # "our" retries, stacked on top of the SDK's
        try:
            client.responses.create(model="gpt-6-luna", input="hi")
        except openai.InternalServerError as exc:
            last = exc
    return last


for sdk_retries in (3, 0):
    calls = 0
    client = OpenAI(api_key="sk-test", max_retries=sdk_retries,
                    http_client=openai.DefaultHttpxClient(transport=httpx2.MockTransport(always_503)))
    err = my_retry_loop(client)
    print(f"SDK max_retries={sdk_retries}: 4 wrapper attempts -> {calls} upstream calls (code={err.code})")

SDK max_retries=3: 4 wrapper attempts -> 16 upstream calls (code=server_is_overloaded)
SDK max_retries=0: 4 wrapper attempts -> 4 upstream calls (code=server_is_overloaded)

# script_id: day_084_reliability_fallbacks/herd_simulation
# Install: nothing (stdlib only). Simulates 50 clients that all fail at t=0.
import random
from collections import Counter

CLIENTS, ATTEMPTS, BASE, CAP, BIN = 50, 3, 0.5, 8.0, 0.5


def retry_times(jitter: bool, rng: random.Random) -> list[float]:
    t, times = 0.0, []
    for attempt in range(1, ATTEMPTS + 1):
        ceiling = min(CAP, BASE * 2 ** (attempt - 1))
        t += rng.uniform(0, ceiling) if jitter else ceiling
        times.append(t)
    return times


def histogram(jitter: bool) -> None:
    rng = random.Random(42)
    bins = Counter(int(t / BIN) for _ in range(CLIENTS) for t in retry_times(jitter, rng))
    print("full jitter" if jitter else "no jitter (fixed exponential)")
    for b in range(8):                                        # 0-4 s in half-second bins
        n = bins.get(b, 0)
        print(f"  {b * BIN:3.1f}-{(b + 1) * BIN:3.1f}s {n:3d} {'#' * (n // 2)}")


histogram(jitter=False)
histogram(jitter=True)

no jitter (fixed exponential)
  0.0-0.5s   0
  0.5-1.0s  50 #########################
  1.0-1.5s   0
  1.5-2.0s  50 #########################
  2.0-2.5s   0
  2.5-3.0s   0
  3.0-3.5s   0
  3.5-4.0s  50 #########################
full jitter
  0.0-0.5s  70 ###################################
  0.5-1.0s  29 ##############
  1.0-1.5s  23 ###########
  1.5-2.0s   9 ####
  2.0-2.5s  13 ######
  2.5-3.0s   6 ###
  3.0-3.5s   0
  3.5-4.0s   0

# script_id: day_084_reliability_fallbacks/atlas_resilience
# atlas/resilience.py: retries, breakers, rate limits, fallbacks and chaos tools (Day 84).
# Install: uv add "openai>=3.46,<4" pydantic    Env (optional): ATLAS_MODEL, ATLAS_FALLBACK_MODELS, ORBIT_DATA
# Services: none. Every demo in this file runs offline against FakeAsyncClient (defined below).
from __future__ import annotations

import asyncio
import email.utils
import enum
import json
import os
import random
import re
import time
from collections import Counter
from dataclasses import dataclass
from pathlib import Path
from types import SimpleNamespace
from typing import Any, Callable, Sequence

import httpx2
import openai
from pydantic import BaseModel

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
# Day 82's Settings.fallback_models reads this env var: comma-separated, same provider by default
FALLBACK_MODELS = tuple(m for m in os.environ.get("ATLAS_FALLBACK_MODELS", "gpt-6.1-terra").split(",") if m)
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
# 429s about money, not load (error-codes guide): retrying can't fix them
BILLING_CODES = {"insufficient_quota", "credit_balance_exhausted", "organization_spend_limit_exceeded",
                 "project_spend_limit_exceeded", "organization_usage_limit_exceeded"}


@dataclass(frozen=True)
class RetryPolicy:
    max_attempts: int = 4          # per model, counting the first try
    base_s: float = 0.5
    cap_s: float = 8.0
    honor_retry_after: bool = True
    max_wait_s: float = 20.0       # longest single sleep a user-facing call accepts; beyond it, fall back

# script_id: day_084_reliability_fallbacks/atlas_resilience
def is_retryable(exc: BaseException) -> bool:
    if isinstance(exc, (openai.APIConnectionError, TimeoutError)):   # APITimeoutError is a subclass
        return True
    if isinstance(exc, openai.APIStatusError):
        if exc.code in BILLING_CODES or exc.type == "insufficient_quota":
            return False                                             # billing: alert, don't retry
        return exc.status_code in (408, 409, 429) or exc.status_code >= 500
    return False


def retry_after_s(exc: BaseException) -> float | None:
    """The server's wait hint in seconds: retry-after-ms, then retry-after (seconds or an HTTP date)."""
    if not isinstance(exc, openai.APIStatusError):
        return None
    headers = exc.response.headers
    for name, scale in (("retry-after-ms", 0.001), ("retry-after", 1.0)):
        try:
            return max(0.0, float(headers[name]) * scale)
        except (KeyError, ValueError):
            continue
    try:
        return max(0.0, email.utils.parsedate_to_datetime(headers["retry-after"]).timestamp() - time.time())
    except (KeyError, TypeError, ValueError):
        return None


def backoff_s(attempt: int, policy: RetryPolicy, retry_after: float | None = None) -> float:
    if retry_after is not None and policy.honor_retry_after:
        return retry_after + random.uniform(0, policy.base_s)       # a minimum, plus a little spread
    return random.uniform(0, min(policy.cap_s, policy.base_s * 2 ** (attempt - 1)))   # full jitter


def fake_error(status: int, code: str, *, retry_after: str | None = None,
               type_: str = "rate_limit_error") -> openai.APIStatusError:
    """The exception the SDK would raise for this status, for tests and chaos runs."""
    cls = {400: openai.BadRequestError, 429: openai.RateLimitError}.get(
        status, openai.InternalServerError if status >= 500 else openai.APIStatusError)
    request = httpx2.Request("POST", "https://api.openai.com/v1/responses")
    response = httpx2.Response(status, headers={"retry-after": retry_after} if retry_after else {}, request=request)
    return cls(f"Error code: {status} - {code}", response=response, body={"code": code, "type": type_})


if __name__ == "__main__":
    random.seed(0)
    for label, exc in [("429 slow_down", fake_error(429, "slow_down", retry_after="2")),
                       ("429 billing", fake_error(429, "credit_balance_exhausted", type_="insufficient_quota")),
                       ("503 overloaded", fake_error(503, "server_is_overloaded")),
                       ("400 bad request", fake_error(400, "invalid_value"))]:
        hint, ok = retry_after_s(exc), is_retryable(exc)
        wait = f"{backoff_s(1, RetryPolicy(), hint):.2f}s" if ok else "-"
        print(f"{label:16} retry={ok!s:5} hint={hint} wait={wait}")

429 slow_down    retry=True  hint=2.0 wait=2.42s
429 billing      retry=False hint=None wait=-
503 overloaded   retry=True  hint=None wait=0.38s
400 bad request  retry=False hint=None wait=-

# script_id: day_084_reliability_fallbacks/atlas_resilience
class CircuitState(enum.Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"


class CircuitBreaker:
    """Per-model fuse. Methods never await, so under asyncio each call is atomic (add a Lock for threads)."""

    def __init__(self, name: str, failure_threshold: int = 5, reset_after_s: float = 30.0,
                 clock: Callable[[], float] = time.monotonic) -> None:
        self.name, self.failure_threshold, self.reset_after_s, self.clock = name, failure_threshold, reset_after_s, clock
        self.state, self.failures, self.opened_at = CircuitState.CLOSED, 0, 0.0
        self._probe_at: float | None = None          # when the half-open probe went out

    def allow(self) -> bool:
        if self.state is CircuitState.OPEN and self.clock() - self.opened_at >= self.reset_after_s:
            self._set_state(CircuitState.HALF_OPEN)
        if self.state is CircuitState.HALF_OPEN:
            # one probe at a time; a probe lost to a crash or cancel expires after reset_after_s
            if self._probe_at is not None and self.clock() - self._probe_at < self.reset_after_s:
                return False
            self._probe_at = self.clock()
            return True
        return self.state is CircuitState.CLOSED

    def record_success(self) -> None:
        self.failures, self._probe_at = 0, None
        if self.state is not CircuitState.CLOSED:
            self._set_state(CircuitState.CLOSED)

    def record_failure(self) -> None:                # call it ONLY for retryable failures
        self.failures, self._probe_at = self.failures + 1, None
        if self.state is CircuitState.HALF_OPEN or self.failures >= self.failure_threshold:
            self.opened_at = self.clock()
            if self.state is not CircuitState.OPEN:
                self._set_state(CircuitState.OPEN)

    def _set_state(self, new: CircuitState) -> None:   # one place to hook logs and metrics (Exercise 1)
        self.state = new

# script_id: day_084_reliability_fallbacks/atlas_resilience
if __name__ == "__main__":
    now = [0.0]
    breaker = CircuitBreaker("gpt-6-luna", failure_threshold=3, reset_after_s=30, clock=lambda: now[0])
    for call in range(1, 11):
        now[0] += 5
        if not breaker.allow():
            result = "skipped"
        elif now[0] < 45:
            breaker.record_failure(); result = "sent, failed"
        else:
            breaker.record_success(); result = "sent, ok"
        print(f"t={now[0]:>3.0f}s call {call:>2}: {result:13} -> {breaker.state.value}")

t=  5s call  1: sent, failed  -> closed
t= 10s call  2: sent, failed  -> closed
t= 15s call  3: sent, failed  -> open
t= 20s call  4: skipped       -> open
t= 25s call  5: skipped       -> open
t= 30s call  6: skipped       -> open
t= 35s call  7: skipped       -> open
t= 40s call  8: skipped       -> open
t= 45s call  9: sent, ok      -> closed
t= 50s call 10: sent, ok      -> closed

# script_id: day_084_reliability_fallbacks/atlas_resilience
class TokenBucket:
    """Refills at rate_per_s up to capacity; each call spends from it. One for RPM, one for TPM."""

    def __init__(self, rate_per_s: float, capacity: float) -> None:
        self.rate, self.capacity = rate_per_s, capacity
        self.tokens, self.updated = capacity, time.monotonic()
        self._lock = asyncio.Lock()                  # waiters queue in order instead of racing

    def _refill(self) -> None:
        now = time.monotonic()
        self.tokens = min(self.capacity, self.tokens + (now - self.updated) * self.rate)
        self.updated = now

    async def acquire(self, amount: float = 1) -> None:
        if amount > self.capacity:
            raise ValueError(f"call needs {amount:.0f} but the bucket only holds {self.capacity:.0f}")
        async with self._lock:
            self._refill()
            while self.tokens < amount:
                await asyncio.sleep((amount - self.tokens) / self.rate)
                self._refill()
            self.tokens -= amount


def estimate_tokens(kwargs: dict[str, Any]) -> int:
    """Rough TPM charge: ~4 characters per input token plus the whole output ceiling."""
    text = json.dumps(kwargs.get("input", "")) + str(kwargs.get("instructions") or "")
    return len(text) // 4 + int(kwargs.get("max_output_tokens") or 1_000)


if __name__ == "__main__":
    async def burst_demo() -> None:
        lines = (DATA_DIR / "eval" / "qa_gold.jsonl").read_text(encoding="utf-8").splitlines()
        questions = [json.loads(line)["question"] for line in lines]
        rpm = TokenBucket(rate_per_s=600 / 60, capacity=5)           # 600 RPM, bursts of 5
        tpm = TokenBucket(rate_per_s=120_000 / 60, capacity=2_000)   # 120K TPM
        start = time.monotonic()

        async def send(q: str) -> None:
            await rpm.acquire(1)
            await tpm.acquire(estimate_tokens({"input": q, "max_output_tokens": 300}))

        await asyncio.gather(*(send(q) for q in questions))
        print(f"{len(questions)} questions released in {time.monotonic() - start:.1f}s, paced under both limits")

    asyncio.run(burst_demo())

30 questions released in 3.8s, paced under both limits

# script_id: day_084_reliability_fallbacks/atlas_resilience
@dataclass
class CallMeta:
    model: str
    attempts: int
    fallback: bool
    degraded: bool
    latency_ms: float


class AllModelsUnavailable(RuntimeError):
    """Every model was skipped (open breaker) or ran out of attempts: switch to a degraded mode."""

    def __init__(self, models: Sequence[str], attempts: int) -> None:
        super().__init__(f"no model available after {attempts} attempts: {', '.join(models)}")
        self.attempts = attempts


class _ModelPool:
    """Per-model breakers, rate limits and counters, plus the hooks subclasses override."""

    def __init__(self, client: Any, models: Sequence[str], *, policy: RetryPolicy | None = None,
                 rpm: TokenBucket | None = None, tpm: TokenBucket | None = None,
                 failure_threshold: int = 5, reset_after_s: float = 30.0) -> None:
        self._api = client                   # the real async client (max_retries=0) in prod, a fake in tests
        self.models, self.policy = list(models), policy or RetryPolicy()
        self.rpm, self.tpm = rpm, tpm
        self.breakers = {m: CircuitBreaker(m, failure_threshold, reset_after_s) for m in self.models}
        self.upstream_calls = self.retries = 0

    async def _throttle(self, kwargs: dict[str, Any]) -> None:
        if self.rpm:
            await self.rpm.acquire(1)
        if self.tpm:
            await self.tpm.acquire(estimate_tokens(kwargs))

    async def _call_upstream(self, model: str, kwargs: dict[str, Any], is_retry: bool) -> Any:
        self.upstream_calls += 1
        self.retries += is_retry
        return await self._api.responses.create(model=model, **kwargs)

    def _allow_retry(self) -> bool:          # hook: a retry budget plugs in here (Exercise 2)
        return True

# script_id: day_084_reliability_fallbacks/atlas_resilience
class ResilientLLM(_ModelPool):
    async def create(self, **kwargs: Any) -> tuple[Any, CallMeta]:
        """responses.create with retries, breakers and fallbacks. Not for streams (see Pitfalls)."""
        start, attempts, last = time.perf_counter(), 0, None
        for i, model in enumerate(self.models):
            breaker = self.breakers[model]
            for attempt in range(1, self.policy.max_attempts + 1):
                # check the budget BEFORE allow(): allow() may hand out the single half-open probe
                if (attempt > 1 and not self._allow_retry()) or not breaker.allow():
                    break                                       # open fuse or no budget: next model
                await self._throttle(kwargs)
                attempts += 1
                try:
                    resp = await self._call_upstream(model, kwargs, is_retry=attempt > 1)
                except Exception as exc:
                    if not is_retryable(exc):
                        breaker.record_success()                # the server answered; the request is the problem
                        raise
                    breaker.record_failure()
                    last = exc
                    wait = backoff_s(attempt, self.policy, retry_after_s(exc))
                    if (breaker.state is CircuitState.OPEN or attempt == self.policy.max_attempts
                            or wait > self.policy.max_wait_s):
                        break                                   # don't sleep for nothing: next model now
                    await asyncio.sleep(wait)
                    continue
                breaker.record_success()
                ms = round((time.perf_counter() - start) * 1000, 1)
                return resp, CallMeta(model, attempts, fallback=i > 0, degraded=False, latency_ms=ms)
        raise AllModelsUnavailable(self.models, attempts) from last

# script_id: day_084_reliability_fallbacks/atlas_resilience
class FaultInjector:
    """Seeded fault plan, e.g. {"429": 0.3, "503": 0.2, "timeout": 0.1}. Models in `down` always 503."""
    CODES = {"429": "slow_down", "503": "server_is_overloaded", "500": "server_error", "400": "invalid_value"}

    def __init__(self, rates: dict[str, float], *, seed: int = 0, down: set[str] = frozenset()) -> None:
        self.rates, self.down, self.rng = rates, set(down), random.Random(seed)

    def pick(self, model: str) -> BaseException | None:
        if model in self.down:
            return fake_error(503, "server_is_overloaded", type_="service_unavailable_error")
        roll = self.rng.random()
        for kind, rate in self.rates.items():
            if roll < rate:
                if kind == "timeout":
                    return openai.APITimeoutError(request=httpx2.Request("POST", "https://api.openai.com/v1/responses"))
                return fake_error(int(kind), self.CODES.get(kind, "error"), retry_after="0.05" if kind == "429" else None)
            roll -= rate
        return None


class FakeAsyncClient:
    """Same call shape as the async OpenAI client's responses.create, offline; counts calls per model."""

    def __init__(self, faults: FaultInjector, latency_s: float = 0.01) -> None:
        self.faults, self.latency_s, self.calls = faults, latency_s, Counter()
        self.responses = self                          # so callers write fake.responses.create(...)

    async def create(self, *, model: str, **kwargs: Any) -> SimpleNamespace:
        self.calls[model] += 1
        fault = self.faults.pick(model)
        await asyncio.sleep(self.latency_s * (10 if isinstance(fault, openai.APITimeoutError) else 1))
        if fault:
            raise fault
        return SimpleNamespace(model=model, output_text=f"[{model}] answer to: {str(kwargs.get('input'))[:40]}")


if __name__ == "__main__":
    async def fallback_demo() -> None:
        fake = FakeAsyncClient(FaultInjector({"429": 0.3}, seed=3, down={MODEL}))
        llm = ResilientLLM(fake, [MODEL, *FALLBACK_MODELS], policy=RetryPolicy(base_s=0.01, cap_s=0.1),
                           failure_threshold=3)
        for q in ["How do leases work?", "What does ORB-1301 mean?", "How do I tune heartbeat_seconds?"]:
            _, meta = await llm.create(input=q, max_output_tokens=300)
            print(f"{meta.model:14} attempts={meta.attempts} fallback={meta.fallback} "
                  f"luna_breaker={llm.breakers[MODEL].state.value} upstream={dict(fake.calls)}")

    asyncio.run(fallback_demo())

gpt-6.1-terra  attempts=5 fallback=True luna_breaker=open upstream={'gpt-6-luna': 3, 'gpt-6.1-terra': 2}
gpt-6.1-terra  attempts=1 fallback=True luna_breaker=open upstream={'gpt-6-luna': 3, 'gpt-6.1-terra': 3}
gpt-6.1-terra  attempts=1 fallback=True luna_breaker=open upstream={'gpt-6-luna': 3, 'gpt-6.1-terra': 4}

# script_id: day_084_reliability_fallbacks/atlas_resilience
class Citation(BaseModel):            # compact copies of Day 15's schemas; R5 imports atlas.schemas
    doc_id: str
    quote: str


class Answer(BaseModel):
    text: str
    citations: list[Citation]
    degraded: bool = False            # new today: tells the UI to show a "limited answer" banner


class DocChunk(BaseModel):
    id: str
    doc_id: str
    section: str
    applies_to: str
    text: str


STOP = {"the", "and", "how", "what", "does", "for", "with", "when", "which", "can", "you", "are", "orbit"}


def terms(text: str) -> set[str]:
    return {w for w in re.findall(r"[a-z0-9_]+(?:[.-][a-z0-9_]+)*", text.lower()) if len(w) >= 3} - STOP


def load_chunks(data_dir: Path = DATA_DIR) -> list[DocChunk]:
    """One chunk per '## ' section of each Orbit doc (Day 25 chunks better; this is enough to cite)."""
    chunks = []
    for path in sorted((data_dir / "docs").glob("*.md")):
        _, front, body = path.read_text(encoding="utf-8").split("---\n", 2)
        meta = dict(line.split(": ", 1) for line in front.strip().splitlines() if ": " in line)
        for n, part in enumerate(body.split("\n## ")):
            title, _, text = part.strip().partition("\n")
            chunks.append(DocChunk(id=f"{meta['id']}#{n}", doc_id=meta["id"], section=title.strip("# \n"),
                                   applies_to=meta.get("applies_to", ""), text=text.strip()))
    return chunks


def keyword_search(question: str, chunks: list[DocChunk], k: int = 3) -> list[DocChunk]:
    """Term overlap needs no model call, so it still works when every provider is down."""
    q = terms(question)

    def score(c: DocChunk) -> int:               # a match in the doc id or heading counts double
        return 2 * len(q & terms(c.doc_id.replace("-", " ") + " " + c.section)) + len(q & terms(c.text))

    return sorted(chunks, key=score, reverse=True)[:k]

# script_id: day_084_reliability_fallbacks/atlas_resilience
def degraded_answer(question: str, hits: list[DocChunk]) -> Answer:
    """Retrieval-only answer: no generation, only quotes and links the user can check."""
    hits = [h for h in hits if terms(question) & terms(h.section + " " + h.text)]
    if not hits:
        return Answer(text="Atlas can't reach its language models right now. Please try again in a few minutes.",
                      citations=[], degraded=True)
    def first_paragraph(text: str) -> str:      # skip tables, code and headings
        prose = (p for p in text.split("\n\n") if p.strip() and p.lstrip()[0] not in "|`#-*>")
        return next(prose, text).replace("\n", " ")[:160]

    cites = [Citation(doc_id=h.doc_id, quote=first_paragraph(h.text)) for h in hits]
    lines = [f"- {h.doc_id} > {h.section}: {c.quote}" for h, c in zip(hits, cites)]
    return Answer(text="Atlas can't generate an answer right now. The most relevant Orbit docs:\n" + "\n".join(lines),
                  citations=cites, degraded=True)


async def resilient_ask(llm: ResilientLLM, question: str, chunks: list[DocChunk]) -> tuple[Answer, CallMeta]:
    """Compact stand-in for Day 28's ask(): a model answer if possible, retrieval-only if not."""
    start = time.perf_counter()
    try:
        resp, meta = await llm.create(input=question, max_output_tokens=300)
        return Answer(text=resp.output_text, citations=[]), meta
    except AllModelsUnavailable as exc:
        ms = round((time.perf_counter() - start) * 1000, 1)
        meta = CallMeta("none", exc.attempts, fallback=True, degraded=True, latency_ms=ms)
        return degraded_answer(question, keyword_search(question, chunks)), meta


if __name__ == "__main__":
    async def degraded_demo() -> None:
        fake = FakeAsyncClient(FaultInjector({}, down={MODEL, *FALLBACK_MODELS}))
        llm = ResilientLLM(fake, [MODEL, *FALLBACK_MODELS], policy=RetryPolicy(base_s=0.01, cap_s=0.1))
        answer, meta = await resilient_ask(llm, "How do leases and heartbeats work?", load_chunks())
        print(meta)
        print(answer.text)

    asyncio.run(degraded_demo())

CallMeta(model='none', attempts=8, fallback=True, degraded=True, latency_ms=159.7)
Atlas can't generate an answer right now. The most relevant Orbit docs:
- leases-and-heartbeats > Heartbeat loss: ORB-1102: If the broker stops receiving a worker's heartbeats, Orbit records **ORB-1102 WorkerHeartbeatLost** and the dashboard marks the worker as lost. Its jobs are red
- leases-and-heartbeats > Leases and heartbeats: Leases let Orbit recover jobs from workers that crash, freeze or lose the network. They are also the main reason a job can **run twice**.
- leases-and-heartbeats > Heartbeats: A fixed deadline would force you to set the lease longer than your slowest job. Heartbeats avoid that. Every `heartbeat_seconds`, the worker supervisor renews t

# script_id: day_084_reliability_fallbacks/test_resilience
# Install: uv add pytest "openai>=3.46,<4" pydantic    Run: pytest test_resilience.py -q
# Imports atlas_resilience.py from the same folder (R5: from atlas.resilience import ...).
import asyncio

import openai
import pytest

from atlas_resilience import (AllModelsUnavailable, CircuitBreaker, CircuitState, FakeAsyncClient,
                              FaultInjector, ResilientLLM, RetryPolicy, backoff_s, degraded_answer,
                              fake_error, keyword_search, load_chunks, retry_after_s)

FAST = RetryPolicy(max_attempts=3, base_s=0.001, cap_s=0.01)


def make(rates: dict | None = None, down: set = frozenset()) -> tuple[FakeAsyncClient, ResilientLLM]:
    fake = FakeAsyncClient(FaultInjector(rates or {}, down=set(down)), latency_s=0)
    return fake, ResilientLLM(fake, ["primary", "backup"], policy=FAST, failure_threshold=3)


def test_breaker_opens_and_stops_traffic():
    fake, llm = make(down={"primary"})
    for _ in range(5):
        asyncio.run(llm.create(input="q"))
    assert llm.breakers["primary"].state is CircuitState.OPEN
    assert fake.calls["primary"] == 3              # all three during request 1; later requests skip it


def test_half_open_allows_exactly_one_probe():
    now = [0.0]
    breaker = CircuitBreaker("m", failure_threshold=1, reset_after_s=10, clock=lambda: now[0])
    breaker.allow(); breaker.record_failure()
    now[0] = 10
    assert [breaker.allow() for _ in range(3)] == [True, False, False]


def test_no_retry_and_no_breaker_trip_on_400():
    fake, llm = make(rates={"400": 1.0})
    with pytest.raises(openai.BadRequestError):
        asyncio.run(llm.create(input="q"))
    assert fake.calls == {"primary": 1} and llm.breakers["primary"].state is CircuitState.CLOSED


def test_retry_after_is_honoured():
    exc = fake_error(429, "slow_down", retry_after="2")
    assert retry_after_s(exc) == 2.0
    assert 2.0 <= backoff_s(1, FAST, retry_after_s(exc)) <= 2.0 + FAST.base_s


def test_fallback_is_used_and_labelled():
    _, llm = make(down={"primary"})
    _, meta = asyncio.run(llm.create(input="q"))
    assert meta.model == "backup" and meta.fallback and not meta.degraded


def test_degraded_when_everything_is_down():
    _, llm = make(down={"primary", "backup"})
    with pytest.raises(AllModelsUnavailable):
        asyncio.run(llm.create(input="q"))
    answer = degraded_answer("How do leases work?", keyword_search("How do leases work?", load_chunks()))
    assert answer.degraded and answer.citations


if __name__ == "__main__":
    raise SystemExit(pytest.main([__file__, "-q", "-p", "no:cacheprovider"]))

......                                                                   [100%]
6 passed

# script_id: day_084_reliability_fallbacks/live_smoke
# Install: uv add "openai>=3.46,<4" pydantic    Env: OPENAI_API_KEY, ATLAS_MODEL, ATLAS_FALLBACK_MODELS (optional)
# Cost: one short call, well under $0.001 on gpt-6-luna (as of 2026-11, verify at the pricing page).
import asyncio

from openai import AsyncOpenAI

from atlas_resilience import FALLBACK_MODELS, MODEL, ResilientLLM, TokenBucket


async def main() -> None:
    # max_retries=0: ResilientLLM owns retries. timeout=30: a user-facing call shouldn't wait the default 600 s.
    async with AsyncOpenAI(max_retries=0, timeout=30.0) as client:
        llm = ResilientLLM(client, [MODEL, *FALLBACK_MODELS],
                           rpm=TokenBucket(rate_per_s=500 / 60, capacity=10),       # example limits: use yours
                           tpm=TokenBucket(rate_per_s=200_000 / 60, capacity=20_000))
        resp, meta = await llm.create(input="In one sentence, why do job queues use leases?",
                                      max_output_tokens=200, reasoning={"effort": "none"})
        print(resp.output_text)
        print(meta)


asyncio.run(main())

A lease lets a worker claim a job for a limited time, so if the worker dies the job is released and retried.
CallMeta(model='gpt-6-luna', attempts=1, fallback=False, degraded=False, latency_ms=812.4)

# script_id: day_084_reliability_fallbacks/gateway_litellm
# Install: uv add "litellm==1.108.1" mcp   Pin it: 1.82.7 and 1.82.8 were compromised on PyPI. Commit uv.lock.
# (In our 1.108.1 test the Responses path imported `mcp`, hence the extra package.)
# Env: none. Services: none. Offline: the primary is a closed port, the backup is mocked.
import sys

try:
    from litellm import Router
except ImportError:
    print("litellm is not installed: uv add 'litellm==1.108.1' mcp")
    sys.exit(0)

router = Router(
    model_list=[
        {"model_name": "primary", "litellm_params": {"model": "openai/gpt-6-luna", "api_key": "sk-test",
                                                     "api_base": "http://127.0.0.1:9/v1"}},   # nothing listens
        {"model_name": "backup", "litellm_params": {"model": "openai/gpt-6.1-terra", "api_key": "sk-test",
                                                    "mock_response": "answer from backup"}},
    ],
    fallbacks=[{"primary": ["backup"]}], num_retries=0, timeout=5)

for name in ("backup", "primary"):
    try:
        print(f"router.responses(model={name!r}) -> {router.responses(model=name, input='hi').output_text}")
    except Exception as exc:
        print(f"router.responses(model={name!r}) -> no fallback: {type(exc).__name__}")

router.responses(model='backup') -> answer from backup
router.responses(model='primary') -> no fallback: InternalServerError