Incident: A provider change broke a quarter of requests

import os
import sys

if os.path.basename(os.getcwd()) != "northdesk-platform":   # not in the repository yet
    if "google.colab" in sys.modules:                         # in Colab: the repository lives in your Drive
        from google.colab import drive
        drive.mount("/content/drive")
        os.chdir("/content/drive/MyDrive")
    os.makedirs("northdesk-platform", exist_ok=True)
    os.chdir("northdesk-platform")
print("working in", os.path.basename(os.getcwd()))

from pathlib import Path
from IPython.core.magic import register_cell_magic

WROTE = set()                              # files this session's %%ensure wrote

@register_cell_magic
def ensure(line, cell):
    """%%ensure path: save the cell unless the file exists. %%ensure -a path: add to a file this session wrote."""
    *flags, name = line.split()
    path, text = Path(name), cell.rstrip("\n") + "\n"
    if flags == ["-a"]:
        if name in WROTE:
            with path.open("a", encoding="utf-8") as f:
                f.write("\n\n" + text)
            print("added to", path)
        return
    if path.exists():
        WROTE.discard(name)                # a kept file gets no added parts either
        print("kept", path)
        return
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(text, encoding="utf-8")
    WROTE.add(name)
    print("wrote", path)

%%ensure northdesk/__init__.py
"""Northdesk platform: the design, and the code that proves it (course: AI System Design)."""

%%ensure northdesk/trace.py
"""Northdesk call log, version 1: one JSON line per model call. Module 9 turns it into span trees."""
import json
import os
import time
import uuid
from datetime import datetime, timezone

LOG_PATH = "logs/calls.jsonl"


def logged_ask(ask, model, messages, *, feature, tenant_id, model_name, prompt_version, path=LOG_PATH):
    """Call the model through ask() and log the call, whether it succeeded or not."""
    start = time.perf_counter()
    reply, err = ask(model, messages)
    usage = (getattr(reply, "usage_metadata", None) or {}) if reply is not None else {}
    record = {"call_id": uuid.uuid4().hex[:12], "ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
              "feature": feature, "tenant_id": tenant_id, "model": model_name, "prompt_version": prompt_version,
              "input": [list(m) for m in messages], "output": None if err else reply.text,
              "input_tokens": usage.get("input_tokens"), "output_tokens": usage.get("output_tokens"),
              "latency_ms": round((time.perf_counter() - start) * 1000), "status": err or "ok"}
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")
    return reply, err


def read_calls(path=LOG_PATH):
    """Every logged call, oldest first."""
    with open(path, encoding="utf-8") as f:
        return [json.loads(line) for line in f]

def ask(model, messages):
    """Returns (reply, None), or (None, reason) after two refusals."""
    for attempt in range(2):                     # one retry: the next reply is usually fine
        try:
            return model.invoke(messages), None
        except Exception as err:
            last = err
    return None, f"provider_error: {type(last).__name__}"

import os
import random
from collections import Counter
from types import SimpleNamespace
from northdesk.trace import logged_ask, read_calls

ANSWER = SimpleNamespace(text="You get 30 days of annual leave a year. [t-tallis-d002]",
                         usage_metadata={"input_tokens": 2958, "output_tokens": 363})


class Primary:
    """Scripted stand-in for the primary provider (simulated). After `changed_at` requests, a change on the
    provider's side breaks `share` of new requests; a retry of a broken request fails the same way."""
    def __init__(self, changed_at, share=0.24, seed=20261005):
        self.changed_at, self.share, self.rng = changed_at, share, random.Random(seed)
        self.requests, self.last, self.broken = 0, None, False

    def invoke(self, messages):
        if messages is not self.last:          # a new request, not ask()'s retry of the same one
            self.last, self.requests = messages, self.requests + 1
            self.broken = self.requests > self.changed_at and self.rng.random() < self.share
        if self.broken:
            raise ConnectionError("upstream error")
        return ANSWER


class Secondary:
    """Scripted stand-in for the secondary provider: it always answers (simulated)."""
    def invoke(self, messages):
        return ANSWER

def replay(answer, log, n=600):
    """n Ask requests from Tallis staff, each a new message list, answered by `answer(messages, log)`."""
    if os.path.exists(log):
        os.remove(log)
    return [answer([("user", "How many days of annual leave do I get?")], log) for _ in range(n)]


primary = Primary(changed_at=300)
direct = lambda messages, log: logged_ask(ask, primary, messages, feature="ask", tenant_id="t-tallis",
                                          model_name="primary", prompt_version="v8", path=log)
results = replay(direct, "logs/incident-before.jsonl")
calls = read_calls("logs/incident-before.jsonl")
failed = [i for i, (reply, err) in enumerate(results) if err]
print(len(failed), "of", len(results), "requests failed; the first at request", failed[0] + 1)
print(f"after the change: {sum(i >= 300 for i in failed) / 300:.1%} failed | before it: {sum(i < 300 for i in failed)}")
print("what changed at Northdesk:", {c["model"] for c in calls}, {c["prompt_version"] for c in calls})

79 of 600 requests failed; the first at request 303
after the change: 26.3% failed | before it: 0
what changed at Northdesk: {'primary'} {'v8'}

def first_alert(calls, model, window=50, limit=0.05):
    """How many of `model`'s calls had been made when over `limit` of its last `window` failed; None if never."""
    recent, seen = [], 0
    for c in calls:
        if c["model"] != model:
            continue
        seen += 1
        recent = (recent + [c["status"].startswith("provider_error")])[-window:]
        if len(recent) == window and sum(recent) / window > limit:
            return seen
    return None


alert = first_alert(calls, "primary")
print("alert at primary call", alert, "=", alert - 300, "requests after the change")
print(f"at the shared traffic's busiest five minutes (2.40 requests a second): {(alert - 300) / 2.40:.0f} s")

alert at primary call 306 = 6 requests after the change
at the shared traffic's busiest five minutes (2.40 requests a second): 2 s

def ask_with_fallback(messages, log, chain):
    """Try each (name, model) in turn through the call log; return the first answer, or the last error."""
    for name, model in chain:
        reply, err = logged_ask(ask, model, messages, feature="ask", tenant_id="t-tallis",
                                model_name=name, prompt_version="v8", path=log)
        if err is None:
            return reply, None
    return None, err


chain = [("primary", Primary(changed_at=300)), ("secondary", Secondary())]
fixed = replay(lambda messages, log: ask_with_fallback(messages, log, chain), "logs/incident-after.jsonl")
after = read_calls("logs/incident-after.jsonl")
print(sum(err is not None for _, err in fixed), "of", len(fixed), "requests failed")
print("calls by model:", dict(Counter(c["model"] for c in after)),
      "| alert at primary call", first_alert(after, "primary"))

0 of 600 requests failed
calls by model: {'primary': 600, 'secondary': 79} | alert at primary call 306

def test_provider_change_is_caught_and_contained():
    chain = [("primary", Primary(changed_at=100)), ("secondary", Secondary())]
    results = replay(lambda messages, log: ask_with_fallback(messages, log, chain), "logs/incident-test.jsonl", n=300)
    calls = read_calls("logs/incident-test.jsonl")
    assert first_alert(calls[:100], "primary") is None              # quiet before the change
    alert = first_alert(calls, "primary")
    assert alert is not None and alert - 100 <= 50, alert           # caught within one window
    assert all(err is None for _, err in results)                   # and no user saw it


test_provider_change_is_caught_and_contained()
print("passed")

passed