#!/usr/bin/env python3
"""Executable system-design learning models. Python 3.10+, standard library only.

Run: python3 concepts.py <model> or python3 concepts.py test

These finite simulations/SQLite transactions specify exactly what is tested.
They are NOT production implementations of distributed consensus, durable queueing,
payments, geographic routing, or collaboration. SQLite examples use one process and
one database. Assertions verify the stated invariants, not unmodeled network faults.
"""
from __future__ import annotations

import argparse
from array import array
from bisect import bisect_left
from collections import Counter, deque
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
import hashlib
import itertools
import json
import math
import random
import sqlite3
from statistics import NormalDist
import threading
import time


def ring_owner(tokens, position):
    """Pre: nonempty sorted unique numeric tokens. Post: first token >= position,
    wrapping to the first token. The caller maps tokens to physical owners."""
    if not tokens or tokens != sorted(set(tokens)):
        raise ValueError("nonempty sorted unique tokens required")
    return tokens[bisect_left(tokens, position) % len(tokens)]


def hashing():
    before, after = [10, 40, 80], [10, 25, 40, 80]
    changed = [p for p in range(100) if ring_owner(before, p) != ring_owner(after, p)]
    assert changed == list(range(11, 26))
    assert ring_owner(before, 40) == 40 and ring_owner(before, 99) == 10
    print("Changed positions:", changed, "(only interval (10,25])")


@dataclass
class Bucket:
    """Single-owner model. Invariant: 0 <= tokens <= capacity; timestamp never
    decreases. Not a shared distributed limiter; no failover/durability model."""
    capacity: float
    rate: float
    timestamp: float = 0.0
    tokens: float | None = None

    def __post_init__(self):
        if not all(math.isfinite(x) for x in (self.capacity, self.rate, self.timestamp)):
            raise ValueError("finite parameters required")
        if self.capacity <= 0 or self.rate <= 0:
            raise ValueError("positive capacity and refill rate required")
        self.tokens = self.capacity

    def allow(self, now, cost=1.0):
        """Post: admitted requests spend cost; rejected requests spend nothing.
        Clock rollback cannot refill or move stored time backward."""
        if not math.isfinite(now) or not math.isfinite(cost) or not 0 < cost <= self.capacity:
            raise ValueError("finite timestamp and 0 < cost <= capacity required")
        elapsed = max(0.0, now - self.timestamp)
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.timestamp = max(self.timestamp, now)
        if self.tokens < cost:
            return False
        self.tokens -= cost
        return True


def bucket():
    b = Bucket(10, 1)
    assert all(b.allow(0) for _ in range(10)) and not b.allow(0)
    assert all(b.allow(3) for _ in range(3)) and not b.allow(3)
    assert not b.allow(2) and not b.allow(3)  # rollback must not refill twice
    assert b.allow(4) and b.timestamp == 4
    assert 0 <= b.tokens <= b.capacity
    print("Initial burst 10; refill 3; rollback does not mint tokens.")


def pack_id(elapsed_ms, worker, sequence):
    """Encode the 41/10/12 unsigned fields. Not an allocator: callers must
    ensure exclusive worker ownership and restart-safe nonreuse of field tuples."""
    for value, bits in ((elapsed_ms, 41), (worker, 10), (sequence, 12)):
        if type(value) is not int or not 0 <= value < 1 << bits:
            raise ValueError(f"integer field must fit {bits} bits")
    return (elapsed_ms << 22) | (worker << 12) | sequence


class Snowflake:
    """Bounded-wait, single-process model. Pre: this worker is exclusively
    owned and no previous process used these timestamp/sequence tuples.
    NOT restart-safe: production allocation must enforce those preconditions.
    State changes only after a valid tuple can be returned; failures preserve it.
    """
    def __init__(self, worker, epoch_ms=1577836800000, wall_ms=None,
                 monotonic=time.monotonic, pause=time.sleep, wait_seconds=0.01):
        pack_id(0, worker, 0)  # validate worker, not an optimizable assert
        if type(epoch_ms) is not int or not math.isfinite(wait_seconds) or wait_seconds <= 0:
            raise ValueError("integer epoch and finite positive wait required")
        self.worker, self.epoch = worker, epoch_ms
        self.wall = wall_ms or (lambda: time.time_ns() // 1_000_000)
        self.monotonic, self.pause, self.wait = monotonic, pause, wait_seconds
        self.last, self.sequence = -1, 0
        self.lock = threading.Lock()

    def next_id(self):
        with self.lock:
            deadline = self.monotonic() + self.wait
            target = self.last + (self.sequence == 4095)
            # Invariant: generation state is unchanged until a valid tuple exists.
            # Bound: at most 1000 polls, also bounded by a local elapsed deadline.
            for _ in range(1000):
                timestamp = self.wall() - self.epoch
                pack_id(timestamp, self.worker, 0)  # timestamp horizon/type
                if timestamp >= target:
                    sequence = self.sequence + 1 if timestamp == self.last else 0
                    value = pack_id(timestamp, self.worker, sequence)
                    self.last, self.sequence = timestamp, sequence
                    return value
                remaining = deadline - self.monotonic()
                if remaining <= 0:
                    break
                self.pause(min(remaining, 0.0001))
            raise TimeoutError("clock rollback or exhausted millisecond; state unchanged")


def ids():
    value = pack_id(123456789, (3 << 5) | 7, 42)
    assert value == 517815304351786
    assert (value >> 22, (value >> 12) & 1023, value & 4095) == (123456789, 103, 42)
    assert pack_id(100, 3, 0) == pack_id(100, 3, 0)  # unsafe restart repeats a tuple
    assert pack_id(100, 3, 0) != pack_id(100, 7, 0)
    for args in ((-1, 0, 0), (1 << 41, 0, 0), (0, 1024, 0), (0, 0, 4096)):
        try:
            pack_id(*args)
        except ValueError:
            pass
        else:
            raise AssertionError("invalid field accepted")
    print("ID", value, "decodes to elapsed_ms=123456789, dc=3, worker=7, sequence=42")
    print("41-bit horizon:", round((1 << 41) / (1000 * 86400 * 365.2425), 2), "years")
    ticks = {"wall": 100, "elapsed": 0.0}
    def pause(seconds):
        ticks["elapsed"] += seconds
    generator = Snowflake(3, epoch_ms=0, wall_ms=lambda: ticks["wall"],
                          monotonic=lambda: ticks["elapsed"], pause=pause)
    issued = [generator.next_id() for _ in range(4096)]
    assert len(set(issued)) == 4096 and issued == sorted(issued)
    state = generator.last, generator.sequence
    try:
        generator.next_id()
    except TimeoutError:
        pass
    else:
        raise AssertionError("exhausted millisecond accepted")
    assert state == (generator.last, generator.sequence)
    ticks["wall"] = 101
    assert generator.next_id() > issued[-1]
    print("4096 sequence values unique; bounded exhaustion preserves state; next millisecond resumes.")


ALPHABET = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"


def encode(n):
    """Post: canonical base62 of a nonnegative integer. Loop invariant: the
    reversed emitted digits and remaining n represent the input. Bound: n."""
    if type(n) is not int or n < 0:
        raise ValueError("nonnegative integer required")
    if n == 0:
        return "0"
    digits = []
    while n:
        n, digit = divmod(n, 62)
        digits.append(ALPHABET[digit])
    return "".join(reversed(digits))


def decode(text):
    if not text or len(text) > 32:
        raise ValueError("nonempty bounded text required")
    result = 0
    for char in text:
        result = result * 62 + ALPHABET.index(char)
    return result


def encoding():
    for n in [0, 1, 61, 62, 125, 10**9, (1 << 63) - 1]:
        assert decode(encode(n)) == n
    for invalid in ("", "!", "-1"):
        try:
            decode(invalid)
        except ValueError:
            pass
        else:
            raise AssertionError("invalid code accepted")
    try:
        encode(-1)
    except ValueError:
        pass
    else:
        raise AssertionError("negative integer accepted")
    print("125 ->", encode(125), "; seven-character space:", 62**7)


class Frontier:
    """Single-thread/owner scheduling model, no network calls. At most one
    in-flight item per origin; next start >= completion + delay. Time is supplied."""
    def __init__(self, delay=1):
        if not math.isfinite(delay) or delay < 0:
            raise ValueError("finite nonnegative delay required")
        self.delay, self.queues, self.busy, self.next_at = delay, {}, set(), {}

    def submit(self, origin, url):
        self.queues.setdefault(origin, deque()).append(url)

    def take(self, now):
        for origin in sorted(self.queues):  # finite bound: number of origins
            queue = self.queues[origin]
            if queue and origin not in self.busy and now >= self.next_at.get(origin, 0):
                self.busy.add(origin)
                return origin, queue.popleft()
        return None

    def complete(self, origin, now):
        if origin not in self.busy:
            raise ValueError("origin has no in-flight fetch")
        self.busy.remove(origin)
        self.next_at[origin] = now + self.delay


def frontier():
    f = Frontier(1)
    f.submit("https://a.example:443", "A")
    f.submit("https://a.example:443", "B")
    assert f.take(0)[1] == "A"
    assert f.take(1) is None and f.take(4) is None
    f.complete("https://a.example:443", 5)
    assert f.take(5.9) is None and f.take(6)[1] == "B"
    n = 10**10
    pairs = n * (n-1) / (2 * 2**64)
    print("A runs t=0..5; B cannot start until t=6.")
    print("64-bit collision pairs at 10B IDs:", round(pairs, 3), "; collision chance:", f"{1-math.exp(-pairs):.1%}")


class Contributions:
    """One SQLite database/connection protected by a process mutex. Atomic
    ID claim + contribution + result; rollback leaves neither claim nor effect.
    No external effect, no distributed replication, and no expiring dedup state."""
    def __init__(self):
        self.db = sqlite3.connect(":memory:", isolation_level=None, check_same_thread=False)
        self.db.executescript("CREATE TABLE events(id TEXT PRIMARY KEY, amount INTEGER);"
                              "CREATE TABLE total(n INTEGER NOT NULL); INSERT INTO total VALUES(0);")
        self.lock = threading.Lock()

    def apply(self, identity, amount, fail=False):
        if not identity or type(amount) is not int or amount < 0:
            raise ValueError("stable ID and nonnegative integer contribution required")
        with self.lock:
            self.db.execute("BEGIN IMMEDIATE")
            try:
                old = self.db.execute("SELECT amount FROM events WHERE id=?", (identity,)).fetchone()
                if old:
                    if old[0] != amount:
                        raise ValueError("same identity with different payload")
                    self.db.execute("COMMIT")
                    return False
                self.db.execute("INSERT INTO events VALUES(?,?)", (identity, amount))
                self.db.execute("UPDATE total SET n=n+?", (amount,))
                if fail:
                    raise RuntimeError("injected failure before commit")
                self.db.execute("COMMIT")
                return True
            except BaseException:
                self.db.execute("ROLLBACK")
                raise

    def total(self):
        with self.lock:
            return self.db.execute("SELECT n FROM total").fetchone()[0]


def dedup():
    c = Contributions()
    try:
        c.apply("e1", 1, fail=True)
    except RuntimeError:
        pass
    assert c.total() == 0
    assert c.apply("e1", 1) and c.apply("e2", 1) and not c.apply("e1", 1)
    with ThreadPoolExecutor(max_workers=8) as pool:
        results = list(pool.map(lambda _: c.apply("e3", 1), range(20)))
    assert sum(results) == 1 and c.total() == 3
    try:
        c.apply("e1", 2)
    except ValueError:
        pass
    else:
        raise AssertionError("payload mismatch accepted")
    assert c.total() == 3
    c.db.close()
    print("Replay, 20 concurrent retries, rollback, and payload mismatch: correct local total 3.")


def hll_registers(items, p=12):
    """Fixed precision, stable 64-bit hash, one-byte registers (not packed six
    bits). Invariant: each register is the max rank of inserted hash suffixes."""
    if not 4 <= p <= 16:
        raise ValueError("precision 4..16 required for this model")
    registers, width = bytearray(1 << p), 64-p
    for item in items:
        digest = hashlib.blake2b(str(item).encode(), digest_size=8).digest()
        h = int.from_bytes(digest, "big")
        index, rest = h >> width, h & ((1 << width)-1)
        rank = width - rest.bit_length() + 1
        registers[index] = max(registers[index], rank)
    return registers


def hll_estimate(registers):
    m = len(registers)
    alpha = 0.7213 / (1 + 1.079 / m)
    raw = alpha * m * m / sum(2.0 ** -r for r in registers)
    zeros = registers.count(0)
    return m * math.log(m / zeros) if zeros and raw <= 2.5*m else raw


def hll():
    a = hll_registers(range(10000))
    assert a == hll_registers(itertools.chain(range(10000), range(10000)))
    b = hll_registers(range(5000, 15000))
    merged = bytearray(max(x, y) for x, y in zip(a, b))
    assert merged == hll_registers(range(15000))
    estimate = hll_estimate(merged)
    assert hll_estimate(hll_registers([])) == 0
    print(f"Union exact size 15000; HLL estimate {estimate:.0f}, observed error {(estimate/15000-1):+.2%}")
    print(f"Model register bytes {len(a)}; expected relative standard error {1.04/math.sqrt(len(a)):.2%}")


class CountMin:
    """Finite insertion-only model with stable row-salted hash. No proof that
    this concrete digest family provides the theorem's ideal independence."""
    def __init__(self, width=256, depth=4):
        if width <= 0 or depth <= 0:
            raise ValueError("positive dimensions required")
        self.width = width
        self.rows = [array('Q', [0]) * width for _ in range(depth)]

    def columns(self, item):
        for r in range(len(self.rows)):
            h = hashlib.blake2b(f"{r}:{item}".encode(), digest_size=8).digest()
            yield r, int.from_bytes(h, "big") % self.width

    def add(self, item, weight=1):
        if type(weight) is not int or weight < 0:
            raise ValueError("nonnegative integer weight required")
        cells = list(self.columns(item))
        if any(self.rows[r][col] + weight >= 1 << 64 for r, col in cells):
            raise OverflowError("counter horizon exceeded")
        for r, col in cells:
            self.rows[r][col] += weight

    def estimate(self, item):
        return min(self.rows[r][col] for r, col in self.columns(item))


def topk():
    one, two = Counter(a=10, x=9), Counter(b=10, x=9)
    candidates = {one.most_common(1)[0][0], two.most_common(1)[0][0]}
    total = one + two
    assert total.most_common(1) == [("x", 18)] and "x" not in candidates
    cms = CountMin()
    for name, weight in total.items():
        cms.add(name, weight)
    assert all(cms.estimate(k) >= v for k, v in total.items())
    print("Local top-1 candidates", sorted(candidates), "; real global winner", total.most_common(1))


def rollups():
    groups = [[10], [20, 20, 20]]
    weighted = sum(map(sum, groups)) / sum(map(len, groups))
    wrong = sum(sum(g)/len(g) for g in groups) / len(groups)
    assert weighted == 17.5 and wrong == 15
    snapshots = [100, 120, 5, 15]
    observed_increase = sum(b-a if b >= a else b for a, b in zip(snapshots, snapshots[1:]))
    assert observed_increase == 35  # reset-to-zero assumption, missed resets unmodeled
    print("Weighted mean", weighted, "; mean-of-means", wrong, "; reset-aware observed increase", observed_increase)


class Inbox:
    """Sequences start at 1; payload identity is immutable. Cursor covers
    exactly the highest contiguous stored prefix, not max sequence observed."""
    def __init__(self):
        self.entries, self.cursor = {}, 0

    def receive(self, sequence, payload):
        if type(sequence) is not int or sequence <= 0:
            raise ValueError("positive sequence required")
        if sequence in self.entries and self.entries[sequence] != payload:
            raise ValueError("sequence payload mismatch")
        self.entries[sequence] = payload
        # Invariant: all sequence numbers <= cursor exist. Bound: stored count.
        while self.cursor+1 in self.entries:
            self.cursor += 1


def delivery():
    box = Inbox()
    for seq in range(1, 11):
        box.receive(seq, f"m{seq}")
    box.receive(12, "m12")
    assert box.cursor == 10
    box.receive(11, "m11")
    box.receive(12, "m12")
    assert box.cursor == 12 and len(box.entries) == 12
    print("After 12 arrived cursor stayed 10; after gap 11 arrived cursor became 12.")


@dataclass
class Resource:
    """Single-owner atomic-call model. Highest token is assumed persistent;
    real check+effect must use one transactional/conditional storage operation."""
    fence: int = 0
    value: str = ""

    def write(self, token, value):
        if type(token) is not int or token < 0:
            raise ValueError("nonnegative grant token required")
        if token < self.fence:
            return False
        self.fence, self.value = token, value
        return True


def fencing():
    r = Resource()
    assert r.write(33, "old")
    assert r.write(33, "still passes before fence 34 is installed")
    assert r.write(34, "new") and not r.write(33, "stale")
    assert r.value == "new" and r.fence == 34
    print("Token 33 rejected only after resource observed 34; final value", r.value)


def workflow():
    """Finite provider-key model: a lost reply is not a rolled-back effect.
    Real providers' idempotency scope, expiry, status lookup, and errors differ."""
    provider, local = {}, {"p42": "pending"}
    def charge(key, amount):
        if key in provider and provider[key] != amount:
            raise ValueError("provider operation payload mismatch")
        provider.setdefault(key, amount)
        return provider[key]
    charge("charge:p42", 1000)
    local["p42"] = "unknown"  # response lost after provider success
    assert len(provider) == 1 and local["p42"] == "unknown"
    charge("charge:p42", 1000)  # safe retry under this provider's model
    local["p42"] = "completed"
    assert len(provider) == 1
    provider["refund:p42"] = -1000  # distinct corrective operation
    assert sum(provider.values()) == 0
    print("Lost response kept unknown; same-key retry reused charge; refund has a distinct key.")


def ledger():
    """Local simplified signed-wallet ledger, not full account-type accounting."""
    db = sqlite3.connect(":memory:", isolation_level=None)
    db.executescript("CREATE TABLE accounts(id TEXT PRIMARY KEY,balance INTEGER CHECK(balance>=0));"
                     "INSERT INTO accounts VALUES('alice',1000),('bob',0);"
                     "CREATE TABLE tx(id TEXT PRIMARY KEY,amount INTEGER NOT NULL);"
                     "CREATE TABLE postings(tx TEXT,account TEXT,amount INTEGER);")
    def transfer(identity, amount):
        if type(amount) is not int or amount <= 0:
            raise ValueError("positive cents required")
        db.execute("BEGIN IMMEDIATE")
        try:
            previous = db.execute("SELECT amount FROM tx WHERE id=?", (identity,)).fetchone()
            if previous:
                if previous[0] != amount:
                    raise ValueError("same transfer ID with different amount")
                db.execute("COMMIT")
                return "replay"
            won = db.execute("UPDATE accounts SET balance=balance-? WHERE id='alice' AND balance>=?", (amount, amount)).rowcount
            if not won:
                raise ValueError("insufficient funds")
            db.execute("UPDATE accounts SET balance=balance+? WHERE id='bob'", (amount,))
            db.execute("INSERT INTO tx VALUES(?,?)", (identity, amount))
            db.executemany("INSERT INTO postings VALUES(?,?,?)", [(identity, "alice", -amount), (identity, "bob", amount)])
            db.execute("COMMIT")
            return "committed"
        except BaseException:
            db.execute("ROLLBACK")
            raise
    assert transfer("t1", 1000) == "committed" and transfer("t1", 1000) == "replay"
    try:
        transfer("t1", 500)
    except ValueError:
        pass
    else:
        raise AssertionError("transfer payload mismatch accepted")
    try:
        transfer("t2", 1000)
    except ValueError:
        pass
    assert db.execute("SELECT SUM(balance) FROM accounts").fetchone()[0] == 1000
    assert db.execute("SELECT SUM(amount) FROM postings").fetchone()[0] == 0
    assert dict(db.execute("SELECT id,balance FROM accounts")) == {"alice": 0, "bob": 1000}
    print("Alice 0, Bob 1000 cents; balanced postings; second spend rejected.")
    db.close()


def seats():
    db = sqlite3.connect(":memory:")
    db.execute("CREATE TABLE seats(id TEXT PRIMARY KEY,status TEXT,holder TEXT,hold_id TEXT,expires INTEGER,version INTEGER)")
    db.execute("INSERT INTO seats VALUES('S','available',NULL,NULL,0,0)")
    def claim(user, hold, now):
        return db.execute("UPDATE seats SET status='held',holder=?,hold_id=?,expires=?,version=version+1 WHERE id='S' AND status='available'", (user, hold, now+300)).rowcount
    def book(user, hold, now):
        return db.execute("UPDATE seats SET status='booked',version=version+1 WHERE id='S' AND status='held' AND holder=? AND hold_id=? AND expires>?", (user, hold, now)).rowcount
    assert claim("U", "H1", 0) == 1 and claim("V", "other", 0) == 0
    assert book("U", "H1", 300) == 0  # expiry boundary, even before sweeper
    db.execute("UPDATE seats SET status='available',version=version+1 WHERE hold_id='H1' AND expires<=300")
    assert claim("U", "H2", 301) == 1
    assert book("U", "H1", 302) == 0 and book("U", "H2", 302) == 1
    db.commit()
    print("One claimant; expired H1 and stale same-user callback rejected; H2 booked.")
    db.close()


def auction():
    bids, reserve = [500, 400, 200], 100  # integer cents in a single-item model
    price = max(reserve, sorted(bids, reverse=True)[1])
    assert price == 400
    truthful, shaded = 0.10*(5-4), 0.05*(5-2)
    assert math.isclose(truthful, 0.10) and math.isclose(shaded, 0.15) and shaded > truthful
    print("Exact second price 400 cents; GSP truthful utility $0.10/view, shaded $0.15/view.")


def nearby(points, x, y, radius, cell_size=100):
    """Planar metres only. Conservative rectangular cell covering, then exact
    Euclidean refinement. Not latitude/longitude or a geodesic/road-ETA service."""
    if not all(math.isfinite(v) for v in (x, y, radius, cell_size)) or radius < 0 or cell_size <= 0:
        raise ValueError("finite position, nonnegative radius, positive cell size required")
    index = {}
    for identity, px, py in points:
        cell = math.floor(px/cell_size), math.floor(py/cell_size)
        index.setdefault(cell, []).append((identity, px, py))
    result = []
    for cx in range(math.floor((x-radius)/cell_size), math.floor((x+radius)/cell_size)+1):
        for cy in range(math.floor((y-radius)/cell_size), math.floor((y+radius)/cell_size)+1):
            for identity, px, py in index.get((cx, cy), []):
                distance = math.hypot(px-x, py-y)
                if distance <= radius:
                    result.append((distance, identity))
    return sorted(result)


def geo():
    points = [("across", 101, 50), ("far", 340, 50), ("other", -120, 50)]
    assert nearby(points, 99, 50, 5) == [(2.0, "across")]
    for radius in (0, 5, 100, 250):
        exact = sorted((math.hypot(px-99, py-50), name) for name, px, py in points if math.hypot(px-99, py-50) <= radius)
        assert nearby(points, 99, 50, radius) == exact
    print("Boundary neighbor found; expanded radius matches brute force. Units: planar metres.")


def prefix_map(counts, k=3):
    """Demo assumes already-normalized, allowed strings and nonnegative counts.
    Tie order: descending frequency then ascending string. Finite input only."""
    if type(k) is not int or k <= 0:
        raise ValueError("positive integer k required")
    buckets = {}
    for query, count in counts.items():
        if not query or type(count) is not int or count < 0:
            raise ValueError("nonempty strings and nonnegative integer counts required")
        for i in range(1, len(query)+1):
            buckets.setdefault(query[:i], []).append((-count, query))
    return {p: [q for _, q in sorted(values)[:k]] for p, values in buckets.items()}


def autocomplete():
    m = prefix_map({"best": 90, "bee": 50, "beer": 40, "bet": 10})
    assert m["be"] == ["best", "bee", "beer"] and m["bee"] == ["bee", "beer"]
    assert m.get("zzz", []) == []
    assert prefix_map({"ab": 1, "aa": 1})["a"] == ["aa", "ab"]
    print("be ->", m["be"], "; bee ->", m["bee"])


def projection():
    """Rebuild from finite ordered shipment observations. A strict legal gate
    with arrival-order-only processing misses a deferred successor."""
    events = [(2, "in_transit"), (1, "picked_up")]
    transition = {("created", "picked_up"): "picked_up", ("picked_up", "in_transit"): "in_transit"}
    arrived_state = "created"
    for _, code in events:
        arrived_state = transition.get((arrived_state, code), arrived_state)
    rebuilt = "created"
    for _, code in sorted(events):
        rebuilt = transition.get((rebuilt, code), rebuilt)
    assert arrived_state == "picked_up" and rebuilt == "in_transit"
    edge = (2, False)
    old_follow = (1, True)
    edge = max(edge, old_follow, key=lambda e: e[0])
    assert edge == (2, False)
    print("Arrival-only gate:", arrived_state, "; ordered replay:", rebuilt, "; stale follow rejected.")


def collaboration():
    """One-character insert-only OT diamond; not a full OT control protocol.
    Also demonstrate a two-phase set's deletion merge, not a text-sequence CRDT."""
    a, b = (1, "alice", "X"), (1, "bob", "Y")
    def transform(op, other):
        pos, site, char = op
        if (other[0], other[1]) < (pos, site):
            pos += 1
        return pos, site, char
    def apply(text, op):
        pos, _, char = op
        if not 0 <= pos <= len(text):
            raise ValueError("position outside string")
        return text[:pos] + char + text[pos:]
    left = apply(apply("AB", a), transform(b, a))
    right = apply(apply("AB", b), transform(a, b))
    assert left == right == "AXYB"
    added_a, removed_a = {"x"}, {"x"}
    added_b, removed_b = {"x", "y"}, set()
    merged = (added_a | added_b) - (removed_a | removed_b)
    assert merged == {"y"}
    print("Insert-only OT diamond:", left, "; two-phase set keeps x deleted.")


def versions():
    """Atomic version-CAS local model. Real publication needs one storage
    transaction/conditional operation, not these statements across RPC calls."""
    state = {"version": 7, "value": "original"}
    def commit(base, value):
        if state["version"] != base:
            return False
        state.update(version=base+1, value=value)
        return True
    assert commit(7, "A") and not commit(7, "B")
    assert state == {"version": 8, "value": "A"}
    # Old-only writer after rollback changes the authoritative column.
    row = {"full_name": "Ada", "display_name": "Ada"}
    row["full_name"] = "Grace"
    assert row["display_name"] != row["full_name"]
    row["display_name"] = row["full_name"]  # required catch-up before recutover
    print("Only one base-7 commit wins; rollback writer requires recutover reconciliation.")


def fixed_chunks(data, size=1024):
    if type(size) is not int or size <= 0:
        raise ValueError("positive integer chunk size required")
    return [data[i:i+size] for i in range(0, len(data), size)]


def cdc_chunks(data, gear, mask=1023, minimum=256, maximum=4096):
    """Bounded educational gear-hash chunking, not a production CDC standard.
    Invariant: emitted nonempty chunks partition data[:start], each <= maximum.
    Bound: remaining input bytes. The last chunk may be shorter than minimum."""
    if len(gear) != 256 or minimum <= 0 or maximum < minimum or mask <= 0:
        raise ValueError("valid gear table and positive chunk bounds required")
    chunks, start, fingerprint = [], 0, 0
    for i, byte in enumerate(data):
        fingerprint = ((fingerprint << 1) + gear[byte]) & 0xffffffff
        size = i+1-start
        if size >= maximum or (size >= minimum and fingerprint & mask == 0):
            chunks.append(data[start:i+1])
            start, fingerprint = i+1, 0
    if start < len(data):
        chunks.append(data[start:])
    return chunks


def chunks():
    rng = random.Random(7)
    gear = [rng.getrandbits(32) for _ in range(256)]
    before = bytes(rng.getrandbits(8) for _ in range(64000))
    after = before[:1000] + b"hello, world" + before[1000:]
    for label, split in (("fixed", fixed_chunks), ("content-defined", lambda data: cdc_chunks(data, gear))):
        a, b = split(before), split(after)
        assert b"".join(a) == before and b"".join(b) == after
        assert all(0 < len(chunk) <= 4096 for chunk in b)
        known = {hashlib.sha256(chunk).digest() for chunk in a}
        changed = [chunk for chunk in b if hashlib.sha256(chunk).digest() not in known]
        print(label, "chunks", len(b), "; changed chunks", len(changed), "; changed bytes", sum(map(len, changed)))
    assert cdc_chunks(b"", gear) == []


def quorum():
    """Exhaustive finite set arithmetic; not a Raft/Paxos implementation."""
    majorities = [set(c) for c in itertools.combinations(range(5), 3)]
    assert all(a & b for a in majorities for b in majorities)
    for placement, safe in (("AAABC", False), ("AABBC", True)):
        spans = all(len({placement[n] for n in q}) >= 2 for q in majorities)
        survives = all(sum(p != region for p in placement) >= 3 for region in set(placement))
        assert spans == survives == safe
        print(placement, "every majority spans regions and survives one-region loss:", safe)
    entry_term, current_term, replicas = 4, 5, 3
    assert replicas >= 3 and not (replicas >= 3 and entry_term == current_term)
    print("A replicated old-term entry does not meet Raft's direct count-to-commit rule.")


def observability():
    budget_requests = 1_000_000_000 * (1-0.999)
    assert math.isclose(budget_requests, 1_000_000)
    assert math.isclose(30*24*60*(1-0.999), 43.2)
    assert math.isclose(0.02/0.001, 20) and math.isclose(14.4/720, 0.02)
    expected_retained_errors = 10*0.01
    assert expected_retained_errors == 0.1  # tail cannot recover head-discarded traces
    ids = [f"u{i}" for i in range(10000)]
    digests = {hashlib.sha256(x.encode()).digest() for x in ids}
    assert len(digests) == len(ids)
    print("Request budget 1M; separate time budget 43.2 min; error burn 20x.")
    print("Ten errors with 1% head sampling: expected 0.1 forwarded; hashing kept 10,000 label values.")


def search():
    logs = [{"day": "2026-10-18", "svc": "api", "level": "ERROR", "text": "timeout"},
            {"day": "2026-10-19", "svc": "api", "level": "INFO", "text": "ok"},
            {"day": "2026-10-19", "svc": "api", "level": "ERROR", "text": "timeout"}]
    postings = {}
    for index, row in enumerate(logs):
        for field in ("svc", "level"):
            postings.setdefault((field, row[field]), set()).add(index)
    matches = postings[("svc", "api")] & postings[("level", "ERROR")]
    assert matches == {0, 2}
    selected = [{field: row[field] for field in ("svc", "level")} for row in logs if row["day"] == "2026-10-19"]
    assert len(selected) == 2 and all(set(row) == {"svc", "level"} for row in selected)
    print("Inverted intersection IDs", sorted(matches), "; pruned/projected rows", selected)


def capacity():
    used, installed, utilization, growth = 6000, 10000, 0.75, 0.03
    usable = installed*utilization
    months = math.log(usable/used)/math.log(1+growth)
    order_in = months-3-0.5
    buy = max(0, math.ceil(used*(1+growth)**12/utilization)-installed)
    assert 7.54 < months < 7.56 and 4.04 < order_in < 4.06 and buy > 0
    sum_p90 = 10*NormalDist(100, 20).inv_cdf(0.9)
    p90_sum = NormalDist(1000, 20*math.sqrt(10)).inv_cdf(0.9)
    assert sum_p90 > p90_sum  # independent normal example only
    print(f"Usable {usable:.0f}; exhaustion {months:.2f} months; order in {order_in:.2f}; buy {buy} units.")
    print(f"Sum P90s {sum_p90:.0f}; P90 of independent sum {p90_sum:.0f}.")


class Clock:
    def __init__(self):
        self.now = 0.0

    def advance(self, seconds):
        if not math.isfinite(seconds) or seconds < 0:
            raise ValueError("finite nonnegative clock advance required")
        self.now += seconds


class WindowLimiter:
    """Single-thread exact admissions within (now-window,now], injected clock.
    Invariant: retained timestamps belong to the window; count <= limit."""
    def __init__(self, limit, window, clock):
        if type(limit) is not int or limit <= 0 or not math.isfinite(window) or window <= 0:
            raise ValueError("positive limit/window required")
        self.limit, self.window, self.clock = limit, window, clock
        self.calls = deque()

    def allow(self):
        now = self.clock.now
        while self.calls and self.calls[0] <= now-self.window:
            self.calls.popleft()  # finite bound: number of expired calls
        if len(self.calls) >= self.limit:
            return False
        self.calls.append(now)
        return True


def testing():
    clock = Clock()
    limiter = WindowLimiter(2, 60, clock)
    assert limiter.allow() and limiter.allow() and not limiter.allow()
    clock.advance(59)
    assert not limiter.allow()
    clock.advance(1)
    assert limiter.allow() and limiter.allow() and not limiter.allow()
    print("t=0: two allowed; t=59: rejected; t=60: old boundary expired, two allowed; no sleep.")


MODELS = {name: globals()[name] for name in (
    "hashing", "bucket", "ids", "encoding", "frontier", "dedup", "hll", "topk",
    "rollups", "delivery", "fencing", "workflow", "ledger", "seats", "auction",
    "geo", "autocomplete", "projection", "collaboration", "versions", "chunks", "quorum",
    "observability", "search", "capacity", "testing")}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("model", choices=["test", *MODELS])
    args = parser.parse_args()
    selected = MODELS.items() if args.model == "test" else [(args.model, MODELS[args.model])]
    for name, run in selected:
        print(f"\n[{name}]")
        run()
    print(f"\nPASS: {len(MODELS) if args.model == 'test' else 1} model(s), deterministic invariant checks.")


if __name__ == "__main__":
    main()