Lesson 2: What still carries over from classical design

import random
from datetime import date, datetime, timedelta

COURSE_DATE = date(2026, 10, 18)   # versions, model names and prices checked on this day
DATA_AS_OF = date(2026, 10, 5)     # the snapshot: count days to this, never to the clock
WINDOW = (date(2026, 9, 7), date(2026, 10, 4))   # 28 days of traffic and traces, Monday to Sunday
SEED = 20261005

EVENTS = {   # fixed dates the data and the lessons share (times are UTC)
    "tallis_onboarded": date(2026, 9, 14),              # Tallis Retail Group switches Ask on for all stores
    "primary_degraded": (datetime(2026, 9, 22, 13, 0), datetime(2026, 9, 22, 13, 40)),
    "prompt_v8": datetime(2026, 9, 29, 10, 0),          # Ask prompt v7 -> v8: longer answers, 12 chunks not 8
}

FLEET = {   # the whole Northdesk fleet in October 2026 (fixed facts, simulated)
    "tenants": 240, "seats": 410_000, "ask_dau": 38_000, "ask_requests_per_dau": 6.2,
    "cases_runs_per_day": 9_000, "documents": 1_100_000, "avg_tokens_per_document": 6_000,
    "seat_price_usd_month": {"ask": 8.00, "cases": 4.00},
}

TARGET = {   # Joel's design target for Q2 2027: the roadmap's worked estimate
    "ask_dau": 100_000, "requests_per_user": 10, "input_tokens": 2_000, "output_tokens": 500,
    "target_seconds": 3.0, "peak_factor": 4, "cases_runs_per_day": 60_000,
}

TENANTS = [   # the six tenants in the sample data (fictional)
    # id, name, country, residency, seats, utc_offset_hours, currency, isolation, features
    ("t-fairhaven", "Fairhaven Housing", "UK", "uk", 2_400, 1, "GBP", "shared", ("ask",)),
    ("t-brennan", "Brennan & Kloss Logistics", "DE", "eu", 9_800, 2, "EUR", "shared", ("ask", "cases")),
    ("t-halcyon", "Halcyon Health Partners", "US", "us", 31_000, -4, "USD", "siloed", ("ask",)),
    ("t-tallis", "Tallis Retail Group", "UK", "uk", 54_000, 1, "GBP", "shared", ("ask", "cases")),
    ("t-oriel", "Oriel Software", "US", "us", 1_200, -7, "USD", "shared", ("ask", "cases")),
    ("t-marlow", "Marlow County Council", "UK", "uk", 6_500, 1, "GBP", "hybrid", ("ask",)),
]
TENANT_IDS = [t[0] for t in TENANTS]
STAFF = ["Helen", "Joel", "Funmi", "Ana", "Leila", "Gareth"]

GROUPS = ["all-staff", "hr", "it", "finance", "facilities", "operations", "managers", "hr-confidential", "execs"]
DEPARTMENTS = ["hr", "it", "finance", "facilities", "operations"]

FIRST = ["Aisha", "Ben", "Chloe", "Dmitri", "Elena", "Femi", "Grace", "Hamza", "Isla", "Jonas", "Keiko", "Liam",
         "Maya", "Nikhil", "Olga", "Pedro", "Quinn", "Rania", "Sven", "Tara", "Umar", "Vera", "Wei", "Yara"]
LAST = ["Abbott", "Bakshi", "Costa", "Dunne", "Eriksen", "Fofana", "Gill", "Horvath", "Iqbal", "Jansen", "Kaur",
        "Lopes", "Moreau", "Nwosu", "Ortiz", "Park", "Quigley", "Rossi", "Sato", "Tiwari", "Ueda", "Vance", "Weber", "Zielinski"]


def make_users(seed=SEED):
    """60 users per tenant: department, groups and role. Emails use the reserved .example domain."""
    rng = random.Random(seed + 1)
    users = []
    for tid, name, *_ in TENANTS:
        domain = tid[2:] + ".example"
        for i in range(60):
            dept = DEPARTMENTS[i % 5]
            groups = ["all-staff", dept]
            role = "employee"
            if i % 6 == 0:
                groups.append("managers"); role = "manager"
            if dept == "hr" and i % 3 == 0:
                groups.append("hr-confidential")
            if i in (0, 1):
                groups.append("execs"); role = "executive"
            if dept == "it" and i % 4 == 1:
                role = "it-admin"
            first, last = rng.choice(FIRST), rng.choice(LAST)
            users.append({"user_id": f"{tid}-u{i:03d}", "tenant_id": tid, "name": f"{first} {last}",
                          "email": f"{first.lower()}.{last.lower()}{i}@{domain}", "department": dept,
                          "role": role, "groups": sorted(set(groups))})
    return users


USERS = make_users()
print(len(USERS), "users;", sum("hr-confidential" in u["groups"] for u in USERS), "can read hr-confidential;",
      USERS[0]["user_id"], USERS[0]["groups"])

HOUR_WEIGHT = [0.02] * 7 + [0.3, 1.0, 1.6, 1.3, 1.1, 0.8, 1.0, 1.2, 1.0, 0.8, 0.5, 0.25] + [0.08] * 5   # local hour
DAY_WEIGHT = [1.2, 1.0, 1.0, 1.0, 0.8, 0.07, 0.07]   # Monday to Sunday
ASK_RATE = FLEET["ask_dau"] / FLEET["seats"] * FLEET["ask_requests_per_dau"]   # Ask requests per seat per working day
CASES_RATE = 0.03                                                              # Cases runs per seat per working day


def make_traffic(seed=SEED):
    """Requests per 5-minute bucket (UTC) for each tenant and feature over the 28-day window."""
    rng = random.Random(seed + 4)
    rows = []
    start = datetime(WINDOW[0].year, WINDOW[0].month, WINDOW[0].day)
    for b in range(28 * 288):
        ts = start + timedelta(minutes=5 * b)
        for tid, name, country, res, seats, off, cur, iso, feats in TENANTS:
            local = ts + timedelta(hours=off)
            share = HOUR_WEIGHT[local.hour] / sum(HOUR_WEIGHT) / 12 * DAY_WEIGHT[local.weekday()]
            active = seats
            if tid == "t-tallis":   # 10% of stores before 14 September, then a three-day ramp to all
                ramp = (ts.date() - EVENTS["tallis_onboarded"]).days
                active = seats * (0.1 if ramp < 0 else min(1.0, 0.1 + 0.3 * (ramp + 1)))
            for feature in feats:
                mu = active * (ASK_RATE if feature == "ask" else CASES_RATE) * share
                count = max(0, round(rng.gauss(mu, mu ** 0.5))) if mu > 0 else 0
                rows.append((ts.strftime("%Y-%m-%dT%H:%M"), tid, feature, count))
    return rows


TRAFFIC = make_traffic()
ask_rows = [r for r in TRAFFIC if r[2] == "ask"]
per_bucket = {}
for ts, tid, feature, count in ask_rows:
    per_bucket[ts] = per_bucket.get(ts, 0) + count
total = sum(per_bucket.values())
peak_ts = max(per_bucket, key=per_bucket.get)
avg_rps, peak_rps = total / (28 * 86_400), per_bucket[peak_ts] / 300
print(len(TRAFFIC), "rows;", total, "Ask requests in 28 days;", f"average {avg_rps:.2f} RPS,",
      f"peak {peak_rps:.2f} RPS at {peak_ts} UTC, peak factor {peak_rps / avg_rps:.1f}x")

def make_traces(n_ask=1200, n_cases=240, seed=SEED):
    """Sampled request traces: each is a tree of spans with per-hop latency (ms) and token counts."""
    rng = random.Random(seed + 5)
    weights = {}
    for ts, tid, feature, count in TRAFFIC:
        weights[(ts, tid, feature)] = count
    deg_start, deg_end = EVENTS["primary_degraded"]

    def sample(feature, k, inside=None):
        keys = sorted(x for x in weights if x[2] == feature and (inside is None or inside(x[0])))
        return rng.choices(keys, weights=[weights[x] for x in keys], k=k)

    in_incident = lambda ts: deg_start <= datetime.fromisoformat(ts) < deg_end
    # the incident window is kept at a higher rate, as a tail-sampling tracer keeps slow and failed requests
    ask_keys = sorted(sample("ask", n_ask) + sample("ask", 120, in_incident))
    case_keys = sorted(sample("cases", n_cases))
    traces = []

    def span(spans, name, parent, start, dur, **attrs):
        sid = f"s{len(spans) + 1:02d}"
        spans.append({"span_id": sid, "parent_id": parent, "name": name, "start_ms": round(start),
                      "duration_ms": max(1, round(dur)), "attrs": attrs})
        return sid, start + dur

    def model_call(spans, parent, t, ts, tier, tokens, long_output):
        degraded = deg_start <= ts < deg_end
        provider = "primary"
        if degraded and rng.random() < 0.3:   # the primary provider times out at the gateway's 8-second limit
            _, t = span(spans, "model.call", parent, t, 8000, provider="primary", model_tier=tier, status="timeout", **tokens)
            provider, tier = "secondary", "small"
        out = max(40, min(900, round(rng.gauss(*long_output))))
        tps = rng.gauss(160, 25) if tier == "small" else rng.gauss(52, 8)
        ttft = rng.uniform(120, 250) + tokens["input_tokens"] / 6000 * 1000
        queue = rng.uniform(0, 40)
        if degraded and provider == "primary":
            ttft, queue = ttft * 4, queue + rng.uniform(200, 900)
        reasoning = round(rng.uniform(0, 300)) if tier == "large" else 0
        decode = (out + reasoning) / tps * 1000
        _, t = span(spans, "model.call", parent, t, queue + ttft + decode, provider=provider, model_tier=tier,
                    status="ok", queue_ms=round(queue), ttft_ms=round(ttft), decode_ms=round(decode),
                    output_tokens=out, reasoning_tokens=reasoning, **tokens)
        return t, provider

    for i, (ts_s, tid, _f) in enumerate(ask_keys):
        ts = datetime.fromisoformat(ts_s) + timedelta(seconds=rng.randrange(300))
        v8 = ts >= EVENTS["prompt_v8"]
        spans = []
        root, _ = span(spans, "ask.request", None, 0, 0)
        _, t = span(spans, "gateway.auth", root, 0, rng.uniform(5, 15))
        _, t = span(spans, "session.load", root, t, rng.uniform(3, 10))
        ret, _ = span(spans, "retrieve", root, t, 0)
        _, t1 = span(spans, "retrieve.embed", ret, t, rng.uniform(15, 40))
        _, t2 = span(spans, "retrieve.vector", ret, t1, rng.uniform(20, 60))
        _, t3 = span(spans, "retrieve.bm25", ret, t, rng.uniform(8, 25))
        k = 12 if v8 else 8
        found = rng.random() > 0.03
        _, t = span(spans, "retrieve.rerank", ret, max(t2, t3), rng.uniform(60, 180), candidates=40, kept=k if found else 0)
        spans[int(ret[1:]) - 1]["duration_ms"] = round(t - spans[int(ret[1:]) - 1]["start_ms"])
        _, t = span(spans, "prompt.build", root, t, rng.uniform(2, 6))
        turns = rng.choice([0, 0, 1, 1, 2, 3, 4])
        system = 700 if v8 else 650
        context = k * 160 if found else 0
        user = rng.randrange(20, 80)
        tokens = {"system_tokens": system, "tool_schema_tokens": 0, "history_tokens": turns * 180,
                  "context_tokens": context, "user_tokens": user}
        tokens["input_tokens"] = sum(tokens.values())
        tokens["cached_tokens"] = system if rng.random() < 0.6 else 0
        tier = "small" if rng.random() < 0.6 else "large"
        mean_out = ((365, 110) if v8 else (260, 90)) if found else (45, 5)
        t, provider = model_call(spans, root, t, ts, tier, tokens, mean_out)
        _, t = span(spans, "validate.output", root, t, rng.uniform(5, 20))
        _, t = span(spans, "session.save", root, t, rng.uniform(3, 8))
        spans[0]["duration_ms"] = round(t)
        status = "no_context" if not found else ("fallback" if provider == "secondary" else "ok")
        traces.append({"trace_id": f"ask-{i + 1:04d}", "ts": ts.isoformat(), "tenant_id": tid, "feature": "ask",
                       "prompt_version": "v8" if v8 else "v7", "status": status, "spans": spans})

    tools = [("tool.hris.get_employee", 80, 300), ("tool.assets.get_device", 100, 250),
             ("tool.tickets.create", 150, 400), ("tool.hris.get_leave_balance", 80, 250)]
    for i, (ts_s, tid, _f) in enumerate(case_keys):
        ts = datetime.fromisoformat(ts_s) + timedelta(seconds=rng.randrange(300))
        spans = []
        root, t = span(spans, "cases.run", None, 0, 0)
        steps = rng.choice([2, 3, 3, 4, 4, 5, 6])
        outcome = rng.choices(["resolved", "approval_requested", "escalated", "stopped_max_steps"], [70, 18, 10, 2])[0]
        history = 0
        for s in range(steps):
            step, _ = span(spans, "agent.step", root, t, 0, step=s + 1)
            step_start = t
            tokens = {"system_tokens": 900, "tool_schema_tokens": 1400, "history_tokens": history,
                      "context_tokens": 0, "user_tokens": rng.randrange(20, 80)}
            tokens["input_tokens"] = sum(tokens.values())
            tokens["cached_tokens"] = 2300 if rng.random() < 0.7 else 0
            t, _p = model_call(spans, step, t, ts, "large", tokens, (120, 40))
            name, lo, hi = rng.choice(tools)
            _, t = span(spans, name, step, t, rng.uniform(lo, hi))
            history += 350
            spans[int(step[1:]) - 1]["duration_ms"] = round(t - step_start)
        if outcome == "approval_requested":
            _, t = span(spans, "approval.wait", root, t, rng.uniform(2, 40) * 60_000)
        spans[0]["duration_ms"] = round(t)
        traces.append({"trace_id": f"cases-{i + 1:04d}", "ts": ts.isoformat(), "tenant_id": tid, "feature": "cases",
                       "prompt_version": "c3", "status": outcome, "spans": spans})
    return traces


TRACES = make_traces()
ask_tr = [t for t in TRACES if t["feature"] == "ask"]
lat = sorted(t["spans"][0]["duration_ms"] for t in ask_tr)
print(len(TRACES), "traces;", len(ask_tr), "Ask;", f"Ask p50 {lat[len(lat) // 2]} ms, p95 {lat[int(len(lat) * 0.95)]} ms;",
      {s: sum(t["status"] == s for t in ask_tr) for s in ("ok", "fallback", "no_context")})

import os
import statistics

os.makedirs("northdesk/basics", exist_ok=True)       # %%writefile does not create folders
for path in ("northdesk/__init__.py", "northdesk/basics/__init__.py"):
    open(path, "a").close()

ask_traces = [t for t in TRACES if t["feature"] == "ask"]
mean_s = statistics.mean(t["spans"][0]["duration_ms"] for t in ask_traces) / 1000
target_avg = TARGET["ask_dau"] * TARGET["requests_per_user"] / 86_400
target_peak = target_avg * TARGET["peak_factor"]
print(f"Ask mean latency {mean_s:.1f} s (simulated)")
print(f"busiest five minutes: {peak_rps:.2f} RPS x {mean_s:.1f} s = {peak_rps * mean_s:.0f} in flight")
print(f"target: {target_avg:.1f} RPS average, {target_peak:.1f} at peak")
print(f"in flight at the target peak: {target_peak * mean_s:.0f} at {mean_s:.1f} s, {target_peak * 0.2:.0f} at 0.2 s")

Ask mean latency 5.5 s (simulated)
busiest five minutes: 2.40 RPS x 5.5 s = 13 in flight
target: 11.6 RPS average, 46.3 at peak
in flight at the target peak: 255 at 5.5 s, 9 at 0.2 s

%%writefile northdesk/basics/replicas.py
"""Replicas behind a round-robin load balancer, simulated in one process (no network)."""
from itertools import cycle


class Replica:
    def __init__(self, name, store=None):
        self.name = name
        self.local = {}      # in-process memory: lost on restart, invisible to the other replicas
        self.store = store   # shared session store (a dict standing in for Redis); None = keep state locally
        self.up = True

    def handle(self, session_id, question):
        sessions = self.local if self.store is None else self.store
        history = sessions.setdefault(session_id, [])
        seen = len(history)  # earlier turns this replica can see
        history.append(question)
        return self.name, seen


class Balancer:
    """Send each request to the next healthy replica in turn."""
    def __init__(self, replicas):
        self.replicas = replicas
        self.order = cycle(replicas)

    def route(self, session_id, question):
        for _ in range(len(self.replicas)):
            replica = next(self.order)
            if replica.up:
                return replica.handle(session_id, question)
        raise RuntimeError("no healthy replica")

from northdesk.basics.replicas import Replica, Balancer

TURNS = ["What is the daily meal limit?", "Does that apply abroad?", "And on a two-day trip?",
         "Who approves the claim?", "How long do I have to submit it?", "Can I claim a taxi?"]
SESSION = "t-brennan-u002/s1"   # Aisha Moreau's conversation (added in this lesson)

for label, store in (("state in each replica", None), ("shared session store", {})):
    replicas = [Replica(name, store) for name in "abc"]
    balancer = Balancer(replicas)
    results = []
    for turn, question in enumerate(TURNS):
        if turn == 3:
            replicas[1].up = False                     # replica b fails after the third turn
        results.append(balancer.route(SESSION, question))
    full = sum(seen == turn for turn, (_, seen) in enumerate(results))
    print(f"{label:22} {results}  full history on {full} of {len(TURNS)} turns")

state in each replica  [('a', 0), ('b', 0), ('c', 0), ('a', 1), ('c', 1), ('a', 2)]  full history on 1 of 6 turns
shared session store   [('a', 0), ('b', 1), ('c', 2), ('a', 3), ('c', 4), ('a', 5)]  full history on 6 of 6 turns

%%writefile northdesk/basics/queue_sim.py
"""A queue in front of a fixed number of model-call slots, simulated in time (seconds)."""
import heapq
import random
from collections import deque


def simulate(rps, seconds, slots, service_times, bound=None, seed=0):
    """Arrivals at `rps` for `seconds`; each holds one of `slots` for a sampled service time.
    bound: most requests allowed to wait; arrivals beyond it are rejected at once (backpressure)."""
    rng = random.Random(seed)
    free = [0.0] * slots                  # when each slot is next free
    waiting = deque()                     # start times of accepted requests not yet started
    waits, rejected, t = [], 0, 0.0
    while True:
        t += rng.expovariate(rps)         # next arrival (random, at the given average rate)
        if t > seconds:
            break
        while waiting and waiting[0] <= t:
            waiting.popleft()
        if bound is not None and len(waiting) >= bound:
            rejected += 1                 # a fast "try again" instead of a long wait
            continue
        start = max(t, heapq.heappop(free))
        heapq.heappush(free, start + rng.choice(service_times))
        if start > t:
            waiting.append(start)
        waits.append(start - t)
    waits.sort()
    p95 = waits[int(len(waits) * 0.95)] if waits else 0.0
    return {"accepted": len(waits), "rejected": rejected, "p95_wait_s": round(p95, 1)}

from northdesk.basics.queue_sim import simulate

service = [t["spans"][0]["duration_ms"] / 1000 for t in ask_traces]   # seconds each, from the traces
SLOTS, BOUND = 16, 16
for multiple in (0.5, 1, 1.25, 1.5, 2):
    rps = peak_rps * multiple
    open_queue = simulate(rps, 300, SLOTS, service, bound=None, seed=SEED)
    bounded = simulate(rps, 300, SLOTS, service, bound=BOUND, seed=SEED)
    print(f"{multiple:>4}x peak ({rps:.2f} RPS)  unbounded p95 wait {open_queue['p95_wait_s']:>6} s"
          f"  |  bounded p95 wait {bounded['p95_wait_s']:>4} s, rejected {bounded['rejected']}")

 0.5x peak (1.20 RPS)  unbounded p95 wait    0.0 s  |  bounded p95 wait  0.0 s, rejected 0
   1x peak (2.40 RPS)  unbounded p95 wait    2.6 s  |  bounded p95 wait  2.6 s, rejected 0
1.25x peak (3.00 RPS)  unbounded p95 wait    8.8 s  |  bounded p95 wait  5.4 s, rejected 45
 1.5x peak (3.60 RPS)  unbounded p95 wait   59.0 s  |  bounded p95 wait  6.5 s, rejected 172
   2x peak (4.81 RPS)  unbounded p95 wait  176.7 s  |  bounded p95 wait  7.0 s, rejected 508

%%writefile northdesk/basics/retry.py
"""Timeouts, retries with jittered backoff and an idempotency key, on a simulated clock (ms)."""


class SlowTickets:
    """Scripted stand-in for tickets.create: the first call creates the ticket, but replies too late."""
    def __init__(self, slow_ms=9_000, fast_ms=200):
        self.tickets, self.by_key, self.calls = [], {}, 0
        self.slow_ms, self.fast_ms = slow_ms, fast_ms

    def create(self, summary, key=None):
        self.calls += 1
        if key in self.by_key:                     # seen this key: return the first result, do nothing new
            return self.by_key[key], self.fast_ms
        ticket = f"TCK-{len(self.tickets) + 1:03d}"
        self.tickets.append((ticket, summary))     # the side effect happens whether or not we wait for it
        if key is not None:
            self.by_key[key] = ticket
        return ticket, self.slow_ms if self.calls == 1 else self.fast_ms


def call_with_retry(call, timeout_ms, attempts, rng, base_ms=250):
    """Try up to `attempts` times; give up waiting on each try at timeout_ms. Returns (result, ms, tries)."""
    clock = 0.0
    for n in range(attempts):
        result, took_ms = call()
        if took_ms <= timeout_ms:
            return result, round(clock + took_ms), n + 1
        clock += timeout_ms                        # we stopped waiting; the work may still have been done
        clock += rng.uniform(0, base_ms * 2 ** n)  # backoff with full jitter before the next try
    return None, round(clock), attempts

import sqlite3

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE spans (trace_id TEXT, ts TEXT, span_id TEXT, parent_id TEXT,"
           " name TEXT, duration_ms INTEGER, status TEXT)")
db.executemany("INSERT INTO spans VALUES (?, ?, ?, ?, ?, ?, ?)",
               [(t["trace_id"], t["ts"], s["span_id"], s["parent_id"], s["name"], s["duration_ms"],
                 s["attrs"].get("status")) for t in ask_traces for s in t["spans"]])
print(db.execute("SELECT COUNT(DISTINCT trace_id), COUNT(*) FROM spans").fetchone(), "traces and span rows")

QUERY = """SELECT substr(ts, 12, 4) || '0' AS window, COUNT(*), SUM(status = 'timeout')
           FROM spans WHERE name = 'model.call' AND ts >= '2026-09-22T12:30' AND ts < '2026-09-22T14:10'
           GROUP BY window ORDER BY window"""
for window, calls, timeouts in db.execute(QUERY):
    print(f"{window} UTC  model calls {calls:>2}  timeouts {timeouts:>2}", "ALERT" if timeouts / calls > 0.05 else "")

(1320, 15871) traces and span rows
12:30 UTC  model calls  1  timeouts  0 
12:50 UTC  model calls  1  timeouts  0 
13:00 UTC  model calls 42  timeouts  8 ALERT
13:10 UTC  model calls 47  timeouts 10 ALERT
13:20 UTC  model calls 36  timeouts  8 ALERT
13:30 UTC  model calls 30  timeouts  5 ALERT
13:40 UTC  model calls  1  timeouts  0 
13:50 UTC  model calls  4  timeouts  0 

import random
from northdesk.basics.retry import SlowTickets, call_with_retry

for key in (None, "t-brennan-u002/cases-req-1"):
    tool = SlowTickets()
    result = call_with_retry(lambda: tool.create("cracked laptop screen", key=key),
                             timeout_ms=8_000, attempts=3, rng=random.Random(SEED))
    print(f"key={key}: returned {result[0]} after {result[1]} ms in {result[2]} tries;",
          f"tickets in the system: {[t for t, _ in tool.tickets]}")

key=None: returned TCK-002 after 8269 ms in 2 tries; tickets in the system: ['TCK-001', 'TCK-002']
key=t-brennan-u002/cases-req-1: returned TCK-001 after 8269 ms in 2 tries; tickets in the system: ['TCK-001']

def check_cache_key(fn):
    q, docs = "What is the daily meal limit?", {"t-brennan-d003": 1}
    base = fn("t-brennan", q, "v8", docs)
    assert base == fn("t-brennan", "  what is the DAILY meal limit? ", "v8", docs), "spacing or case changed the key"
    assert base != fn("t-tallis", q, "v8", {"t-brennan-d003": 1}), "two tenants share a key: a cross-tenant leak"
    assert base != fn("t-brennan", q, "v7", docs), "the prompt version is missing from the key"
    assert base != fn("t-brennan", q, "v8", {"t-brennan-d003": 2}), "a document version is missing from the key"
    print("all four checks passed")

import hashlib
import json


def cache_key(tenant_id, question, prompt_version, doc_versions):
    normalised = " ".join(question.lower().split())
    parts = [tenant_id, normalised, prompt_version, sorted(doc_versions.items())]
    return hashlib.sha256(json.dumps(parts).encode()).hexdigest()


check_cache_key(cache_key)