Can one user's chat leak into another's?

"""Keep the data inside ../widget.html equal to the data in leak_sim.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

from leak_sim import EXPERIMENTS, GUESSES, QUESTIONS, RIGHT, SEQUENCE, TRIES, USERS, percent, simulate, timing

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


def config():
    shared, salted = timing(False), timing(True)
    average = {"shared": [round(w) for w in shared["average"]], "salted": [round(w) for w in salted["average"]]}
    return {"users": [{"name": u.name, "order": u.order, "street": u.street} for u in USERS],
            "questions": [{"q": q, "a": a} for q, a in QUESTIONS],
            "sequence": [list(r) for r in SEQUENCE],
            "opening": next(s.n for s in simulate(False) if s.leak),
            "guesses": GUESSES, "right": RIGHT, "tries": TRIES, "runs": EXPERIMENTS, "chance": percent(1, len(GUESSES)),
            "axis": -(-max(max(v) for v in average.values()) // 10) * 10,
            "timing": {"shared": {"average": average["shared"], "one": shared["one"], "many": shared["many"]},
                       "salted": {"average": average["salted"], "one": salted["one"], "many": salted["many"]}}}


def embed(html):
    """html with the config block rewritten from leak_sim.py."""
    block = json.dumps(config())
    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 leak_sim.py")

"""Can one user's chat leak into another's? A small simulation of two shared caches in front of a shop's help bot.

Python 3 standard library only (plus lesson 23's code, imported, not copied). Run `python3 show_numbers.py` to print every
number the lesson quotes.

The toy, in five rules:
  1. A made-up shop has three made-up users. Each has two private facts, an order code and a street.
  2. The help bot answers four kinds of question. Three answers carry a private fact (an order code or a street) and
     one, opening hours, is the same for everyone.
  3. A response cache sits in front of the bot and keeps whole answers. Its key is either the question alone ("shared")
     or the user plus the question ("per user"). A request whose key is already stored is a hit and gets the stored answer.
  4. A hit is a leak when the stored answer is not the answer this user should get. In this toy every such answer carries
     another user's order code or street.
  5. A prompt cache (lesson 8's idea: a stored start of a prompt is reused whole or not at all) makes the wait before the
     first word shorter when it is hit. The waits come from lesson 23's best-case reading time plus a random queueing delay
     with a fixed seed. Real waits are noisier and real caches store per block; nothing here touches a real service.
"""
from __future__ import annotations

import random
import sys
from dataclasses import dataclass
from pathlib import Path

LESSONS = Path(__file__).resolve().parents[2]
sys.path.append(str(LESSONS / "why-ai-waits-before-the-first-word" / "code"))
import prefill_sim                                        # lesson 23: the best-case time of reading a prompt


# --- the shop, its users and its questions -------------------------------------------------------------------

@dataclass(frozen=True)
class User:
    name: str                   # "A", "B", "C": the widget and the lesson write "User A"
    order: str                  # a made-up order code
    street: str                 # a made-up street, no number, no town


USERS = [User("A", "A-1042", "Mill Lane"), User("B", "B-2210", "Quarry Road"), User("C", "C-3377", "Orchard Close")]
USER_BY_NAME = {u.name: u for u in USERS}

# (question as the user types it, the answer). {order} and {street} are filled from the user.
QUESTIONS = [
    ("Where is my order?", "Order {order} is out for delivery."),
    ("What is my delivery address?", "We deliver to {street}."),
    ("What are your opening hours?", "We open 9 to 5, Monday to Saturday."),
    ("Can you gift-wrap order {order}?", "Yes, order {order} can be gift-wrapped."),
]
PUBLIC = 2                      # the question whose answer is the same for everyone

# (user, question number): fixed, so every run gives the same table.
SEQUENCE = [("A", 0), ("B", 0), ("A", 2), ("C", 2), ("B", 1), ("A", 1), ("C", 0),
            ("A", 0), ("B", 3), ("B", 3), ("C", 1), ("B", 0), ("A", 2), ("C", 0)]


def fill(template: str, user: User) -> str:
    return template.replace("{order}", user.order).replace("{street}", user.street)


def question(user: User, q: int) -> str:
    """The words the user types; question 3 holds the user's own order code, so it differs from user to user."""
    return fill(QUESTIONS[q][0], user)


def answer(user: User, q: int) -> str:
    """The answer this user should get."""
    return fill(QUESTIONS[q][1], user)


# --- the response cache --------------------------------------------------------------------------------------

@dataclass(frozen=True)
class Step:
    n: int                      # request number, from 1
    user: str
    question: str
    key: str                    # what the cache looks up
    hit: bool
    written_by: int | None      # on a hit, the number of the request whose answer is stored
    served: str                 # the answer the user got
    right: str                  # the answer the user should have got
    leak: bool                  # served is not right


def cache_key(user: User, text: str, per_user: bool) -> str:
    return f"{user.name} | {text}" if per_user else text


def simulate(per_user: bool, sequence=SEQUENCE) -> list[Step]:
    """Run the requests through a response cache that never expires. A miss writes the right answer for the asker."""
    store: dict[str, tuple[str, int]] = {}
    steps = []
    for n, (name, q) in enumerate(sequence, 1):
        user, text, right = USER_BY_NAME[name], question(USER_BY_NAME[name], q), answer(USER_BY_NAME[name], q)
        key = cache_key(user, text, per_user)
        if key in store:
            served, by = store[key]
            steps.append(Step(n, name, text, key, True, by, served, right, served != right))
        else:
            store[key] = (right, n)
            steps.append(Step(n, name, text, key, False, None, right, right, False))
    return steps


def percent(part: int, whole: int) -> int:
    """Whole percent, rounded half up, in integers so that the widget's JavaScript prints the same digits."""
    return (200 * part + whole) // (2 * whole)


def totals(steps: list[Step]) -> dict:
    hits, leaks, n = sum(s.hit for s in steps), sum(s.leak for s in steps), len(steps)
    return {"requests": n, "hits": hits, "leaks": leaks, "rate": percent(hits, n)}


# --- the prompt cache and the wait before the first word ------------------------------------------------------

INSTRUCTION = 900               # tokens that start every prompt of the shop's bot, the same for everyone
NOTE = 100                      # tokens of the user's own note, right after them
TAIL = 20                       # tokens of the new question, last
TARGET = "Mill Lane"            # the street in User A's note
GUESSES = ["Elm Road", "Mill Lane", "Fern Close"]   # what User B tries; the middle one is right
RIGHT = GUESSES.index(TARGET)
QUEUE_MS = 60.0                 # mean of the random queueing delay added to every wait, in milliseconds (a toy figure)
TRIES = 25                      # waits measured per guess in one experiment
EXPERIMENTS = 2000
SEED = 2026


def prompt(street: str) -> list:
    """Token labels, equal where the text is equal: the shared instruction, the note about a street, and a fresh question."""
    return [("sys", i) for i in range(INSTRUCTION)] + [(street, i) for i in range(NOTE)] + [("question", i) for i in range(TAIL)]


class PromptCache:
    """Stored starts of prompts. A start is kept under a salt; a request reads a stored start only under the same salt,
    and only if its prompt begins with the whole of it (lesson 8's rule)."""

    def __init__(self):
        self.starts: list[tuple[str, list]] = []

    def store(self, salt: str, start: list) -> None:
        self.starts.append((salt, start))

    def reused(self, salt: str, prompt_tokens: list) -> int:
        return max((len(s) for t, s in self.starts if t == salt and prompt_tokens[:len(s)] == s), default=0)


def wait_ms(total: int, reused: int) -> float:
    """Best-case time to read the tokens that are not reused, from lesson 23's model at the chip's peak, in milliseconds."""
    return prefill_sim.prefill(total - reused).seconds * 1000


def ideal_waits(salted: bool) -> list[float]:
    """The wait of each of User B's three guesses, with no random delay, right after User A sent its note.
    Not salted: both users share one key space. Salted: each user's prompts are stored under that user's name."""
    cache = PromptCache()
    cache.store("A" if salted else "", prompt(TARGET)[:INSTRUCTION + NOTE])
    waits = []
    for street in GUESSES:
        tokens = prompt(street)
        waits.append(wait_ms(len(tokens), cache.reused("B" if salted else "", tokens)))
    return waits


def timing(salted: bool, seed: int = SEED, experiments: int = EXPERIMENTS, tries: int = TRIES) -> dict:
    """User B picks the guess with the shortest wait, either from one try per guess or from the average of `tries`.
    Each experiment draws fresh delays; the same seed gives the same draws in both modes, so only the key differs.
    The cache is as User A left it before every try (a real request would fill the cache itself)."""
    rng, ideal = random.Random(seed), ideal_waits(salted)
    one = many = 0
    sums = [0.0] * len(ideal)
    for _ in range(experiments):
        waits = [[w + rng.expovariate(1 / QUEUE_MS) for _ in range(tries)] for w in ideal]
        first = [w[0] for w in waits]
        means = [sum(w) / tries for w in waits]
        one += first.index(min(first)) == RIGHT
        many += means.index(min(means)) == RIGHT
        sums = [s + m for s, m in zip(sums, means)]
    return {"ideal": ideal, "average": [s / experiments for s in sums], "one": percent(one, experiments), "many": percent(many, experiments)}

"""Print every number the lesson quotes. Run it from this folder: python3 show_numbers.py"""
from leak_sim import (EXPERIMENTS, GUESSES, INSTRUCTION, NOTE, QUEUE_MS, QUESTIONS, RIGHT, SEQUENCE, TAIL, TRIES, ideal_waits, simulate,
                      timing, totals, wait_ms)

shared, per_user = simulate(False), simulate(True)

print("== The requests (the lesson's table)")
print("  # | user | question | shared: result | per user: result")
for a, b in zip(shared, per_user):
    def result(s):
        return "LEAK" if s.leak else f"hit (request {s.written_by})" if s.hit else "miss"
    print(f"  {a.n} | {a.user} | {a.question} | {result(a)} | {result(b)}")

print("\n== Counts")
for name, steps in (("shared key (question only)", shared), ("per-user key (user + question)", per_user)):
    t = totals(steps)
    print(f"  {name}: {t['hits']} hits of {t['requests']}, {t['leaks']} leaks, hit rate {t['rate']}%")
harmless = [s.n for s, p in zip(shared, per_user) if s.hit and not s.leak and not p.hit]
print(f"  hits the shared key got right that the per-user key gives up: requests {harmless}")
good_shared = sum(s.hit and not s.leak for s in shared)
good_user = sum(s.hit and not s.leak for s in per_user)
print(f"  hits that served the right answer: {good_shared} shared, {good_user} per user")
print(f"  leaked answers by who got them: " + ", ".join(f"{s.n} (User {s.user} got an answer written for request {s.written_by})" for s in shared if s.leak))
public = [i for i, (_, a) in enumerate(QUESTIONS) if "{" not in a]
print(f"  questions: {len(QUESTIONS)}; asked in the sequence: {len({q for _, q in SEQUENCE})}; answers with no private fact: question numbers {public}")

print("\n== The prompt cache and the wait before the first word")
tokens = INSTRUCTION + NOTE + TAIL
print(f"  prompt: {INSTRUCTION} shared instruction + {NOTE} note + {TAIL} question = {tokens} tokens; the stored start is {INSTRUCTION + NOTE}")
print(f"  best-case wait if the start is reused: {wait_ms(tokens, INSTRUCTION + NOTE):.1f} ms; read again from the top: {wait_ms(tokens, 0):.1f} ms")
blocks = wait_ms(tokens, INSTRUCTION)
print(f"  with storage in blocks a wrong guess still reuses the {INSTRUCTION}-token instruction and reads {NOTE + TAIL} tokens: {blocks:.1f} ms, a gap of {blocks - wait_ms(tokens, INSTRUCTION + NOTE):.1f} ms to the right guess")
print(f"  random queueing delay added to each wait: mean {QUEUE_MS:g} ms; {TRIES} waits per guess; {EXPERIMENTS} experiments; guesses {GUESSES}, right one is number {RIGHT + 1}")
for salted in (False, True):
    t = timing(salted)
    label = "user in the key (salted)" if salted else "shared key"
    print(f"  {label}:")
    print(f"    waits with no delay (ms): " + ", ".join(f"{g} {w:.1f}" for g, w in zip(GUESSES, ideal_waits(salted))))
    print(f"    average of {TRIES} waits, averaged over {EXPERIMENTS} experiments (ms): " + ", ".join(f"{g} {w:.0f}" for g, w in zip(GUESSES, t["average"])))
    print(f"    the fastest guess is the right one: {t['one']}% with one try per guess, {t['many']}% with the average of {TRIES} (chance is 33%)")

"""Guards for leak_sim.py: the counts the lesson quotes, and the two promises it makes (the user in the key stops every leak and
every timing signal)."""
import random
import re
from pathlib import Path

from leak_sim import (EXPERIMENTS, GUESSES, INSTRUCTION, NOTE, PUBLIC, QUESTIONS, QUEUE_MS, RIGHT, SEED, SEQUENCE, TAIL, TRIES, USERS,
                      PromptCache, answer, cache_key, ideal_waits, percent, prompt, question, simulate, timing, totals, wait_ms)

LESSON = Path(__file__).resolve().parent.parent


def test_the_lessons_counts():
    """Guards a table the lesson quotes drifting: 6 leaks in 10 hits with the shared key, none in 5 hits with the user in it. Also guards
    prose and figures that drift from the code: the numbers the lesson quotes from `show_numbers.py`, in the sentence, table row or figure label that
    quotes it, is rebuilt here from the code's results and must appear in lesson.md or in the figures' text."""
    shared, per_user = simulate(False), simulate(True)
    assert totals(shared) == {"requests": 14, "hits": 10, "leaks": 6, "rate": 71}
    assert totals(per_user) == {"requests": 14, "hits": 5, "leaks": 0, "rate": 36}

    lesson = (LESSON / "lesson.md").read_text(encoding="utf-8")
    figures = {f.name: " ".join(re.findall(r"<text[^>]*>([^<]*)</text>", f.read_text(encoding="utf-8"))) for f in (LESSON / "figs").glob("*.svg")}
    ts, tu, n = totals(shared), totals(per_user), len(SEQUENCE)
    right_shared, right_user = sum(s.hit and not s.leak for s in shared), sum(s.hit and not s.leak for s in per_user)
    given_up = [s.n for s, p in zip(shared, per_user) if s.hit and not s.leak and not p.hit]
    won = [s.n for s, p in zip(shared, per_user) if s.leak and p.hit]
    assert len(USERS) == 3 and len(QUESTIONS) == 4 and given_up == [4] and won == [12, 14]
    tokens = INSTRUCTION + NOTE + TAIL
    hit_ms, miss_ms, block_ms = wait_ms(tokens, INSTRUCTION + NOTE), wait_ms(tokens, 0), wait_ms(tokens, INSTRUCTION)
    sh, us = timing(False), timing(True)
    chance, gap, spread = percent(1, len(GUESSES)), miss_ms - hit_ms, QUEUE_MS / TRIES ** 0.5
    label = lambda s: "LEAK " + str(s.written_by) if s.leak else f"hit {s.written_by}" if s.hit else "miss"
    short = ["order", "street", "hours", "gift"]
    a, b = USERS[0], USERS[1]

    quoted = [
        # the opening paragraph, the explanation of each key, and the timing section
        f"In a toy run of {n} requests from three made-up customers, the shelf answered {ts['hits']} of them without asking the bot, "
        f"and {ts['leaks']} of those answers went to the wrong customer",
        f"Over the {n} requests, this key gave {ts['hits']} hits, {ts['leaks']} of them leaks, a hit rate (the share of requests the shelf answered) of {ts['rate']}%",
        f"The same {n} requests give {tu['hits']} hits, no leaks and a hit rate of {tu['rate']}%",
        f"In the toy a hit waits {hit_ms:.1f} milliseconds (ms) at best and a miss {miss_ms:.1f} ms",
        f"the quickest of three guesses is the real one {sh['one']}% of the time, and {sh['many']}% after averaging {TRIES} waits each. Chance is {chance}%",
        f"every guess waits the same: {us['many']}% either way",
        # Go deeper: the counts
        f"The {n} requests come in a fixed order",
        f"With the shared key the {n} requests give {ts['hits']} hits, {ts['leaks']} leaks and a hit rate of {ts['rate']}%. "
        f"With the user in the key they give {tu['hits']} hits, {tu['leaks']} leaks and {tu['rate']}%",
        f"because {ts['leaks']} of its {ts['hits']} hits were wrong",
        f"in this sequence the user key does not lose: {right_user} against {right_shared}",
        f"It gives up request {given_up[0]}, the harmless hours hit, and wins requests {won[0]} and {won[1]}, which were leaks under the shared key",
        # Go deeper: the timing toy
        f"Each prompt has {INSTRUCTION} tokens that start every request of the shop's bot, {NOTE} tokens of the user's note and {TAIL} tokens of the new question: {tokens:,} in all",
        f"the server stored the first {INSTRUCTION + NOTE:,} tokens",
        f"matches only the {INSTRUCTION}-token instruction",
        f"so all {tokens:,} tokens are read",
        f"{hit_ms:.1f} ms for {TAIL} tokens and {miss_ms:.1f} ms for {tokens:,}",
        f"B's guesses all read {tokens:,}",
        f"with a mean of {QUEUE_MS:g} ms",
        f"One experiment draws {TRIES} waits for each guess",
        f"over {EXPERIMENTS:,} experiments, with the fixed seed {SEED}",
        *(f"| {g} | {round(x)} | {round(y)} |" for g, x, y in zip(GUESSES, sh["average"], us["average"])),
        f"| One try | {sh['one']}% | {us['one']}% |", f"| {TRIES} tries | {sh['many']}% | {us['many']}% |",
        f"The gap between a hit and a miss is {gap:.1f} ms, and the delay has a mean and a spread of about {QUEUE_MS:g} ms",
        f"so {TRIES} tries cut the spread to about {spread:.0f} ms",
        f"{us['many']}% is a blind guess",
        # What the toy leaves out: the block-storage sentence, with its own gap
        f"a wrong guess would still reuse the shared {INSTRUCTION}-token instruction and read only {NOTE + TAIL} tokens",
        f"The toy's model puts those {NOTE + TAIL} tokens at {block_ms:.1f} ms, like the right guess's {TAIL}",
        f"would then show a gap of {block_ms - hit_ms:.1f} ms, so the toy's {gap:.1f} ms gap comes from its whole-or-nothing rule",
        # the 14 request rows
        *(f"| {x.n}. {x.user}: {short[q]} | {label(x)} | {label(y)} |" for (_, q), x, y in zip(SEQUENCE, shared, per_user)),
        # Proof it works, and the Check-yourself answers
        f"{n}, {ts['hits']}, {ts['leaks']} and {ts['rate']}% with the shared key, and {n}, {tu['hits']}, {tu['leaks']} and {tu['rate']}% with the user in it",
        f"{sh['one']}% and {sh['many']}% shared, {us['one']}% and {us['many']}% salted",
        f"the hit rate falls from {ts['rate']}% to {tu['rate']}% when the user goes in the key",
        f"of the shared key's {ts['hits']} hits were leaks, so only {right_shared} were right. The user key has {tu['hits']} hits and all are right: "
        f"it gives up one harmless hit (request {given_up[0]}) and wins two (requests {won[0]} and {won[1]})",
        f"one try per guess finds the real street {sh['one']}% of the time and the average of {TRIES} finds it {sh['many']}%. With the user in the key both give {us['one']}%",
        f"The real guess waits {gap:.1f} ms less, but the random delay of about {QUEUE_MS:g} ms hides that in one try",
        f"Averaging {TRIES} waits cuts the noise to about {spread:.0f} ms",
    ]
    missing = [text for text in quoted if text not in lesson]
    assert not missing, missing

    # the figures: the key, the stored answer and the hit in the two cache figures, and the bars and their numbers in the wait figure
    text_a = question(a, 0)
    in_figs = [
        ("shared-shelf.svg", f"1. User {a.name} asks: {text_a}"), ("shared-shelf.svg", f"2. User {b.name} asks: {question(b, 0)}"),
        ("shared-shelf.svg", f"Key: {cache_key(a, text_a, False)}"), ("shared-shelf.svg", f"Kept: {answer(a, 0)}"),
        ("shared-shelf.svg", f"Hit. {b.name} gets Order {a.order}, not {b.order}."),
        ("user-in-key.svg", f"Key: {cache_key(a, text_a, True)}"), ("user-in-key.svg", f"Key: {cache_key(b, question(b, 0), True)}"),
        ("user-in-key.svg", f"Kept: {answer(a, 0)}"), ("user-in-key.svg", f"Kept: {answer(b, 0)}"),
        ("wait-by-guess.svg", f"Wait in ms: average of {TRIES} tries, {EXPERIMENTS:,} runs"),
        ("wait-by-guess.svg", " ".join(g + (" (real)" if i == RIGHT else "") for i, g in enumerate(GUESSES))),
    ]
    missing_figs = [(name, text) for name, text in in_figs if text not in figures[name]]
    assert not missing_figs, missing_figs
    waits_svg = (LESSON / "figs" / "wait-by-guess.svg").read_text(encoding="utf-8")
    averages = [round(w) for w in sh["average"] + us["average"]]
    assert [int(w) for w in re.findall(r">(\d+)</text>", waits_svg)] == averages
    assert [int(w) for w in re.findall(r'<rect x="190" [^>]*?width="(\d+)"', waits_svg)] == [round(w * 2.4) for w in averages]


def test_a_leak_is_another_users_private_fact():
    """Guards counting a leak wrongly: a leak must be a hit that hands one user another user's order code or street,
    and the opening-hours answer, the same for everyone, must never count."""
    by_name = {u.name: u for u in USERS}
    rng = random.Random(7)
    sequence = [(rng.choice(USERS).name, rng.randrange(len(QUESTIONS))) for _ in range(300)]
    steps = simulate(False, sequence)
    leaks = [s for s in steps if s.leak]
    assert len(leaks) > 100
    for s in leaks:
        assert s.hit and s.served != s.right
        others = [u for u in USERS if u.name != s.user]
        assert any(u.order in s.served or u.street in s.served for u in others)
        assert by_name[s.user].order not in s.served and by_name[s.user].street not in s.served
    asked_public = [i for i, (_, q) in enumerate(sequence) if q == PUBLIC]
    assert asked_public and not any(steps[i].leak for i in asked_public)


def test_the_user_in_the_key_removes_every_leak_and_lowers_the_hit_count():
    """Guards the lesson's claim for any request order, not only the fixed one: no leak with the user in the key, and never more hits."""
    rng = random.Random(11)
    for _ in range(40):
        sequence = [(rng.choice(USERS).name, rng.randrange(len(QUESTIONS))) for _ in range(rng.randrange(1, 60))]
        shared, per_user = totals(simulate(False, sequence)), totals(simulate(True, sequence))
        assert per_user["leaks"] == 0 and per_user["hits"] <= shared["hits"]
    assert all(s.served == s.right for s in simulate(True))


def test_a_stored_start_is_reused_whole_or_not_at_all_and_only_under_the_same_salt():
    """Guards the timing toy's cache rules (lesson 8's rule, plus the salt): a wrong note reuses nothing, a right note under another salt reuses nothing."""
    cache = PromptCache()
    cache.store("A", prompt(GUESSES[RIGHT])[:INSTRUCTION + NOTE])
    assert cache.reused("A", prompt(GUESSES[RIGHT])) == INSTRUCTION + NOTE
    assert cache.reused("A", prompt(GUESSES[0])) == 0          # shares the instruction, but not the whole stored start
    assert cache.reused("B", prompt(GUESSES[RIGHT])) == 0


def test_the_wait_shows_a_hit_only_without_the_user_in_the_key():
    """Guards the timing view: shared, only the right guess is quick; salted, all three guesses wait exactly the same."""
    shared, salted = ideal_waits(False), ideal_waits(True)
    miss = wait_ms(INSTRUCTION + NOTE + 20, 0)
    assert shared[RIGHT] < 10 and miss > 50 and [w for i, w in enumerate(shared) if i != RIGHT] == [miss, miss]
    assert salted == [miss, miss, miss]


def test_noise_hides_one_try_and_many_tries_find_the_guess_unless_the_user_is_in_the_key():
    """Guards the lesson's numbers: shared key, the fastest of one try is right about two times in three and of 25 tries almost always;
    with the user in the key both stay at chance, one in three. Also pins the fixed seed's outcome."""
    shared, salted = timing(False), timing(True)
    assert (shared["one"], shared["many"]) == (68, 99)
    assert (salted["one"], salted["many"]) == (33, 33)
    assert timing(False) == shared
    assert max(salted["average"]) - min(salted["average"]) < 1 and shared["average"][1] < min(shared["average"][0], shared["average"][2]) - 30

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

from embed_widget import WIDGET, config, embed
from leak_sim import EXPERIMENTS, GUESSES, RIGHT, TRIES, simulate, timing, totals

HTML = WIDGET.read_text(encoding="utf-8")
FIGS = sorted((WIDGET.parent / "figs").glob("*.svg"))

# 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="lk-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 {simulate, totals, percent};')(cfg);
const bad = [];
fx.modes.forEach((m) => {
  const steps = api.simulate(m.perUser);
  if (JSON.stringify(steps) !== JSON.stringify(m.steps)) bad.push(['steps', m.perUser]);
  if (JSON.stringify(api.totals(steps)) !== JSON.stringify(m.totals)) bad.push(['totals', m.perUser]);
});
fx.percents.forEach(([a, b, want]) => { if (api.percent(a, b) !== want) bad.push(['percent', a, b]); });
console.log(JSON.stringify({ modes: fx.modes.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 fx = JSON.parse(fs.readFileSync(fixture, 'utf8'));
const cfgText = html.match(/<script type="application\/json" id="lk-config">([\s\S]*?)<\/script>/)[1];
const code = html.match(/<script>([\s\S]*?)<\/script>/)[1];

function el(tag) {
  const e = { tag, children: [], listeners: {}, attrs: {}, checked: false, style: {}, _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.attrs[k] = String(v); };
  Object.defineProperty(e, 'textContent', { get: () => e._text, set: (v) => { e._text = v; e.children = []; } });
  return e;
}
function open() {
  const page = {};
  ['lk-config', 'lk-live', 'lk-key', 'lk-keyline', 'lk-big', 'lk-detail', 'lk-first', 'lk-looked', 'lk-sent', 'lk-right', 'lk-label', 'lk-grid',
   'lk-tnote', 'lk-rows', 'lk-pick', 'lk-why'].forEach((id) => { page[id] = el('x'); });
  page['lk-config']._text = cfgText;
  const doc = { getElementById: (id) => page[id], createElement: el, documentElement: { setAttribute() {} } };
  new Function('document', 'window', 'parent', code)(doc, { addEventListener() {} }, {});
  return page;
}
const fire = (e) => (e.listeners.change || e.listeners.click || []).forEach((f) => f());
const chips = (page) => page['lk-grid'].children;
const bad = [];
const want = (what, got, expected) => { if (got !== expected) bad.push([what, got, expected]); };

const start = open();
want('opening: headline', start['lk-big']._text, fx.modes[0].big);
want('opening: first line names the leak', start['lk-first']._text, fx.opening);
want('opening: the leak is marked in the panel', start['lk-detail'].attrs['data-state'], 'leak');
want('opening: chip count', chips(start).length, 14);
want('opening: the switch starts off', start['lk-key'].checked, false);
want('opening: the opening request is pressed', chips(start).map((c) => c.attrs['aria-pressed']).join(), fx.modes[0].pressedAtOpening);
want('opening: timing note carries the run count', start['lk-tnote']._text.includes('2,000 runs'), true);

fx.modes.forEach((m) => {
  const page = open();
  page['lk-key'].checked = m.perUser;
  fire(page['lk-key']);
  want('headline, per user ' + m.perUser, page['lk-big']._text, m.big);
  want('key line, per user ' + m.perUser, page['lk-keyline']._text, m.keyline);
  want('spoken update after the switch, per user ' + m.perUser, page['lk-live']._text, m.requests[1].live);
  want('chip labels, per user ' + m.perUser, chips(page).map((c) => c._text).join('|'), m.chips.join('|'));
  want('chip states, per user ' + m.perUser, chips(page).map((c) => c.attrs['data-state']).join(), m.states.join());
  m.requests.forEach((r, i) => {
    fire(chips(page)[i]);
    ['first', 'looked', 'sent', 'right', 'live'].forEach((k) => want('request ' + (i + 1) + ' ' + k + ', per user ' + m.perUser, page['lk-' + k]._text, r[k]));
    want('request ' + (i + 1) + ' is the only one pressed', chips(page).filter((c) => c.attrs['aria-pressed'] === 'true').length === 1 && chips(page)[i].attrs['aria-pressed'] === 'true', true);
  });
  want('timing rows, per user ' + m.perUser, page['lk-rows'].children.map((r) => r.children[0]._text).join('|'), m.rows.join('|'));
  want('timing bar widths, per user ' + m.perUser, page['lk-rows'].children.map((r) => r.children[1].style.width).join('|'), m.bars.join('|'));
  want('timing pick, per user ' + m.perUser, page['lk-pick']._text, m.pick);
  want('timing reason, per user ' + m.perUser, page['lk-why']._text, m.why);
});
console.log(JSON.stringify({ modes: fx.modes.length, bad: bad.slice(0, 5) }));
process.exit(bad.length ? 1 : 0);
"""


def run_node(tmp_path, name, script, *args):
    (tmp_path / name).write_text(script, encoding="utf-8")
    result = subprocess.run(["node", str(tmp_path / name), *args], capture_output=True, text=True, timeout=60)
    assert result.returncode == 0, result.stdout + result.stderr
    return json.loads(result.stdout)


def label(s):
    return f"{s.n} {s.user} " + ("LEAK" if s.leak else "hit" if s.hit else "miss")


def first_line(steps, i):
    s = steps[i]
    asked = f'User {s.user} asked "{s.question}"'
    if s.leak:
        return f"Leak: {asked} and got User {steps[s.written_by - 1].user}'s answer."
    if s.hit:
        return f"{asked} and got the answer saved at request {s.written_by}. It was right."
    return f"{asked}. Nothing was saved under this key, so the bot answered and saved it."


def test_widget_data_block_matches_leak_sim():
    """Guards a leak_sim.py edit that was not copied into the widget (run embed_widget.py)."""
    assert embed(HTML) == 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="lk-root">'):HTML.index("<script")]
    lookups = set(re.findall(r"\$\('([\w-]+)'\)", HTML[HTML.index("<script>"):]))
    assert lookups and all(f'id="{i}"' in root for i in lookups - {"lk-config"}), sorted(i for i in lookups if f'id="{i}"' not in root)


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="lk-config">.*?</script>', "", HTML, flags=re.S)
    for banned in ("http://", "https://", "fetch(", "XMLHttpRequest", "WebSocket", "sendBeacon", "import("):
        assert banned not in page, banned
    assert " src=" not in page and "@import" not in page and "<form" not in page


def test_widget_keeps_the_site_theme_listener_and_palette():
    assert "d.type !== 'spicyshark:theme'" in HTML and "e.source !== parent" 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)"):
        assert pair in HTML, pair
    assert "prefers-reduced-motion: no-preference" in HTML


def test_widget_rules_match_the_python_toy(tmp_path):
    """Guards the widget's JavaScript drifting from leak_sim.py: both keys, all 14 requests, every field, and the rounded percentages."""
    modes = []
    for per_user in (False, True):
        steps = simulate(per_user)
        modes.append({"perUser": per_user, "totals": totals(steps),
                      "steps": [{"n": s.n, "user": s.user, "question": s.question, "key": s.key, "hit": s.hit, "writtenBy": s.written_by,
                                 "served": s.served, "right": s.right, "leak": s.leak} for s in steps]})
    percents = [[a, b, (200 * a + b) // (2 * b)] for b in range(1, 40) for a in range(b + 1)]
    (tmp_path / "fixture.json").write_text(json.dumps({"modes": modes, "percents": percents}), encoding="utf-8")
    assert run_node(tmp_path, "parity.mjs", PARITY, str(tmp_path / "fixture.json"), str(WIDGET)) == {"modes": 2, "bad": []}


def test_widget_page_shows_what_the_lesson_names(tmp_path):
    """Guards the part a reader sees: it opens on a leak with its sentence, the headline counts, a word on every request (not colour alone),
    the detail of each request in both modes, the sentence spoken to a screen reader after each change, and the timing numbers and bar
    widths that follow the switch."""
    modes = []
    for per_user in (False, True):
        steps, t = simulate(per_user), totals(simulate(per_user))
        sim = timing(per_user)
        avg = [round(w) for w in sim["average"]]
        axis = config()["axis"]
        modes.append({
            "perUser": per_user,
            "big": f"{t['leaks']} of 14 answers went to the wrong user. Hit rate {t['rate']}% ({t['hits']} of 14).",
            "keyline": "Cache key: " + ("the user plus the question." if per_user else "the question alone."),
            "chips": [label(s) for s in steps], "states": ["leak" if s.leak else "hit" if s.hit else "miss" for s in steps],
            "pressedAtOpening": ",".join(["true" if s.n == 2 else "false" for s in steps]),
            "requests": [{"first": first_line(steps, i), "looked": "Key looked up: " + s.key, "sent": "Answer sent: " + s.served,
                          "right": f"Right answer for User {s.user}: {s.right}" if s.leak else "",
                          "live": f"{t['leaks']} of 14 answers went to the wrong user. " + first_line(steps, i)} for i, s in enumerate(steps)],
            "rows": [g + (" (the real street)" if i == RIGHT else "") + f": {avg[i]} ms" for i, g in enumerate(GUESSES)],
            "bars": [f"{a / axis * 100:.1f}%" for a in avg],
            "pick": f"Picking the fastest guess finds the real street {sim['one']}% of the time from one try each and {sim['many']}% from the average of {TRIES}. Guessing blind gets 33%.",
            "why": "Every guess waits the same, so the wait tells User B nothing." if per_user else "The real street waits less, because User A’s stored start is reused for User B."})
    assert modes[0]["chips"][1] == "2 B LEAK" and sum("LEAK" in c for c in modes[0]["chips"]) == 6 and not any("LEAK" in c for c in modes[1]["chips"])
    fixture = {"modes": modes, "opening": 'Leak: User B asked "Where is my order?" and got User A\'s answer.'}
    (tmp_path / "fixture.json").write_text(json.dumps(fixture), encoding="utf-8")
    assert EXPERIMENTS == 2000
    assert run_node(tmp_path, "page.mjs", PAGE_TEST, str(WIDGET), str(tmp_path / "fixture.json")) == {"modes": 2, "bad": []}


def test_figures_are_valid_and_stay_on_the_site_palette():
    """Guards a figure that breaks the site's look or its checks: every colour in a figure is one of the site's, and each has a title,
    a description, a viewBox and a white background."""
    allowed = {"#0a1626", "#44566b", "#b36200", "#00855f", "#d2dae4", "#ebf0f6", "#ffffff", "none"}
    assert 3 <= len(FIGS) <= 5
    for path in FIGS:
        svg = path.read_text(encoding="utf-8")
        used = {c.lower() for c in re.findall(r'(?:fill|stroke)="([^"]+)"', svg)}
        assert used <= allowed, (path.name, used - allowed)
        assert "viewBox=" in svg and "<title>" in svg and "<desc>" in svg and 'fill="#ffffff"' in svg
        assert "<script" not in svg and "href=" not in svg and "<image" not in svg