How to stop one user running up your AI bill

"""A budget for each signed-in user, counted in money and checked before each AI call. Python standard library only.

Prices are the example prices from lesson 7 ($2 and $10 per million tokens), not a quote. Money is counted in
micro-dollars (millionths of a dollar) as whole numbers, so $2 per million tokens is exactly 2 micro-dollars per token.

A call goes through three steps. reserve() sets aside the most the call can cost and refuses if that does not fit.
settle() charges what the call really cost and gives the rest back. release() gives everything back when the service answered with an error.
Time is a parameter (seconds) so a test can move it without waiting.
"""
import math
import threading
from dataclasses import dataclass

PRICE_IN = 2  # micro-dollars per input token
PRICE_OUT = 10  # micro-dollars per output token
MAX_REPLY = 1000  # the reply cap the server sends with every call (lesson 9's max_tokens)
DAY = 86_400  # seconds
MESSAGE_OVERHEAD = 10  # tokens added per message by the chat format: an assumption; check your provider's counter


def cost(input_tokens, output_tokens):
    """Micro-dollars a call costs."""
    return input_tokens * PRICE_IN + output_tokens * PRICE_OUT


def dollars(micro):
    """Text such as "$0.25". Whole cents, halves round up (widget.html does the same sum)."""
    cents = (micro + 5_000) // 10_000
    return f"${cents // 100}.{cents % 100:02d}"


def how_long(seconds):
    """Text such as "9 h 20 min", rounded up to a whole minute."""
    hours, minutes = divmod(max(1, math.ceil(seconds / 60)), 60)
    return f"{hours} h {minutes} min" if hours else f"{minutes} min"


def input_tokens_upper_bound(messages):
    """A count that never falls short for the text in `messages`: a token covers at least one byte of text, so bytes >= tokens (plus the per-message extra). The caller adds the system prompt and tool definitions."""
    return sum(len(m["content"].encode("utf-8")) + MESSAGE_OVERHEAD for m in messages)


class OverBudget(Exception):
    """reserve() refused. `left` is what is still free, `wait` the seconds until the call fits (None: it never will)."""

    def __init__(self, amount, left, wait):
        super().__init__(f"needs {amount}, {left} left")
        self.amount, self.left, self.wait = amount, left, wait


@dataclass(frozen=True)
class Hold:
    user: str
    amount: int  # micro-dollars set aside
    period: int  # the day it was set aside in (steady refill: always 0)


class Budget:
    """Each user may use `limit` micro-dollars. refill="daily": the count restarts at 00:00 UTC.
    refill="steady": the count drains by limit/86400 per second, so the whole budget is back after a day of no use."""

    def __init__(self, limit, refill="daily"):
        if refill not in ("daily", "steady"):
            raise ValueError("refill is daily or steady")
        self.limit, self.refill = limit, refill
        self._state = {}  # user -> (used, stamp); stamp is the day number (daily) or the time of the last update (steady)
        self._lock = threading.Lock()  # one process only: with two processes the state must live in one shared store

    def _fresh(self, user, now):
        """(used, stamp) for this user brought up to `now`. Call it with the lock held."""
        used, stamp = self._state.get(user, (0, None))
        if self.refill == "daily":
            day = int(now // DAY)
            return (used if stamp == day else 0), day
        if stamp is not None:
            used = max(0, used - (now - stamp) * self.limit / DAY)
        return used, now

    def used(self, user, now):
        with self._lock:
            return self._fresh(user, now)[0]

    def reserve(self, user, amount, now):
        """Set `amount` aside, or raise OverBudget. The check and the set-aside happen under one lock, so two calls cannot both take the last of it."""
        with self._lock:
            used, stamp = self._fresh(user, now)
            if used + amount > self.limit:
                excess = used + amount - self.limit
                if amount > self.limit:
                    wait = None  # bigger than the whole budget: waiting never helps
                elif self.refill == "daily":
                    wait = DAY - now % DAY
                else:
                    wait = excess * DAY / self.limit
                raise OverBudget(amount, max(0, self.limit - used), wait)
            self._state[user] = (used + amount, stamp)
            return Hold(user, amount, stamp if self.refill == "daily" else 0)

    def settle(self, hold, actual, now):
        """Replace the set-aside with the real cost. If the real cost is higher, the user is charged it all and may go over."""
        with self._lock:
            used, stamp = self._fresh(hold.user, now)
            if self.refill == "daily" and hold.period != stamp:
                return  # the day this call was set aside in has ended, and its count with it
            self._state[hold.user] = (max(0, used + actual - hold.amount), stamp)

    def release(self, hold, now):
        """The service answered with an error, or never got the call: give the whole set-aside back."""
        self.settle(hold, 0, now)


class CheckOnlyBudget(Budget):
    """The first version most people write: look at what is already charged, make the call, charge afterwards. Nothing is set aside."""

    def reserve(self, user, amount, now):
        return super().reserve(user, 0, now)


def refusal(refused, limit, now):
    """The JSON body for a refused call: a code the app can test, a message a person can read, and when the budget is back."""
    if refused.wait is None:
        return {"error": "request_too_big",
                "message": f"This request could cost up to {dollars(refused.amount)}, more than your whole {dollars(limit)} budget. Send less text."}
    return {"error": "budget_used_up", "resets_at": int(now + refused.wait),
            "message": f"Your AI budget has {dollars(refused.left)} left and this call needs up to {dollars(refused.amount)}. "
                       f"It refills in {how_long(refused.wait)}."}


def handle_chat(budget, upstream, user, request, now, max_reply=MAX_REPLY, count_tokens=input_tokens_upper_bound):
    """Run one chat call for a signed-in user. Returns (status, body).

    `user` is the account id from your sign-in check, never a value the browser sends. `upstream(request, max_reply)`
    calls the AI service and returns {"text": ..., "usage": {"input_tokens": ..., "output_tokens": ...}}. On a
    timeout it must raise Python's own TimeoutError, so map your HTTP library's timeout error to it.
    """
    if user is None:
        return 401, {"error": "sign_in", "message": "Sign in to use the assistant."}
    amount = cost(count_tokens(request["messages"]), max_reply)  # the most this call can cost
    try:
        hold = budget.reserve(user, amount, now)
    except OverBudget as refused:
        return 429, refusal(refused, budget.limit, now)  # no short Retry-After: it tells clients to retry soon, and the budget is back in hours
    try:
        answer = upstream(request, max_reply)
    except TimeoutError:
        # the provider may still bill a call that timed out, so keep the set-aside: the safe side.
        # It stays counted until the daily reset, or drains with a steady refill.
        return 502, {"error": "upstream_timeout", "message": "The AI service did not answer in time. The call may still count against your budget."}
    except Exception:
        budget.release(hold, now)  # the service answered with an error, so give the set-aside back
        return 502, {"error": "upstream_failed", "message": "The AI service failed. Try again in a moment."}
    usage = answer["usage"]
    actual = cost(usage["input_tokens"], usage["output_tokens"])
    budget.settle(hold, actual, now)
    return 200, {"text": answer["text"], "charged": actual}

"""Keep the data inside ../widget.html equal to the data in sim.py and budget.py.

    python3 embed_widget.py          rewrite the data block in widget.html
    python3 embed_widget.py --check  exit 1 if widget.html differs (test_widget.py runs the same check)
"""
import json
import re
import sys
from pathlib import Path

import sim
from budget import DAY, MAX_REPLY, PRICE_IN, PRICE_OUT

WIDGET = Path(__file__).resolve().parent.parent / "widget.html"
CONFIG_BLOCK = re.compile(r'(<script type="application/json" id="qb-config">)(.*?)(</script>)', re.S)


def config():
    return {"users": sim.USERS, "day": DAY, "seed": sim.SEED, "rpm": sim.RPM, "budget": sim.BUDGET, "start": sim.START_RULE,
            "priceIn": PRICE_IN, "priceOut": PRICE_OUT, "maxReply": MAX_REPLY,
            "heavyFrom": sim.HEAVY_FROM, "heavyTo": sim.HEAVY_TO, "heavyIn": sim.HEAVY_IN, "heavyOut": sim.HEAVY_OUT, "heavyGap": sim.HEAVY_GAP,
            "ordinaryIn": sim.ORDINARY_IN, "ordinaryOut": sim.ORDINARY_OUT, "sessions": sim.SESSIONS, "sessionCalls": sim.SESSION_CALLS,
            "callGap": sim.CALL_GAP, "longestSession": sim.LONGEST_SESSION, "rules": [{"id": i, "name": name} for i, name in sim.RULES]}


def embed(html):
    """html with the config block rewritten from sim.py."""
    block = "{\n" + ",\n".join(f" {json.dumps(key)}: {json.dumps(value)}" for key, value in config().items()) + "\n}"
    return CONFIG_BLOCK.sub(lambda m: m.group(1) + "\n" + block + "\n" + m.group(3), html, count=1)


if __name__ == "__main__":
    current = WIDGET.read_text(encoding="utf-8")
    updated = embed(current)
    if "--check" in sys.argv:
        sys.exit(0 if updated == current else 1)
    WIDGET.write_text(updated, encoding="utf-8")
    print("widget.html data block now matches sim.py")

"""Many calls in flight at once, against a budget with the set-aside and one without. Python standard library only.

fire() starts all the calls together. Each call that gets past the budget waits inside the (fake) AI service until every
other call has either got in or been refused, so no call has been charged yet when the last one is checked.
"""
import threading
import time

from budget import MAX_REPLY, handle_chat

CALLS = 20  # calls started at the same moment by one user
BURST_INPUT = 20_000  # input tokens in each call
BURST_OUTPUT = 800  # output tokens each reply really has (the cap is 1,000)
USER = "user-1"


def fire(budget, calls=CALLS, estimate=BURST_INPUT, input_tokens=BURST_INPUT, output_tokens=BURST_OUTPUT):
    """Run `calls` chat calls at once. Returns (statuses, charged micro-dollars). `estimate` is the input count the server believes."""
    gate, lock = threading.Event(), threading.Lock()
    inside = [0]
    statuses = [None] * calls
    request = {"messages": [{"role": "user", "content": "a long document"}]}

    def upstream(req, max_reply):
        with lock:
            inside[0] += 1
        if not gate.wait(10):
            raise TimeoutError("the gate never opened")
        return {"text": "ok", "usage": {"input_tokens": input_tokens, "output_tokens": output_tokens}}

    def call(i):
        statuses[i] = handle_chat(budget, upstream, USER, request, 0.0, MAX_REPLY, lambda messages: estimate)[0]

    threads = [threading.Thread(target=call, args=(i,)) for i in range(calls)]
    for thread in threads:
        thread.start()
    deadline = time.monotonic() + 10
    while inside[0] + sum(s is not None for s in statuses) < calls:  # every call is either inside or already refused
        if time.monotonic() > deadline:
            gate.set()
            raise TimeoutError("calls did not settle into inside or refused")
        time.sleep(0.001)
    gate.set()
    for thread in threads:
        thread.join()
    return statuses, budget.used(USER, 0.0)

"""Print the numbers the lesson quotes. Run: python3 show_numbers.py"""
from budget import Budget, CheckOnlyBudget, cost, dollars
from inflight import BURST_INPUT, fire
from sim import RULES, SEED, simulate_day

LIMIT = 250_000

print(f"A short question (60 in, 30 out) costs {cost(60, 30)} micro-dollars; a long document (50,000 in, 2,000 out) costs {cost(50_000, 2_000)}: "
      f"{cost(50_000, 2_000) / cost(60, 30):.1f} times as much.")
print(f"\nOne simulated day (seed {SEED}), ten users, one heavy:")
for rule, name in RULES:
    day = simulate_day(rule)
    print(f"  {name:20} bill {dollars(day.bill)}  heavy share {day.percent(0)}%  heavy calls refused {day.refused[0]} of {day.served[0] + day.refused[0]}"
          f"  ordinary calls refused {day.ordinary_refused} of {day.ordinary_asked}  ordinary users {min(day.percent(u) for u in range(1, 10))}-{max(day.percent(u) for u in range(1, 10))}% each")
print(f"\n20 calls at once, each costing ${cost(BURST_INPUT, 800) / 1e6:.3f}; the budget is {dollars(LIMIT)}:")
for label, budget, kwargs in (("check only", CheckOnlyBudget(LIMIT), {}), ("set aside first", Budget(LIMIT), {}), ("set aside, input guessed at half", Budget(LIMIT), {"estimate": BURST_INPUT // 2})):
    statuses, charged = fire(budget, **kwargs)
    print(f"  {label:34} admitted {statuses.count(200):2}  refused {statuses.count(429):2}  charged {dollars(charged)} ({charged / LIMIT:.2f} times the budget)")

"""One day of AI calls from ten users, one of them heavy, under three rules. Python standard library only.

A toy, not a model of any real app. Ordinary users chat in a few short sessions; the heavy user sends long documents
all day, one call every few minutes. The three rules differ only in what the server checks before it makes a call:
nothing, a number of calls per minute for each user, or a daily budget in money for each user. A refused call is lost
(the user is told and does not retry). The same seed always gives the same day; widget.html has the same code.
"""
from dataclasses import dataclass

from budget import DAY, MAX_REPLY, Budget, OverBudget, cost

USERS = 10  # user 0 is the heavy user, users 1 to 9 are ordinary
HEAVY = 0
SEED = 7  # the day the lesson shows
RPM = 5  # calls per minute each user may make under the "requests per minute" rule
BUDGET = 250_000  # micro-dollars each user may spend in a day under the budget rule ($0.25)
HEAVY_FROM, HEAVY_TO = 6 * 3600, 22 * 3600  # the heavy user works from 06:00 to 22:00
RULES = (("none", "No limit"), ("rpm", "Requests per minute"), ("budget", "Daily cost budget"))
START_RULE = "rpm"  # the widget opens on this rule

# What one call is made of, as (lowest, highest) token counts. Ordinary users: 3 to 5 sessions a day of 4 to 8 calls, 20 to 89 s apart.
ORDINARY_IN, ORDINARY_OUT = (300, 900), (80, 280)
HEAVY_IN, HEAVY_OUT = (16_000, 24_000), (700, 1000)
HEAVY_GAP = (240, 360)  # seconds between the heavy user's calls
SESSIONS, SESSION_CALLS, CALL_GAP = (3, 5), (4, 8), (20, 89)
LONGEST_SESSION = 8 * 89  # seconds; a session starts early enough to end inside the day

MASK = 0xFFFFFFFF


def random_numbers(seed):
    """Endless numbers in [0, 1), the same on every machine (the mulberry32 generator, as in lesson 10; widget.html has the same code)."""
    state = seed & MASK
    while True:
        state = (state + 0x6D2B79F5) & MASK
        t = ((state ^ (state >> 15)) * (state | 1)) & MASK
        t ^= (t + (((t ^ (t >> 7)) * (t | 61)) & MASK)) & MASK
        yield ((t ^ (t >> 14)) & MASK) / 4294967296


class TokenBucket:
    """From lesson 9: holds up to `capacity` tickets and gains `rate` tickets per second. A call spends one."""

    def __init__(self, capacity, rate, now):
        self.capacity, self.rate = capacity, rate
        self.level, self.last = float(capacity), now

    def take(self, now):
        """Spend one ticket if there is one. Return True if the call is allowed."""
        self.level = min(self.capacity, self.level + (now - self.last) * self.rate)
        self.last = now
        if self.level >= 1:
            self.level -= 1
            return True
        return False


def between(draw, low, high):
    """A whole number from low to high, both included."""
    return low + int(draw() * (high - low + 1))


def make_day(seed):
    """Every call of the day as (second, user, input tokens, output tokens), in time order."""
    draw = random_numbers(seed).__next__
    calls = []
    t = HEAVY_FROM
    while t < HEAVY_TO:
        calls.append((t, HEAVY, between(draw, *HEAVY_IN), between(draw, *HEAVY_OUT)))
        t += between(draw, *HEAVY_GAP)
    for user in range(1, USERS):
        for _ in range(between(draw, *SESSIONS)):
            t = between(draw, 0, DAY - LONGEST_SESSION)
            for _ in range(between(draw, *SESSION_CALLS)):
                calls.append((t, user, between(draw, *ORDINARY_IN), between(draw, *ORDINARY_OUT)))
                t += between(draw, *CALL_GAP)
    return sorted(calls, key=lambda c: (c[0], c[1]))  # equal times keep the order they were made in


@dataclass(frozen=True)
class Day:
    rule: str
    served: tuple  # per user: calls answered
    refused: tuple  # per user: calls refused
    spent: tuple  # per user: micro-dollars charged

    @property
    def bill(self):
        return sum(self.spent)

    def percent(self, user):
        """The user's share of the bill in whole percent, halves round up (whole-number arithmetic, so JavaScript agrees)."""
        return (200 * self.spent[user] + self.bill) // (2 * self.bill) if self.bill else 0

    @property
    def ordinary_refused(self):
        return sum(self.refused[1:])

    @property
    def ordinary_asked(self):
        return sum(self.served[1:]) + self.ordinary_refused


def simulate_day(rule, seed=SEED, rpm=RPM, budget=BUDGET):
    served, refused, spent = [0] * USERS, [0] * USERS, [0] * USERS
    buckets = [TokenBucket(rpm, rpm / 60, 0) for _ in range(USERS)]
    wallet = Budget(budget)
    for t, user, tokens_in, tokens_out in make_day(seed):
        allowed = True
        if rule == "rpm":
            allowed = buckets[user].take(t)
        elif rule == "budget":
            try:
                wallet.settle(wallet.reserve(user, cost(tokens_in, MAX_REPLY), t), cost(tokens_in, tokens_out), t)
            except OverBudget:
                allowed = False
        if allowed:
            served[user] += 1
            spent[user] += cost(tokens_in, tokens_out)
        else:
            refused[user] += 1
    return Day(rule, tuple(served), tuple(refused), tuple(spent))

"""Tests for budget.py and inflight.py. Each test names the failure it guards."""
import re
import textwrap
import threading
import time
from pathlib import Path

import pytest

from budget import (DAY, MAX_REPLY, Budget, CheckOnlyBudget, OverBudget, cost, dollars, handle_chat, how_long, input_tokens_upper_bound, refusal)
from inflight import BURST_INPUT, BURST_OUTPUT, CALLS, fire
from sim import simulate_day

LIMIT = 250_000  # $0.25
REQUEST = {"messages": [{"role": "user", "content": "hello"}]}


def ok_upstream(tokens_in=100, tokens_out=50):
    return lambda request, max_reply: {"text": "hi", "usage": {"input_tokens": tokens_in, "output_tokens": tokens_out}}


def test_one_call_can_cost_hundreds_of_times_another():
    """The lesson's reason a request limit is not a cost limit: a short question and a long document, same price list."""
    short, long_ = cost(60, 30), cost(50_000, 2_000)
    assert (short, long_) == (420, 120_000)
    assert 285 < long_ / short < 286


def test_reserve_allows_an_exact_fit_and_refuses_one_micro_dollar_more():
    """Guards >= instead of > in the check: a user could never spend the last of the budget."""
    budget = Budget(1000)
    budget.reserve("a", 1000, 0)
    with pytest.raises(OverBudget):
        budget.reserve("a", 1, 0)
    with pytest.raises(OverBudget):
        Budget(1000).reserve("a", 1001, 0)


def test_settle_charges_the_real_cost_and_gives_back_the_rest():
    budget = Budget(LIMIT)
    hold = budget.reserve("a", 50_000, 0)
    assert budget.used("a", 0) == 50_000
    budget.settle(hold, 48_000, 0)
    assert budget.used("a", 0) == 48_000


def test_a_reply_that_costs_more_than_the_set_aside_is_charged_in_full():
    """Guards a settle that clamps to the hold: a wrong estimate would then be free, and the budget would drift low."""
    budget = Budget(100)
    hold = budget.reserve("a", 60, 0)
    budget.settle(hold, 90, 0)
    assert budget.used("a", 0) == 90
    budget.settle(budget.reserve("a", 10, 0), 30, 0)  # 90 + 10 fits; the real cost takes the user to 120
    assert budget.used("a", 0) == 120
    with pytest.raises(OverBudget):
        budget.reserve("a", 1, 0)


def test_release_gives_back_the_whole_set_aside():
    """Guards a failed call that keeps its hold: the user would pay for nothing."""
    budget = Budget(LIMIT)
    budget.release(budget.reserve("a", 50_000, 0), 0)
    assert budget.used("a", 0) == 0


def test_users_have_separate_budgets():
    """Guards one shared counter: one heavy user would lock everyone out."""
    budget = Budget(100)
    budget.reserve("heavy", 100, 0)
    budget.reserve("light", 100, 0)
    with pytest.raises(OverBudget):
        budget.reserve("heavy", 1, 0)


def test_daily_budget_restarts_at_midnight_and_a_hold_from_yesterday_is_ignored():
    """Guards a settle that lands after the reset: it would subtract yesterday's hold from today's fresh count and go negative."""
    budget = Budget(100)
    hold = budget.reserve("a", 50, DAY - 5)
    budget.settle(hold, 80, DAY + 3)  # a call that started before midnight ends after it
    assert budget.used("a", DAY + 3) == 0
    old = budget.reserve("b", 50, DAY - 5)
    budget.reserve("b", 40, DAY + 1)
    budget.release(old, DAY + 3)  # and a call that failed after midnight gives nothing back from today's count
    assert budget.used("b", DAY + 3) == 40
    today = budget.reserve("c", 50, DAY + 10)  # a call set aside and settled on the same day after the first midnight still settles
    budget.settle(today, 30, DAY + 20)
    assert budget.used("c", DAY + 20) == 30
    budget.settle(today, 0, DAY + 30)  # a hold settled twice must not take the count below zero and bank free credit
    assert budget.used("c", DAY + 30) == 0


def test_steady_refill_drains_by_limit_over_a_day_and_says_how_long_to_wait():
    budget = Budget(86_400, refill="steady")  # 1 micro-dollar back per second
    budget.reserve("a", 86_400, 0)
    assert budget.used("a", 1000) == 85_400
    with pytest.raises(OverBudget) as refused:
        budget.reserve("a", 2000, 1000)  # 1,000 is free, so the call is 1,000 too big
    assert refused.value.wait == 1000  # at 1 micro-dollar back per second
    budget.reserve("a", 1000, 1000)
    assert budget.used("a", 1000 + DAY) == 0
    budget.reserve("a", 86_400, 1000 + 10 * DAY)  # ten idle days bank no credit: the whole budget is free and not a unit more
    with pytest.raises(OverBudget):
        budget.reserve("a", 1, 1000 + 10 * DAY)


def test_a_call_bigger_than_the_whole_budget_is_told_to_send_less_not_to_wait():
    """Guards a message that says "try again later" for a call that can never fit."""
    status, body = handle_chat(Budget(LIMIT), ok_upstream(), "a", {"messages": [{"role": "user", "content": "x" * 200_000}]}, 0)
    assert status == 429 and body["error"] == "request_too_big" and "resets_at" not in body
    assert "Send less text" in body["message"]


def test_refused_call_gets_a_clear_message_and_the_refill_time_not_a_bare_429():
    """Lesson 10: a used-up quota is not a retry case. The body says so in words and in a code the app can test; there is no retry hint."""
    budget = Budget(LIMIT)
    now = 14 * 3600 + 40 * 60  # 14:40, so 9 h 20 min to midnight
    budget.reserve("a", 245_000, now)
    status, body = handle_chat(budget, ok_upstream(), "a", REQUEST, now)
    assert status == 429
    assert body["error"] == "budget_used_up"
    assert body["resets_at"] == DAY
    assert body["message"] == "Your AI budget has $0.01 left and this call needs up to $0.01. It refills in 9 h 20 min."
    assert not any("retry" in key for key in body)


def test_a_user_who_is_not_signed_in_gets_401_and_costs_nothing():
    calls = []
    status, body = handle_chat(Budget(LIMIT), lambda *a: calls.append(a), None, REQUEST, 0)
    assert (status, body["error"], calls) == (401, "sign_in", [])


def test_an_answered_call_is_charged_the_usage_numbers_not_the_estimate():
    budget = Budget(LIMIT)
    status, body = handle_chat(budget, ok_upstream(100, 50), "a", REQUEST, 0)
    assert (status, body["charged"]) == (200, cost(100, 50))
    assert budget.used("a", 0) == cost(100, 50)


def test_a_failed_upstream_call_costs_the_user_nothing_and_a_timed_out_one_keeps_its_set_aside():
    """Guards a missing release on the error path (each outage would eat the user's budget) and a release on a timeout (the provider may still bill that call)."""
    budget = Budget(LIMIT)

    def broken(request, max_reply):
        raise ConnectionError("down")

    def slow(request, max_reply):
        raise TimeoutError("no answer")

    status, body = handle_chat(budget, broken, "a", REQUEST, 0)
    assert status == 502 and body["error"] == "upstream_failed" and "Nothing was charged" not in body["message"]
    assert budget.used("a", 0) == 0
    status, body = handle_chat(budget, slow, "a", REQUEST, 0)
    assert status == 502 and body["error"] == "upstream_timeout" and "Nothing was charged" not in body["message"]
    assert budget.used("a", 0) == cost(input_tokens_upper_bound(REQUEST["messages"]), MAX_REPLY)


def test_the_set_aside_is_input_plus_the_reply_cap_priced_as_output():
    seen = {}
    budget = Budget(LIMIT)
    handle_chat(budget, lambda request, max_reply: seen.setdefault("held", budget.used("a", 0)) and ok_upstream()(request, max_reply),
                "a", REQUEST, 0, count_tokens=lambda messages: 700)
    assert seen["held"] == cost(700, MAX_REPLY)


def test_the_upper_bound_never_falls_below_the_real_count():
    """A token covers at least one byte of text, so bytes plus the per-message extra is a safe ceiling (English runs near 4 bytes a token)."""
    messages = [{"role": "user", "content": "naïve café 你好"}, {"role": "assistant", "content": "ok"}]
    bound = input_tokens_upper_bound(messages)
    assert bound == len("naïve café 你好".encode()) + len(b"ok") + 2 * 10
    assert bound >= len("naïve café 你好")


@pytest.mark.parametrize("seconds, text", [(9 * 3600 + 20 * 60, "9 h 20 min"), (1, "1 min"), (3601, "1 h 1 min"), (59 * 60, "59 min"), (3600, "1 h 0 min")])
def test_how_long_rounds_up_to_a_whole_minute(seconds, text):
    assert how_long(seconds) == text


@pytest.mark.parametrize("micro, text", [(250_000, "$0.25"), (4_999, "$0.00"), (5_000, "$0.01"), (9_925_000, "$9.93"), (1_000_000, "$1.00")])
def test_dollars_round_halves_up_to_whole_cents(micro, text):
    assert dollars(micro) == text


def test_twenty_calls_at_once_all_pass_a_check_made_before_any_is_charged():
    """The in-flight overrun. Without a set-aside every call sees an empty budget; with it only five fit."""
    each = cost(BURST_INPUT, BURST_OUTPUT)
    assert each == 48_000
    statuses, charged = fire(CheckOnlyBudget(LIMIT))
    assert (statuses.count(200), statuses.count(429), charged) == (CALLS, 0, CALLS * each)  # 20 calls, $0.96, 3.84 times the budget
    statuses, charged = fire(Budget(LIMIT))
    assert (statuses.count(200), statuses.count(429), charged) == (5, 15, 5 * each)  # 5 fit: 5 x $0.05 set aside = $0.25
    assert charged <= LIMIT


def test_two_server_processes_with_their_own_counts_each_allow_the_whole_budget():
    """Why the state must live in one shared store once there is more than one process (two Budget objects stand in for two processes)."""
    first, second = fire(Budget(LIMIT)), fire(Budget(LIMIT))
    assert first[0].count(200) + second[0].count(200) == 10
    assert first[1] + second[1] == 480_000 > 1.9 * LIMIT


def test_a_low_estimate_lets_more_calls_in_and_the_budget_is_overshot():
    """Guards the claim in the lesson: the set-aside is only as good as the estimate. Half the real input size admits 8 calls, not 5."""
    statuses, charged = fire(Budget(LIMIT), estimate=BURST_INPUT // 2)
    assert (statuses.count(200), charged) == (8, 384_000)
    assert charged > 1.5 * LIMIT


def test_the_check_and_the_set_aside_are_one_step_under_threads():
    """Guards a missing lock: a slow read widens the gap between the check and the set-aside, so threads that skip the lock all see an empty budget."""

    class SlowRead(Budget):
        def _fresh(self, user, now):
            fresh = super()._fresh(user, now)
            time.sleep(0.01)  # a real store takes time to answer
            return fresh

    budget, wins = SlowRead(1), []

    def grab():
        try:
            budget.reserve("a", 1, 0)
            wins.append(1)
        except OverBudget:
            pass

    threads = [threading.Thread(target=grab) for _ in range(8)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    assert len(wins) == 1


def test_refusal_message_names_what_is_left_what_is_needed_and_the_refill():
    refused = OverBudget(cost(50_000, MAX_REPLY), 30_000, 3600)  # a 50,000-token input with the reply cap
    assert refusal(refused, LIMIT, 0) == {"error": "budget_used_up", "resets_at": 3600,
                                         "message": "Your AI budget has $0.03 left and this call needs up to $0.11. It refills in 1 h 0 min."}
    over = Budget(LIMIT)
    over.settle(over.reserve("a", 1, 0), 300_000, 0)  # a reply that overran its set-aside: more than the whole budget is used
    with pytest.raises(OverBudget) as refused:
        over.reserve("a", 1, 0)
    assert "$0.00 left" in refusal(refused.value, LIMIT, 0)["message"]  # never a negative amount such as "$-1.95 left"


HERE = Path(__file__).parent
LESSON = (HERE.parent / "lesson.md").read_text()


def test_every_python_block_in_the_lesson_is_a_verbatim_excerpt_of_the_source():
    """The lesson quotes these files. If the code changes and the lesson does not, the reader sees code that was never tested."""
    sources = [(HERE / name).read_text().splitlines() for name in ("budget.py", "inflight.py", "sim.py")]
    blocks = re.findall(r"^```python\n(.*?)^```", LESSON, flags=re.S | re.M)
    assert blocks, "the lesson has no python blocks to check"
    for block in blocks:
        lines = block.rstrip("\n").splitlines()
        windows = (textwrap.dedent("\n".join(source[i:i + len(lines)])) for source in sources for i in range(len(source) - len(lines) + 1))
        assert textwrap.dedent("\n".join(lines)) in windows, f"not in the source:\n{block}"


def test_the_lesson_quotes_the_numbers_the_code_gives():
    """Guards the prose drifting from the code: the excerpt test above covers the code blocks and not the numbers around them."""
    none, budget = simulate_day("none"), simulate_day("budget")
    heavy_calls, all_calls = none.served[0], sum(none.served)
    check_only, set_aside, half = fire(CheckOnlyBudget(LIMIT)), fire(Budget(LIMIT)), fire(Budget(LIMIT), estimate=BURST_INPUT // 2)
    first, second = fire(Budget(LIMIT)), fire(Budget(LIMIT))  # two server processes, each with its own count
    quotes = [
        f"costs ${cost(60, 30) / 1e6:.5f}", f"costs {dollars(cost(50_000, 2_000))}", f"about {round(cost(50_000, 2_000) / cost(60, 30))} times as much",
        f"that user is {none.percent(0)}% of a {dollars(none.bill)} bill", f"cuts the bill to {dollars(budget.bill)} and the heavy user's share to {budget.percent(0)}%",
        f"refuses none of the nine other users' {none.ordinary_asked} calls", f"makes {heavy_calls} of the day's {all_calls} calls, {round(100 * heavy_calls / all_calls)}%",
        f"is refused {budget.refused[0]} of {heavy_calls} calls", f"The bill is {dollars(check_only[1])}", f"the bill is {dollars(set_aside[1])}",
        f"the bill is ${half[1] / 1e6:.3f}", f"{set_aside[0].count(200)} calls each hold", f"{half[0].count(200)} calls fit",
        f"{first[0].count(200) + second[0].count(200)} get in and {dollars(first[1] + second[1])} is charged", f"turns away {simulate_day('budget', budget=50_000).ordinary_refused} of their calls",
        refusal(OverBudget(cost(50_000, MAX_REPLY), 30_000, 3600), LIMIT, 0)["message"],
        f"the nine ordinary users lose {simulate_day('rpm', rpm=1).ordinary_refused} calls",
    ]
    missing = [q for q in quotes if q not in LESSON]
    assert not missing, missing

"""Tests for sim.py: the numbers the lesson and the widget show, and the properties behind them."""
import pytest

from budget import MAX_REPLY, cost, dollars
from sim import (BUDGET, CALL_GAP, HEAVY, ORDINARY_IN, ORDINARY_OUT, RPM, SEED, SESSION_CALLS, SESSIONS, USERS, make_day, random_numbers, simulate_day)


def test_the_same_seed_gives_the_same_day_and_another_seed_a_different_one():
    assert make_day(SEED) == make_day(SEED) and make_day(SEED) != make_day(SEED + 1)
    assert simulate_day("budget") == simulate_day("budget")
    day = make_day(SEED)
    assert len(day) == 394 and day == sorted(day, key=lambda c: (c[0], c[1]))
    assert sum(1 for c in day if c[1] == HEAVY) == 190


def test_the_random_numbers_do_not_change():
    """Pins mulberry32 so the Python and the widget's copy cannot both drift to another stream."""
    gen = random_numbers(1)
    assert [next(gen) for _ in range(3)] == [0.6270739405881613, 0.002735721180215478, 0.5274470399599522]
    assert all(0 <= next(gen) < 1 for _ in range(1000))


@pytest.mark.parametrize("rule, share, bill, heavy_refused", [("none", 94, "$9.93", 0), ("rpm", 94, "$9.93", 0), ("budget", 28, "$0.85", 185)])
def test_the_numbers_the_lesson_names_for_the_day_it_shows(rule, share, bill, heavy_refused):
    day = simulate_day(rule)
    assert (day.percent(HEAVY), dollars(day.bill), day.refused[HEAVY]) == (share, bill, heavy_refused)
    assert (day.ordinary_refused, day.ordinary_asked) == (0, 204)


def test_a_request_limit_does_not_touch_the_heavy_user():
    """The point of the lesson: the heavy user makes one call every four to six minutes, far under 5 a minute, and pays 94% of the bill."""
    none, rpm = simulate_day("none"), simulate_day("rpm")
    assert (rpm.served, rpm.refused, rpm.spent) == (none.served, none.refused, none.spent)


def test_a_very_tight_request_limit_hurts_ordinary_users_and_still_leaves_the_heavy_user_alone():
    """Guards the lesson's claim that lowering the request limit does not help: at 1 call a minute it still misses the heavy user and turns ordinary users away."""
    day = simulate_day("rpm", rpm=1)
    assert day.refused[HEAVY] == 0 and day.ordinary_refused == 69
    assert day.percent(HEAVY) >= 90


def test_over_many_days_the_budget_cuts_the_heavy_share_and_turns_no_ordinary_user_away():
    for seed in range(1, 201):
        none, budget = simulate_day("none", seed), simulate_day("budget", seed)
        assert none.percent(HEAVY) >= 90 and budget.percent(HEAVY) <= 35, seed
        assert budget.ordinary_refused == 0 and simulate_day("rpm", seed).ordinary_refused == 0, seed
        assert budget.bill < none.bill / 5, seed
        for cents in range(15, 55, 5):  # every widget step in the lesson's "size the budget" range: no ordinary call turned away on any day
            assert simulate_day("budget", seed, budget=cents * 10_000).ordinary_refused == 0, (seed, cents)


def test_no_ordinary_user_can_reach_the_default_budget():
    """The most an ordinary user can send in a day, plus the largest set-aside, still fits: so the budget never refuses one of them at its default."""
    most_calls = SESSIONS[1] * SESSION_CALLS[1]
    dearest_call = cost(ORDINARY_IN[1], ORDINARY_OUT[1])
    largest_set_aside = cost(ORDINARY_IN[1], MAX_REPLY)
    assert most_calls * dearest_call + largest_set_aside < BUDGET


def test_under_the_budget_rule_no_user_is_ever_charged_more_than_the_budget():
    """The reason to set aside the most a call can cost: the real cost is never above it, so the total stays under the limit."""
    for seed in range(1, 31):
        assert max(simulate_day("budget", seed).spent) <= BUDGET, seed


def test_every_call_is_either_served_or_refused_and_the_bill_is_the_sum_of_served_costs():
    calls = make_day(SEED)
    for rule in ("none", "rpm", "budget"):
        day = simulate_day(rule)
        assert sum(day.served) + sum(day.refused) == len(calls)
        for user in range(USERS):
            assert day.served[user] + day.refused[user] == sum(1 for c in calls if c[1] == user)
    assert simulate_day("none").spent == tuple(sum(cost(c[2], c[3]) for c in calls if c[1] == u) for u in range(USERS))


def test_a_smaller_budget_cuts_the_bill_and_a_bigger_one_lets_the_heavy_user_pay_more():
    bills = [simulate_day("budget", budget=cents * 10_000).bill for cents in (10, 25, 50)]
    assert bills == sorted(bills) and len(set(bills)) == 3
    assert simulate_day("budget", budget=5 * 10_000).ordinary_refused == 77  # below the reserve of a few calls, ordinary users are hit too


def test_call_gaps_stay_inside_the_day():
    assert max(c[0] for c in make_day(SEED)) < 86_400 and CALL_GAP[1] * SESSION_CALLS[1] <= 8 * 89 and RPM == 5

"""Guards for ../widget.html: its data block, what it may show, and that its rules give the same answers as sim.py."""
import json
import re
import subprocess

from budget import dollars
from embed_widget import WIDGET, config, embed
from sim import BUDGET, RPM, RULES, SEED, make_day, random_numbers, simulate_day

HTML = WIDGET.read_text(encoding="utf-8")
NAMES = dict(RULES)

# Reads the rules out of the widget between its marker comments, runs them, and compares with the Python answers.
PARITY = r"""
import fs from 'node:fs';
const [fixture, widget] = process.argv.slice(2);
const fx = JSON.parse(fs.readFileSync(fixture, 'utf8'));
const html = fs.readFileSync(widget, 'utf8');
const cfg = JSON.parse(html.match(/<script type="application\/json" id="qb-config">([\s\S]*?)<\/script>/)[1]);
const js = html.slice(html.indexOf("/* ---------- the toy's rules"), html.indexOf('/* ---------- the page'));
const api = new Function('cfg', "'use strict';" + js + '; return {randomNumbers, makeDay, simulateDay, summary, dollars};')(cfg);
const bad = [];
fx.numbers.forEach((n) => {
  const next = api.randomNumbers(n.seed);
  n.values.forEach((v, i) => { if (next() !== v) bad.push(['number', n.seed, i]); });
});
fx.days.forEach((d) => { if (JSON.stringify(api.makeDay(d.seed)) !== JSON.stringify(d.calls)) bad.push(['day', d.seed]); });
fx.runs.forEach((r) => {
  const got = api.simulateDay(r.rule, r.seed, r.rpm, r.budget);
  if (JSON.stringify([got.served, got.refused, got.spent]) !== JSON.stringify([r.served, r.refused, r.spent])) bad.push(['run', r.rule, r.seed, r.rpm, r.budget]);
  if (JSON.stringify(api.summary(got)) !== JSON.stringify(r.summary)) bad.push(['summary', r.rule, r.seed, r.rpm, r.budget]);
});
fx.money.forEach((m) => { if (api.dollars(m.micro) !== m.text) bad.push(['dollars', m.micro]); });
console.log(JSON.stringify({ numbers: fx.numbers.length, days: fx.days.length, runs: fx.runs.length, money: fx.money.length, bad: bad.slice(0, 5) }));
process.exit(bad.length ? 1 : 0);
"""


# Runs the page script (everything after the rules) against a small stand-in for the DOM, and reads back what a reader would see.
PAGE_TEST = r"""
import fs from 'node:fs';
const [widget, fixture] = process.argv.slice(2);
const html = fs.readFileSync(widget, 'utf8');
const states = JSON.parse(fs.readFileSync(fixture, 'utf8')).states;
const cfgText = html.match(/<script type="application\/json" id="qb-config">([\s\S]*?)<\/script>/)[1];
const code = html.match(/<script>([\s\S]*?)<\/script>/)[1];

function el(tag) {
  const e = { tag, children: [], listeners: {}, className: '', checked: false, value: '', style: {}, attributes: {}, _text: '' };
  e.addEventListener = (type, f) => (e.listeners[type] = e.listeners[type] || []).push(f);
  e.appendChild = (c) => (e.children.push(c), c);
  e.setAttribute = (k, v) => { e.attributes[k] = v; };
  Object.defineProperty(e, 'textContent', { get: () => e._text, set: (v) => { e._text = String(v); e.children = []; } });
  return e;
}
function open() {
  const page = {};
  ['qb-config', 'qb-live', 'qb-picks', 'qb-what', 'qb-big', 'qb-sub', 'qb-bars', 'qb-again', 'qb-note', 'qb-rows',
   'qb-rpm', 'qb-budget', 'qb-rpm-v', 'qb-budget-v'].forEach((id) => { page[id] = el('x'); });
  page['qb-config']._text = cfgText;
  const doc = { getElementById: (id) => page[id], createElement: el, createTextNode: (text) => ({ text }), documentElement: { setAttribute() {} } };
  new Function('document', 'window', 'parent', code)(doc, { addEventListener() {} }, {});
  return page;
}
const fire = (e, type) => (e.listeners[type] || []).forEach((f) => f());
const radios = (page) => page['qb-picks'].children.map((label) => label.children[0]);
const labelText = (page, i) => page['qb-picks'].children[i].children[1].text;
const cells = (tr) => tr.children.map((c) => c._text);

const bad = [];
const want = (what, got, expected) => { if (JSON.stringify(got) !== JSON.stringify(expected)) bad.push([what, got, expected]); };
states.forEach((st) => {
  const page = open();
  [['rpm', st.rpm], ['budget', st.cents]].forEach(([key, v]) => { if (v !== undefined) { page['qb-' + key].value = String(v); fire(page['qb-' + key], 'input'); } });
  const picks = radios(page);
  picks.forEach((r, i) => { r.checked = i === st.pick; });
  if (st.pick !== 1) fire(picks[st.pick], 'change');
  for (let i = 0; i < st.again; i++) fire(page['qb-again'], 'click');
  want(st.name + ': rule description', page['qb-what']._text, st.what);
  want(st.name + ': headline', page['qb-big']._text, st.big);
  want(st.name + ': second line', page['qb-sub']._text, st.sub);
  want(st.name + ': note beside the button', page['qb-note']._text, st.note);
  want(st.name + ': table', page['qb-rows'].children.map(cells), st.rows);
  want(st.name + ': only the picked rule is marked', page['qb-rows'].children.map((tr) => tr.className), st.marks);
  want(st.name + ': slider read-outs', [page['qb-rpm-v']._text, page['qb-budget-v']._text], st.sliders);
  const rows = page['qb-bars'].children;
  want(st.name + ': one row for each user', rows.length, 10);
  want(st.name + ': row text', rows.map((r) => [r.children[0]._text, r.children[2]._text, r.children[3]._text]), st.bars);
  want(st.name + ': the heavy bar is orange, the others green', rows.map((r) => r.children[1].children[0].className.split(' ')[1]), st.colors);
  rows.forEach((r, u) => {
    const w = parseFloat(r.children[1].children[0].style.width);
    if (!(Math.abs(w - st.widths[u]) <= 0.05)) bad.push([st.name + ': bar width for user ' + u, w, st.widths[u]]);
  });
  want(st.name + ': bar label for screen readers', page['qb-bars'].attributes['aria-label'], st.label);
  const touched = st.pick !== 1 || st.again > 0 || st.rpm !== undefined || st.cents !== undefined;
  want(st.name + ': screen reader line', touched ? page['qb-live']._text === st.rule + ': ' + st.big + ' ' + st.sub : page['qb-live']._text === '', true);
});
const headers = [...html.match(/<thead>[\s\S]*?<\/thead>/)[0].matchAll(/<th scope="col">([^<]*)<\/th>/g)].map((m) => m[1]);
want('the results table headers, in the order of the cells the script writes', headers, ['Rule', 'Heavy share', 'Bill', 'Ordinary refused']);
const legend = [...html.matchAll(/<i class="qb-sw (qb-\w+)"><\/i>([^<]*)<\/span>/g)].map((m) => [m[1], m[2]]);
want('the key names each bar color', legend, [['qb-hv', 'heavy user'], ['qb-ok', 'ordinary user']]);
const liveTag = html.match(/<p[^>]*id="qb-live"[^>]*>/)[0];
want('the live region is a polite status read whole', ['role="status"', 'aria-live="polite"', 'aria-atomic="true"'].map((a) => liveTag.includes(a)), [true, true, true]);
const start = radios(open());
want('the three radios share one group name', new Set(start.map((r) => r.name)).size === 1 && !!start[0].name, true);
want('the widget opens on requests per minute', start.map((r) => r.checked), [false, true, false]);
want('the radio labels are the rule names', [0, 1, 2].map((i) => labelText(open(), i)), JSON.parse(cfgText).rules.map((r) => r.name));
console.log(JSON.stringify({ states: states.length, bad }));
process.exit(bad.length ? 1 : 0);
"""


def expected_page(pick, again=0, rpm=RPM, cents=BUDGET // 10_000):
    """What the page should show after these choices, worked out from sim.py."""
    seed, budget = SEED + again, cents * 10_000
    days = {rule: simulate_day(rule, seed, rpm, budget) for rule, _ in RULES}
    rule = RULES[pick][0]
    day = days[rule]
    what = {"none": "The server makes every call.",
            "rpm": f"Each user may make {rpm} call{'' if rpm == 1 else 's'} a minute (a bucket of tickets for each user, as in lesson 9).",
            "budget": f"Each user may spend {dollars(budget)} a day. Before a call the server sets aside the most it can cost, refuses if that does not fit, then charges what it really cost."}[rule]
    big = f"Heavy user: {day.percent(0)}% of a {dollars(day.bill)} bill."
    sub = (f"Ordinary users refused: {day.ordinary_refused} of {day.ordinary_asked} calls. "
           f"Heavy user refused: {day.refused[0]} of {day.served[0] + day.refused[0]} calls.")
    names = ["Heavy"] + [f"User {u}" for u in range(1, 10)]
    return {
        "rule": NAMES[rule], "pick": pick, "again": again, "what": what, "big": big, "sub": sub,
        "note": f"Seed {seed}. Another day draws new calls.",
        "rows": [[name, f"{days[r].percent(0)}%", dollars(days[r].bill), f"{days[r].ordinary_refused} of {days[r].ordinary_asked}"] for r, name in RULES],
        "marks": ["qb-sel" if r == rule else "" for r, _ in RULES],
        "sliders": [str(rpm), dollars(budget)],
        "bars": [[names[u], f"{day.percent(u)}%", str(day.refused[u])] for u in range(10)],
        "colors": ["qb-hv"] + ["qb-ok"] * 9,
        "widths": [100 * s / day.bill for s in day.spent],
        "label": f"{NAMES[rule]}: " + "; ".join(f"{names[u]} {day.percent(u)}% of the bill, {day.refused[u]} refused" for u in range(10)),
    }


PAGE_CHOICES = [
    ("opens on requests per minute", dict(pick=1)),
    ("no limit", dict(pick=0)),
    ("daily cost budget", dict(pick=2)),
    ("budget, another day", dict(pick=2, again=1)),
    ("rpm, two other days", dict(pick=1, again=2)),
    ("rpm at 1 a minute", dict(pick=1, rpm=1)),
    ("budget at 5 cents", dict(pick=2, cents=5)),
    ("budget at 50 cents", dict(pick=2, cents=50)),
    ("a slider alone, no other change", dict(pick=1, rpm=10)),
]


def test_widget_data_block_matches_sim():
    """Guards a sim.py edit that was not copied into the widget (run embed_widget.py)."""
    assert embed(HTML) == HTML


def test_widget_sliders_hold_every_value_the_data_block_sets():
    """Guards a setting changed in sim.py past the end of its slider: the browser would clamp it and the page would show another day."""
    cfg = config()
    for key, value in (("rpm", cfg["rpm"]), ("budget", cfg["budget"] // 10_000)):
        low, high = (int(v) for v in re.search(rf'id="qb-{key}" min="(\d+)" max="(\d+)"', HTML).groups())
        assert low <= value <= high, key
    assert cfg["budget"] % 10_000 == 0 and re.search(r'id="qb-budget" min="5" max="50" step="5"', HTML)


def test_every_element_the_script_looks_up_exists_inside_the_root():
    """Guards a widget that never starts: a lookup of an id that is not in the page throws in the browser."""
    root = HTML[HTML.index('<div class="qb-root">'):HTML.index("<script")]
    script = HTML[HTML.index("<script>"):]
    lookups = set(re.findall(r"\$\('([\w-]+)'\)", script))
    assert "['qb-rpm', 'qb-budget'].forEach" in script, "the slider id list has moved: update this test"
    lookups |= {"qb-rpm", "qb-budget"}
    created = {"qb-pick-" + r["id"] for r in config()["rules"]}  # made by the script, not in the page
    missing = sorted(i for i in lookups if f'id="{i}"' not in root and i not in created)
    assert lookups and not missing, missing


def test_widget_is_small_and_asks_for_nothing_online():
    assert len(HTML.splitlines()) <= 300 and len(HTML.encode("utf-8")) < 40 * 1024
    page = re.sub(r'<script type="application/json" id="qb-config">.*?</script>', "", HTML, flags=re.S)
    for banned in ("http://", "https://", "fetch(", "XMLHttpRequest", "WebSocket", "sendBeacon", "import(", "Math.random"):
        assert banned not in page, banned
    assert " src=" not in page and "@import" not in page


def test_widget_has_no_form_element():
    """Guards a dead button: the lesson page loads widget.html in a sandboxed frame, where a form cannot submit."""
    assert "<form" not in HTML and 'type="submit"' not in HTML


def test_widget_keeps_the_site_theme_listener_and_palette():
    assert "d.type !== 'spicyshark:theme'" in HTML and "e.source !== parent" in HTML and "setAttribute('data-theme', d.theme)" in HTML
    for pair in ("light-dark(#ffffff, #0a0f17)", "light-dark(#ebf0f6, #131b28)", "light-dark(#0a1626, #f4f7fb)",
                 "light-dark(#44566b, #a3b1c4)", "light-dark(#d2dae4, #1c2635)", "light-dark(#66768a, #7a8a9f)",
                 "--green: light-dark(#00855F, #009E73)", "--orange: light-dark(#b36200, #E69F00)",
                 ".qb-root .qb-ok { background: var(--green)", ".qb-root .qb-hv { background: var(--orange)"):
        assert pair in HTML, pair
    assert not re.search(r"0072B2|56B4E9|D55E00", HTML, re.I)


def test_widget_keeps_extra_settings_in_a_closed_more_settings_block():
    """The lesson's brief: the point shows with three radios and one button; the sliders wait behind "More settings"."""
    assert re.search(r"<details>\s*<summary>More settings</summary>", HTML) and "<details open" not in HTML
    visible = HTML[:HTML.index("<details>")]
    assert len(re.findall(r"<button\b", visible)) == 1 and 'type="range"' not in visible


def test_widget_rules_match_the_python_sim(tmp_path):
    """Guards the widget's JavaScript drifting from sim.py: the same numbers, days and results for every rule, seed and setting."""
    numbers = []
    for seed in (0, 1, 2, 3, 7, 12345, 2 ** 31, 2 ** 32 - 1):
        gen = random_numbers(seed)
        numbers.append({"seed": seed, "values": [next(gen) for _ in range(300)]})
    days = [{"seed": seed, "calls": [list(c) for c in make_day(seed)]} for seed in range(0, 21)]
    runs = []
    for rpm, budget in ((RPM, BUDGET), (1, 50_000), (2, 100_000), (10, 500_000), (3, 150_000)):
        for seed in range(1, 21):
            for rule, _ in RULES:
                day = simulate_day(rule, seed, rpm, budget)
                runs.append({"rule": rule, "seed": seed, "rpm": rpm, "budget": budget, "served": list(day.served), "refused": list(day.refused),
                             "spent": list(day.spent),
                             "summary": {"bill": day.bill, "pct": [day.percent(u) for u in range(10)], "ordinaryRefused": day.ordinary_refused,
                                         "ordinaryAsked": day.ordinary_asked, "heavyAsked": day.served[0] + day.refused[0]}})
    money = [{"micro": m, "text": dollars(m)} for m in (0, 4_999, 5_000, 14_999, 15_000, 250_000, 995_000, 9_925_000, 9_934_567, 12_345_678)]
    (tmp_path / "fixture.json").write_text(json.dumps({"numbers": numbers, "days": days, "runs": runs, "money": money}), encoding="utf-8")
    (tmp_path / "parity.mjs").write_text(PARITY, encoding="utf-8")
    result = subprocess.run(["node", str(tmp_path / "parity.mjs"), str(tmp_path / "fixture.json"), str(WIDGET)], capture_output=True, text=True, timeout=120)
    assert result.returncode == 0, result.stdout + result.stderr
    assert json.loads(result.stdout) == {"numbers": 8, "days": 21, "runs": 5 * 20 * 3, "money": 10, "bad": []}


def test_widget_page_shows_what_the_lesson_names(tmp_path):
    """Guards the part a reader sees: headline, second line, bars, comparison table, the note beside the button and the sliders' wiring.

    The lesson names these runs: requests per minute leaves the heavy user at 94% of a $9.93 bill, the budget cuts it to 28% of $0.85."""
    states = [{"name": name, **choice, **expected_page(**choice)} for name, choice in PAGE_CHOICES]
    (tmp_path / "states.json").write_text(json.dumps({"states": states}), encoding="utf-8")
    (tmp_path / "page.mjs").write_text(PAGE_TEST, encoding="utf-8")
    result = subprocess.run(["node", str(tmp_path / "page.mjs"), str(WIDGET), str(tmp_path / "states.json")], capture_output=True, text=True, timeout=60)
    assert result.returncode == 0, result.stdout + result.stderr
    assert json.loads(result.stdout) == {"states": len(PAGE_CHOICES), "bad": []}
    first = states[0]
    assert first["big"] == "Heavy user: 94% of a $9.93 bill."
    assert first["rows"] == [["No limit", "94%", "$9.93", "0 of 204"], ["Requests per minute", "94%", "$9.93", "0 of 204"], ["Daily cost budget", "28%", "$0.85", "0 of 204"]]
    assert states[2]["big"] == "Heavy user: 28% of a $0.85 bill." and states[2]["sub"].startswith("Ordinary users refused: 0 of 204 calls.")