Teaching and audit implementation

"""Assessor-only variants. Select candidate package after its baseline is repaired."""
import argparse
import importlib
from pathlib import Path
import sys
import tempfile
import unittest

parser = argparse.ArgumentParser()
parser.add_argument("--package", choices=["reference", "starter"], default="reference")
args, remaining = parser.parse_known_args()
root = Path(__file__).resolve().parents[2] / "curriculum/02-applications/04-testing/labs/importer"
sys.path.insert(0, str(root))
run = importlib.import_module(args.package + ".importer").run
Store = importlib.import_module(args.package + ".store").Store
transport = importlib.import_module(args.package + ".transport")


def item(ident="held", amount="0.29"):
    return {"id":ident, "amount":amount, "currency":"USD"}


def page(items, cursor=None):
    return {"status":200, "body":{"items":items, "nextCursor":cursor}}


class HeldBack(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.path = str(Path(self.temp.name) / "held.db")
        self.store = Store(self.path)
        self.clock = transport.FakeClock()

    def tearDown(self):
        self.store.close()
        self.temp.cleanup()

    def test_multi_hop_cycle(self):
        server = transport.FakeServer({"START":[page([],"p2")],"p2":[page([],"p3")],"p3":[page([],"p2")]},self.clock)
        with self.assertRaisesRegex(ValueError,"cycle"):
            run(server,self.clock,self.store)
        self.assertEqual(len(server.calls),3)

    def test_retry_consumes_transport_budget(self):
        server = transport.FakeServer({"START":[{"status":429,"retryAfter":0.8,"elapsed":0.3}]},self.clock)
        with self.assertRaises(TimeoutError):
            run(server,self.clock,self.store,budget=1)
        self.assertEqual(len(server.calls),1)
        self.assertEqual(self.clock.waits,[])

    def test_crash_and_equal_decimal_spelling(self):
        server = transport.FakeServer({"START":[page([item(amount="0.290")])]},self.clock)
        def crash():
            raise RuntimeError("lost process")
        with self.assertRaises(RuntimeError):
            run(server,self.clock,self.store,after_persist=crash)
        self.store.close()
        self.store = Store(self.path)
        server = transport.FakeServer({"START":[page([item(amount="0.29")])]},self.clock)
        self.assertEqual(run(server,self.clock,self.store),[("held",29,"USD")])

    def test_late_malformed_page_preserves_progress(self):
        server = transport.FakeServer({"START":[page([item("first")],"next")],
                                      "next":[page([item("valid"),item("bad","0.00001")])]},self.clock)
        with self.assertRaises(ValueError):
            run(server,self.clock,self.store)
        self.assertEqual(self.store.rows(),[("first",29,"USD")])
        self.assertEqual(self.store.checkpoint(),("next",0))


if __name__ == "__main__":
    unittest.main(argv=[sys.argv[0]] + remaining)

import subprocess
import sys
import unittest
from pathlib import Path
from threading import Barrier, Event, Thread
from executor import Executor, Closed, Cancelled, DeadlineExceeded, NestedSubmission


class Tests(unittest.TestCase):
    def test_bounds_outcomes_and_exception_permit_release(self):
        pool = Executor()
        entered, release = Barrier(5), Event()
        def held(ctx):
            entered.wait(timeout=2)
            release.wait(2)
            return 7
        active = [pool.submit(held) for _ in range(4)]
        entered.wait(timeout=2)
        queued = [pool.submit(lambda ctx: 11) for _ in range(8)]
        self.assertEqual(pool.snapshot()["queued"], 8)
        outcome = []
        producer = Thread(target=lambda: outcome.append(pool.submit(lambda ctx: 13)))
        producer.start()
        with pool.cv:
            self.assertTrue(pool.cv.wait_for(lambda: pool.waiting_producers == 1, timeout=2))
        release.set()
        producer.join(2)
        self.assertFalse(producer.is_alive())
        def bad(ctx):
            raise ValueError("task failure")
        failed = pool.submit(bad)
        good = pool.submit(lambda ctx: "slot released")
        pool.shutdown(timeout=3)
        all_tasks = active + queued + outcome + [failed, good]
        for task in all_tasks:
            self.assertTrue(task.done.is_set())
            self.assertEqual(task.terminal_count, 1)
        with self.assertRaises(ValueError):
            failed.result()
        self.assertEqual(good.result(), "slot released")
        state = pool.snapshot()
        self.assertEqual(state["peak_active"], 4)
        self.assertEqual(state["peak_queued"], 8)
        self.assertEqual(state["accepted"], state["completed"])
        self.assertEqual(state["active"], 0)

    def test_shutdown_wakes_blocked_producer_and_cancels_queued(self):
        pool = Executor(workers=1, capacity=1)
        entered, release, rejected = Event(), Event(), Event()
        def held(ctx):
            entered.set()
            release.wait(2)
            return "finished"
        running = pool.submit(held)
        self.assertTrue(entered.wait(2))
        queued = pool.submit(lambda ctx: "must not execute")
        def producer():
            try:
                pool.submit(lambda ctx: "refused")
            except Closed:
                rejected.set()
        thread = Thread(target=producer)
        thread.start()
        with pool.cv:
            self.assertTrue(pool.cv.wait_for(lambda: pool.waiting_producers == 1, timeout=2))
        pool.shutdown(cancel_queued=True, wait=False)
        self.assertTrue(rejected.wait(2))
        with self.assertRaises(Cancelled):
            queued.result(2)
        release.set()
        pool.shutdown(timeout=2)
        self.assertEqual(running.result(), "finished")
        thread.join(2)
        self.assertEqual(pool.snapshot()["accepted"], 2)

    def test_logical_deadlines_and_running_cancel(self):
        now = [0.0]
        pool = Executor(workers=1, capacity=1, clock=lambda: now[0])
        entered, release = Event(), Event()
        def held(ctx):
            entered.set()
            release.wait(2)
        running = pool.submit(held, deadline=10)
        self.assertTrue(entered.wait(2))
        queued = pool.submit(lambda ctx: self.fail("expired task ran"), deadline=3)
        refused = Event()
        def producer():
            try:
                pool.submit(lambda ctx: None, deadline=2)
            except DeadlineExceeded:
                refused.set()
        thread = Thread(target=producer)
        thread.start()
        with pool.cv:
            self.assertTrue(pool.cv.wait_for(lambda: pool.waiting_producers == 1, timeout=2))
        now[0] = 4
        pool.wake()
        self.assertTrue(refused.wait(2))
        pool.cancel(running)
        self.assertFalse(running.done.is_set())  # cancellation does not free a running resource
        release.set()
        pool.shutdown(timeout=2)
        with self.assertRaises(Cancelled):
            running.result()
        with self.assertRaises(DeadlineExceeded):
            queued.result()
        thread.join(2)

    def test_nested_submission_rejected_without_deadlock(self):
        pool = Executor()
        task = pool.submit(lambda ctx: pool.submit(lambda inner: 1).result())
        with self.assertRaises(NestedSubmission):
            task.result(2)
        joining = pool.submit(lambda ctx: pool.shutdown())
        with self.assertRaises(NestedSubmission):
            joining.result(2)
        pool.shutdown(timeout=2)

    def test_cancel_queued_releases_admission_and_finishes_once(self):
        pool = Executor(workers=1, capacity=1)
        entered, release = Event(), Event()
        pool.submit(lambda ctx: (entered.set(), release.wait(2)))
        self.assertTrue(entered.wait(2))
        doomed = pool.submit(lambda ctx: self.fail("cancelled callable ran"))
        self.assertTrue(pool.cancel(doomed))
        self.assertFalse(pool.cancel(doomed))
        replacement = pool.submit(lambda ctx: "reused queue slot")
        release.set()
        pool.shutdown(timeout=2)
        with self.assertRaises(Cancelled):
            doomed.result()
        self.assertEqual(doomed.terminal_count, 1)
        self.assertEqual(replacement.result(), "reused queue slot")
        self.assertEqual(pool.snapshot()["accepted"], pool.snapshot()["completed"])

    def test_shutdown_timeout_does_not_claim_running_task_stopped(self):
        pool = Executor(workers=1)
        entered, release = Event(), Event()
        task = pool.submit(lambda ctx: (entered.set(), release.wait(2)))
        self.assertTrue(entered.wait(2))
        with self.assertRaises(TimeoutError):
            pool.shutdown(timeout=0)
        self.assertFalse(task.done.is_set())
        release.set()
        pool.shutdown(timeout=2)

    def test_watchdog_detects_deliberate_nested_wait_deadlock(self):
        process = subprocess.Popen([sys.executable, str(Path(__file__).with_name("deadlock_demo.py"))],
                                   stdout=subprocess.PIPE, text=True)
        try:
            # communicate has a watchdog; the deadlock itself is established by a Barrier.
            with self.assertRaises(subprocess.TimeoutExpired) as caught:
                process.communicate(timeout=1)
            output = caught.exception.output or b""
            if isinstance(output, bytes):
                output = output.decode()
            self.assertEqual(output.count("parent waiting for queued child"), 4)
        finally:
            process.kill()
            process.communicate(timeout=2)


if __name__ == "__main__":
    unittest.main()

"""Execute real two-session PostgreSQL schedules. Requires psycopg and LAB_PG_DSN."""
import os
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from retry import transaction_retry


def main():
    import psycopg
    dsn = os.environ["LAB_PG_DSN"]
    def connect():
        connection = psycopg.connect(dsn, autocommit=True)
        connection.execute("SET search_path=interview_lab")
        connection.execute("SET statement_timeout='8s'")
        connection.execute("SET lock_timeout='6s'")
        return connection

    with connect() as admin:
        admin.execute(Path(__file__).with_name("schema.sql").read_text())
        with connect() as a, connect() as b:
            # Lost update: both application reads see 1, then write the literal 0.
            a.execute("BEGIN")
            b.execute("BEGIN")
            assert a.execute("SELECT available FROM stock WHERE id=1").fetchone()[0] == 1
            assert b.execute("SELECT available FROM stock WHERE id=1").fetchone()[0] == 1
            for conn, buyer in ((a, "a"), (b, "b")):
                conn.execute("UPDATE stock SET available=0 WHERE id=1")
                conn.execute("INSERT INTO reservations VALUES (%s,1)", (buyer,))
                conn.execute("COMMIT")
            assert admin.execute("SELECT count(*) FROM reservations").fetchone()[0] == 2
            print("FAILURE reproduced: stock=0, reservations=2 for initial stock=1")
            admin.execute("TRUNCATE reservations")
            admin.execute("UPDATE stock SET available=1")
            a.execute("BEGIN")
            b.execute("BEGIN")
            assert a.execute("UPDATE stock SET available=available-1 WHERE id=1 AND available>0 RETURNING id").fetchone() == (1,)
            a.execute("INSERT INTO reservations VALUES ('a',1)")
            with ThreadPoolExecutor(max_workers=1) as pool:
                waiting = pool.submit(b.execute, "UPDATE stock SET available=available-1 WHERE id=1 AND available>0 RETURNING id")
                a.execute("COMMIT")
                assert waiting.result().fetchone() is None
            b.execute("COMMIT")
            assert admin.execute("SELECT count(*) FROM reservations").fetchone()[0] == 1
            assert admin.execute("SELECT available FROM stock WHERE id=1").fetchone()[0] == 0
            print("FIX passed: conditional decrement admits one reservation")

            # Write skew under snapshot isolation: disjoint writes evade row conflict.
            for conn in (a, b):
                conn.execute("BEGIN ISOLATION LEVEL REPEATABLE READ")
                assert conn.execute("SELECT count(*) FROM doctors WHERE on_call").fetchone()[0] == 2
            a.execute("UPDATE doctors SET on_call=false WHERE id=1")
            b.execute("UPDATE doctors SET on_call=false WHERE id=2")
            a.execute("COMMIT")
            b.execute("COMMIT")
            assert admin.execute("SELECT count(*) FROM doctors WHERE on_call").fetchone()[0] == 0
            print("FAILURE reproduced: repeatable-read write skew leaves zero doctors")
            admin.execute("UPDATE doctors SET on_call=true")
            for conn in (a, b):
                conn.execute("BEGIN ISOLATION LEVEL SERIALIZABLE")
                assert conn.execute("SELECT count(*) FROM doctors WHERE on_call").fetchone()[0] == 2
            a.execute("UPDATE doctors SET on_call=false WHERE id=1")
            b.execute("UPDATE doctors SET on_call=false WHERE id=2")
            failures = []
            for conn, doctor in ((a, 1), (b, 2)):
                try:
                    conn.execute("COMMIT")
                except psycopg.Error as error:
                    conn.execute("ROLLBACK")
                    assert error.sqlstate == "40001", error
                    failures.append(doctor)
            assert len(failures) == 1
            def retry_off_call():
                with connect() as conn:
                    try:
                        conn.execute("BEGIN ISOLATION LEVEL SERIALIZABLE")
                        count = conn.execute("SELECT count(*) FROM doctors WHERE on_call").fetchone()[0]
                        if count > 1:
                            conn.execute("UPDATE doctors SET on_call=false WHERE id=%s", (failures[0],))
                        conn.execute("COMMIT")
                        return count
                    except Exception:
                        conn.execute("ROLLBACK")
                        raise
            assert transaction_retry(retry_off_call) == 1
            assert admin.execute("SELECT count(*) FROM doctors WHERE on_call").fetchone()[0] == 1
            print("FIX passed: serialization abort plus complete transaction retry preserves one doctor")

            # Deadlock: opposite lock order. Victim rolls back before retry.
            a.execute("BEGIN")
            b.execute("BEGIN")
            a.execute("UPDATE counters SET hits=hits+1 WHERE id=1")
            b.execute("UPDATE counters SET hits=hits+1 WHERE id=2")
            def finish(conn, second_id):
                try:
                    conn.execute("UPDATE counters SET hits=hits+1 WHERE id=%s", (second_id,))
                    conn.execute("COMMIT")
                    return "ok"
                except psycopg.Error as error:
                    conn.execute("ROLLBACK")
                    return error.sqlstate
            with ThreadPoolExecutor(max_workers=2) as pool:
                first = pool.submit(finish, a, 2)
                second = pool.submit(finish, b, 1)
                outcomes = sorted((first.result(), second.result()))
            assert outcomes == ["40P01", "ok"], outcomes
            assert admin.execute("SELECT hits FROM counters ORDER BY id").fetchall() == [(1,), (1,)]
            print("FAILURE reproduced: deadlock 40P01; victim transaction rolled back")

            def ordered_transaction():
                with connect() as conn:
                    try:
                        conn.execute("BEGIN")
                        conn.execute("SELECT id FROM counters ORDER BY id FOR UPDATE").fetchall()
                        conn.execute("UPDATE counters SET hits=hits+1")
                        conn.execute("COMMIT")
                    except Exception:
                        conn.execute("ROLLBACK")
                        raise
            transaction_retry(ordered_transaction)
            assert admin.execute("SELECT hits FROM counters ORDER BY id").fetchall() == [(2,), (2,)]
            admin.execute("UPDATE counters SET hits=0")
            with ThreadPoolExecutor(max_workers=2) as pool:
                one = pool.submit(transaction_retry, ordered_transaction)
                two = pool.submit(transaction_retry, ordered_transaction)
                one.result()
                two.result()
            assert admin.execute("SELECT hits FROM counters ORDER BY id").fetchall() == [(2,), (2,)]
            print("FIX passed: ordered locks plus bounded whole-transaction retry")


if __name__ == "__main__":
    main()

\set ON_ERROR_STOP on
\pset pager off
SET search_path = interview_lab;
\echo 'Small uniform table: 200 rows. Owner 2 has 2 rows.'
TRUNCATE bookmarks;
DROP INDEX IF EXISTS bookmarks_owner_time;
INSERT INTO bookmarks
SELECT g, 1 + (g % 100), '2026-01-01 UTC'::timestamptz + g * interval '1 second', repeat('x',200), ''
FROM generate_series(1,200) g;
ANALYZE bookmarks;
EXPLAIN (ANALYZE, BUFFERS) SELECT id,title FROM bookmarks WHERE owner_id=2 ORDER BY created_at DESC,id DESC LIMIT 20;
CREATE INDEX bookmarks_owner_time ON bookmarks(owner_id,created_at DESC,id DESC);
ANALYZE bookmarks;
EXPLAIN (ANALYZE, BUFFERS) SELECT id,title FROM bookmarks WHERE owner_id=2 ORDER BY created_at DESC,id DESC LIMIT 20;
\echo 'Large uniform table: 100000 rows. Owner 2 has 1000 rows.'
TRUNCATE bookmarks;
INSERT INTO bookmarks
SELECT g, 1 + (g % 100), '2026-01-01 UTC'::timestamptz + g * interval '1 second', repeat('x',200), ''
FROM generate_series(1,100000) g;
ANALYZE bookmarks;
EXPLAIN (ANALYZE, BUFFERS) SELECT id,title FROM bookmarks WHERE owner_id=2 ORDER BY created_at DESC,id DESC LIMIT 20;
\echo 'Remove matching index for the same large query, then restore it.'
DROP INDEX bookmarks_owner_time;
EXPLAIN (ANALYZE, BUFFERS) SELECT id,title FROM bookmarks WHERE owner_id=2 ORDER BY created_at DESC,id DESC LIMIT 20;
CREATE INDEX bookmarks_owner_time ON bookmarks(owner_id,created_at DESC,id DESC);
\echo 'Skew: owner 1 gets first 90000 rows; owner 2 gets 100 rows.'
UPDATE bookmarks SET owner_id=1 WHERE id<=90000;
ANALYZE bookmarks;
SELECT owner_id,count(*) FROM bookmarks WHERE owner_id IN (1,2) GROUP BY owner_id ORDER BY owner_id;
EXPLAIN (ANALYZE, BUFFERS) SELECT id,title FROM bookmarks WHERE owner_id=2;
EXPLAIN (ANALYZE, BUFFERS) SELECT id,title FROM bookmarks WHERE owner_id=1;
EXPLAIN (ANALYZE, BUFFERS) SELECT id,title FROM bookmarks WHERE owner_id=1 ORDER BY created_at DESC,id DESC LIMIT 20;
\echo 'Keyset page: owner 2, cursor at row 99901; expect at most 20 strictly earlier pairs.'
EXPLAIN (ANALYZE, BUFFERS) SELECT id,title FROM bookmarks
WHERE owner_id=2 AND (created_at,id)<('2026-01-02 03:45:01 UTC'::timestamptz,99901)
ORDER BY created_at DESC,id DESC LIMIT 20;
\echo 'Heap location is independent from key order; HOT eligibility is conditional.'
SELECT id,ctid FROM bookmarks ORDER BY id LIMIT 5;
UPDATE bookmarks SET note='non-indexed update' WHERE id BETWEEN 1 AND 100;
SELECT relname,n_tup_upd,n_tup_hot_upd FROM pg_stat_user_tables WHERE schemaname='interview_lab' AND relname='bookmarks';
-- Statistics can be delayed. This is observation, not a guaranteed HOT count.

import json
from pathlib import Path
import sys
import tempfile
from threading import Event, Thread
import unittest
import urllib.error
import urllib.request

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from server import serve, connect


class APITests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.db = str(Path(self.temp.name) / "bookmarks.db")
        self.server = serve(self.db, 0)
        self.thread = Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        self.url = f"http://127.0.0.1:{self.server.server_port}"

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join()
        self.temp.cleanup()

    def request(self, path, body=None, key="test-key", token="alice-local-token"):
        headers = {"Authorization": "Bearer " + token, "Idempotency-Key": key, "Content-Type": "application/json"}
        request = urllib.request.Request(self.url + path, data=None if body is None else json.dumps(body).encode(),
                                         headers=headers, method="GET" if body is None else "PATCH")
        try:
            response = urllib.request.urlopen(request, timeout=3)
        except urllib.error.HTTPError as error:
            response = error
        with response:
            return response.status, json.load(response)

    def test_owner_enforced_for_reads_writes_and_missing_auth(self):
        self.assertEqual(self.request("/api/bookmarks/secret")[0], 404)
        self.assertEqual(self.request("/api/bookmarks/secret", {"title":"stolen", "expectedVersion":1})[0], 404)
        self.assertEqual(self.request("/api/bookmarks", token="invalid")[0], 401)
        code, page = self.request("/api/bookmarks", token="bob-local-token")
        self.assertEqual([row["id"] for row in page["items"]], ["secret"])

    def test_tied_cursor_complete_without_repeats(self):
        seen, cursor = [], None
        for _ in range(4):
            code, page = self.request("/api/bookmarks?limit=1" + ("&cursor=" + cursor if cursor else ""))
            self.assertEqual(code, 200)
            seen += [item["id"] for item in page["items"]]
            cursor = page["nextCursor"]
            if cursor is None:
                break
        self.assertEqual(seen, ["b", "a", "c"])

    def test_conditional_update_duplicate_response_and_durable_store(self):
        body = {"title": "A", "expectedVersion": 1}
        code, first = self.request("/api/bookmarks/b", body)
        self.assertEqual((code, first["version"]), (200, 2))
        self.assertEqual(self.request("/api/bookmarks/b", body), (200, first))
        code, conflict = self.request("/api/bookmarks/b", {"title":"B", "expectedVersion":1}, key="second")
        self.assertEqual((code, conflict["current"]["title"]), (409, "A"))
        self.assertEqual(self.request("/api/bookmarks/b", {"title":"other", "expectedVersion":2})[0], 409)
        with connect(self.db) as db:
            self.assertEqual(tuple(db.execute("SELECT title,version FROM bookmarks WHERE id='b'").fetchone()), ("A", 2))
            self.assertEqual(db.execute("SELECT count(*) FROM mutations").fetchone()[0], 1)

    def test_concurrent_writers_only_one_wins(self):
        from threading import Barrier
        gate, results = Barrier(3), []
        def writer(name):
            gate.wait()
            results.append(self.request("/api/bookmarks/b", {"title":name, "expectedVersion":1}, key=name)[0])
        threads = [Thread(target=writer, args=(name,)) for name in ("one", "two")]
        for thread in threads:
            thread.start()
        gate.wait()
        for thread in threads:
            thread.join(3)
        self.assertEqual(sorted(results), [200, 409])

    def test_runtime_validation_and_empty_search(self):
        for body in [{"title":"", "expectedVersion":1}, {"title":"A", "expectedVersion":True}, {"title":"A"},
                     {"title":"A" + " " * 200, "expectedVersion":1}, {"title":"A", "expectedVersion":10**100}]:
            self.assertEqual(self.request("/api/bookmarks/b", body)[0], 400)
        for query in ["limit=0", "limit=51", "cursor=bad", "cursor=W10="]:
            self.assertEqual(self.request("/api/bookmarks?" + query)[0], 400)
        self.assertEqual(self.request("/api/bookmarks?q=notpresent")[1]["items"], [])

    def test_unicode_validation_and_durable_idempotence(self):
        for title in ("\0abc", "a\0bc", "\ud800", "line\nbreak", "\u0085"):
            code, body = self.request("/api/bookmarks/b", {"title": title, "expectedVersion": 1})
            self.assertEqual(code, 400)
            self.assertIn("error", body)
        with connect(self.db) as db:
            self.assertEqual(tuple(db.execute("SELECT title,version FROM bookmarks WHERE id='b'").fetchone()), ("Beta", 1))
            self.assertEqual(db.execute("SELECT count(*) FROM mutations").fetchone()[0], 0)
        body = {"title": "Café 👩‍💻", "expectedVersion": 1}
        first = self.request("/api/bookmarks/b", body)
        self.assertEqual((first[0], first[1]["title"], first[1]["version"]), (200, body["title"], 2))
        self.assertEqual(self.request("/api/bookmarks/b", body), first)

    def test_slow_conflict_response_releases_write_transaction(self):
        self.request("/api/bookmarks/b", {"title": "first", "expectedVersion": 1}, key="first")
        entered, release, writer_done = Event(), Event(), Event()
        handler = self.server.RequestHandlerClass
        original = handler.reply
        results, errors = [], []
        def gated_reply(request, status, body):
            if status == 409:
                entered.set()
                if not release.wait(3):
                    raise TimeoutError("conflict response gate")
            return original(request, status, body)
        handler.reply = gated_reply
        def request_in_thread(body, key, done=None):
            try:
                results.append((key, self.request("/api/bookmarks/b", body, key=key)[0]))
            except Exception as exc:
                errors.append(exc)
            finally:
                if done:
                    done.set()
        conflict = Thread(target=request_in_thread, args=({"title": "stale", "expectedVersion": 1}, "stale"))
        writer = Thread(target=request_in_thread, args=({"title": "next", "expectedVersion": 2}, "next", writer_done))
        conflict.start()
        try:
            self.assertTrue(entered.wait(2))
            writer.start()
            self.assertTrue(writer_done.wait(2), "slow response retained a write transaction")
        finally:
            release.set()
            conflict.join(3)
            if writer.ident is not None:
                writer.join(3)
            handler.reply = original
        self.assertEqual(errors, [])
        self.assertEqual(sorted(results), [("next", 200), ("stale", 409)])


if __name__ == "__main__":
    unittest.main()

// Run: node tests/browser.mjs. Uses installed playwright or the Work runtime copy.
import assert from 'node:assert/strict';
import {createRequire} from 'node:module';
import {spawn} from 'node:child_process';
import {mkdtemp, mkdir, rm} from 'node:fs/promises';
import {tmpdir} from 'node:os';
import path from 'node:path';
import {fileURLToPath} from 'node:url';
const require = createRequire(import.meta.url);
const {chromium} = require(require.resolve('playwright', {paths: [process.cwd(), process.env.CODEX_PRIMARY_RUNTIME_NODE_MODULES || process.cwd()]}));
const root = fileURLToPath(new URL('../', import.meta.url));
const temp = await mkdtemp(path.join(tmpdir(), 'bookmark-browser-'));
const server = spawn(process.env.CODEX_PRIMARY_RUNTIME_PYTHON || 'python3', ['server.py', '--port', '0', '--db', path.join(temp, 'data.db')], {cwd: root});
const base = await new Promise((resolve, reject) => {
  const timer = setTimeout(() => reject(new Error('server startup timeout')), 5000);
  server.stdout.on('data', data => {const match = String(data).match(/http:\/\/127\.0\.0\.1:\d+/); if (match) {clearTimeout(timer); resolve(match[0]);}});
  server.on('error', reject);
});
const launch = {headless:true};
if (process.env.BOOKMARK_CHROMIUM_PATH) {
  launch.executablePath = process.env.BOOKMARK_CHROMIUM_PATH;
  launch.args = ['--no-sandbox', '--disable-dev-shm-usage', '--disable-gpu', '--single-process', '--no-zygote'];
}
const browser = await chromium.launch(launch);
const context = await browser.newContext({viewport:{width:1100,height:900}});
const auth = {Authorization:'Bearer alice-local-token'};
const json = async (url, options = {}) => {
  const response = await fetch(base + url, {headers: auth, ...options});
  return {status: response.status, body: await response.json()};
};
async function setTitle(title) {
  const {body: old} = await json('/api/bookmarks/b');
  return json('/api/bookmarks/b', {method:'PATCH', headers:{...auth,'Content-Type':'application/json','Idempotency-Key':crypto.randomUUID()}, body:JSON.stringify({title,expectedVersion:old.version})});
}
function deferred() {let resolve; const promise = new Promise(r => {resolve = r;}); return {promise, resolve};}
async function newPage() {
  await setTitle('Beta');
  const page = await context.newPage();
  page.on('pageerror', error => console.error('BROWSER ERROR', error.message));
  await page.goto(base);
  await page.getByRole('button', {name:'Edit Beta', exact:true}).waitFor();
  return page;
}
async function textIs(page, selector, expected) {
  await page.waitForFunction(({selector, expected}) => document.querySelector(selector)?.textContent.includes(expected), {selector,expected});
}
async function open(page) {await page.getByRole('button',{name:'Edit Beta',exact:true}).click();}
let checks = 0;
async function check(name, fn) {await fn(); checks++; console.log(`PASS ${name}`);}
try {
  await check('save A preserves newer draft B', async () => {
    const page = await newPage(); await open(page);
    const caught = deferred(), release = deferred();
    await page.route('**/api/bookmarks/b', async route => {
      if(route.request().method() !== 'PATCH') return route.continue();
      const response = await route.fetch(); caught.resolve(); await release.promise; await route.fulfill({response});
    });
    await page.getByLabel('Title',{exact:true}).fill('Edit A');
    await page.getByRole('button',{name:'Save title',exact:true}).click(); await caught.promise;
    await page.getByLabel('Title',{exact:true}).fill('Edit B'); release.resolve();
    await textIs(page,'#edit-status','newer draft is unsaved');
    assert.equal(await page.getByLabel('Title',{exact:true}).inputValue(),'Edit B');
    assert.equal((await json('/api/bookmarks/b')).body.title,'Edit A');
    await mkdir(path.join(root,'tests/artifacts'),{recursive:true});
    await page.screenshot({path:path.join(root,'tests/artifacts/draft-preserved.png'),fullPage:true});
    await page.close();
  });
  await check('409 retains B; keyboard conflict recovery has focus and live status', async () => {
    const page = await newPage(); await open(page);
    const caught = deferred(), release = deferred();
    await page.route('**/api/bookmarks/b', async route => {
      if(route.request().method() !== 'PATCH') return route.continue();
      caught.resolve(); await release.promise; await route.continue();
    });
    await page.getByLabel('Title',{exact:true}).fill('Edit A');
    await page.getByLabel('Title',{exact:true}).press('Enter'); await caught.promise;
    await page.getByLabel('Title',{exact:true}).fill('Edit B');
    await setTitle('External edit'); release.resolve();
    await textIs(page,'#edit-status','Conflict.');
    assert.equal(await page.getByLabel('Title',{exact:true}).inputValue(),'Edit B');
    assert.equal(await page.locator('#resolve').evaluate(el=>el===document.activeElement),true);
    assert.equal(await page.locator('#edit-status').getAttribute('role'),'status');
    await page.keyboard.press('Enter');
    assert.equal(await page.locator('#title').evaluate(el=>el===document.activeElement),true);
    await page.unroute('**/api/bookmarks/b');
    await page.keyboard.press('Enter'); await textIs(page,'#edit-status','Saved.');
    assert.equal((await json('/api/bookmarks/b')).body.title,'Edit B');
    await page.close();
  });
  await check('old refetch cannot replace confirmed version or unsaved draft', async () => {
    const page = await newPage(); await open(page);
    const caught = deferred(), release = deferred();
    await page.route('**/api/bookmarks/b', async route => {
      if(route.request().method() !== 'GET') return route.continue();
      const response = await route.fetch(); caught.resolve(); await release.promise; await route.fulfill({response});
    });
    await page.getByRole('button',{name:'Refresh server copy'}).click(); await caught.promise;
    await page.getByLabel('Title',{exact:true}).fill('New confirmed');
    await page.getByRole('button',{name:'Save title',exact:true}).click(); await textIs(page,'#edit-status','Saved.');
    await page.getByLabel('Title',{exact:true}).fill('Unsaved next'); release.resolve();
    await textIs(page,'#edit-status','Server copy unchanged');
    assert.equal(await page.getByLabel('Title',{exact:true}).inputValue(),'Unsaved next');
    assert.match(await page.locator('#confirmed').textContent(),/New confirmed/);
    await page.close();
  });
  await check('lost acknowledgement retries same key and duplicate response increments once', async () => {
    const page = await newPage(); await open(page);
    const before = (await json('/api/bookmarks/b')).body.version;
    let first = true; const keys = [];
    await page.route('**/api/bookmarks/b', async route => {
      if(route.request().method() !== 'PATCH') return route.continue();
      keys.push(route.request().headers()['idempotency-key']);
      const response = await route.fetch();
      if(first) {first=false; return route.abort('failed');}
      await route.fulfill({response});
    });
    await page.getByLabel('Title',{exact:true}).fill('Edit A');
    await page.getByRole('button',{name:'Save title',exact:true}).click();
    await textIs(page,'#edit-status','outcome unavailable');
    await page.getByLabel('Title',{exact:true}).fill('Edit B');
    await page.getByRole('button',{name:'Retry save',exact:true}).click();
    await textIs(page,'#edit-status','newer draft is unsaved');
    assert.equal(keys.length,2); assert.equal(keys[0],keys[1]);
    assert.equal((await json('/api/bookmarks/b')).body.version,before+1);
    assert.equal(await page.getByLabel('Title',{exact:true}).inputValue(),'Edit B');
    await page.close();
  });
  await check('close disposes editor even when transport ignores abort; focus returns', async () => {
    const page = await newPage(); await open(page);
    await page.evaluate(()=>{const original=window.fetch;window.fetch=(url,options)=>original(url,{...options,signal:undefined});});
    const caught = deferred(), release = deferred(), delivered = deferred();
    await page.route('**/api/bookmarks/b', async route => {
      if(route.request().method() !== 'PATCH') return route.continue();
      const response=await route.fetch();caught.resolve();await release.promise;await route.fulfill({response});delivered.resolve();
    });
    await page.getByLabel('Title',{exact:true}).fill('Late edit');
    await page.getByRole('button',{name:'Save title',exact:true}).click();await caught.promise;
    await page.getByRole('button',{name:'Close editor'}).click();
    assert.equal(await page.locator('#editor').isHidden(),true);
    assert.equal(await page.getByRole('button',{name:'Edit Beta',exact:true}).evaluate(el=>el===document.activeElement),true);
    release.resolve();await delivered.promise;
    await page.getByRole('button',{name:'Edit Alpha',exact:true}).click();
    assert.equal(await page.getByLabel('Title',{exact:true}).inputValue(),'Alpha');
    await page.close();
  });
  await check('search race ignores older response; empty, error and keyboard retry work', async () => {
    const page = await newPage();
    await page.evaluate(()=>{const original=window.fetch;window.fetch=(url,options)=>original(url,{...options,signal:undefined});});
    const caught=deferred(), release=deferred(), delivered=deferred();
    await page.route('**/api/bookmarks?*', async route=>{
      if(new URL(route.request().url()).searchParams.get('q')!=='Alpha')return route.continue();
      const response=await route.fetch();caught.resolve();await release.promise;await route.fulfill({response});delivered.resolve();
    });
    await page.getByLabel('Search bookmarks').fill('Alpha');await page.getByLabel('Search bookmarks').press('Enter');await caught.promise;
    await page.getByLabel('Search bookmarks').fill('Beta');await page.getByLabel('Search bookmarks').press('Enter');
    await textIs(page,'#list-status','1 bookmarks');release.resolve();await delivered.promise;
    assert.equal(await page.getByRole('button',{name:'Edit Beta',exact:true}).count(),1);
    assert.equal(await page.getByRole('button',{name:'Edit Alpha',exact:true}).count(),0);
    await page.unroute('**/api/bookmarks?*');
    await page.getByLabel('Search bookmarks').fill('none-match');await page.getByLabel('Search bookmarks').press('Enter');
    await textIs(page,'#list-status','No bookmarks');
    await page.route('**/api/bookmarks?*',route=>route.fulfill({status:503,contentType:'application/json',body:'{"error":"injected outage"}'}));
    await page.getByLabel('Search bookmarks').fill('');await page.getByLabel('Search bookmarks').press('Enter');
    await textIs(page,'#list-status','Search failed');await page.unroute('**/api/bookmarks?*');
    await page.getByRole('button',{name:'Retry search'}).focus();await page.keyboard.press('Enter');await textIs(page,'#list-status','2 bookmarks');
    await page.getByRole('button',{name:'Load more'}).focus();await page.keyboard.press('Enter');await textIs(page,'#list-status','3 bookmarks');
    assert.equal(await page.locator('#results li').count(),3);
    assert.equal(await page.getByRole('button',{name:'Edit Bob private',exact:true}).count(),0);
    await page.close();
  });
  await check('failed new search cannot reuse an old query cursor', async () => {
    const page = await newPage();
    assert.equal(await page.locator('#more').isVisible(),true);
    await page.route('**/api/bookmarks?*',route=>route.fulfill({status:503,contentType:'application/json',body:'{"error":"injected outage"}'}));
    await page.getByLabel('Search bookmarks').fill('Gamma');
    await page.getByLabel('Search bookmarks').press('Enter');
    await textIs(page,'#list-status','Search failed');
    assert.equal(await page.locator('#more').isHidden(),true);
    assert.equal(await page.locator('#results li').count(),0);
    await page.unroute('**/api/bookmarks?*');
    const next = page.waitForRequest(request=>request.url().includes('/api/bookmarks?'));
    await page.getByRole('button',{name:'Retry search'}).click();
    const request = await next;
    assert.equal(new URL(request.url()).searchParams.get('cursor'),null);
    await textIs(page,'#list-status','1 bookmarks');
    assert.equal(await page.getByRole('button',{name:'Edit Gamma',exact:true}).count(),1);
    assert.equal(await page.getByRole('button',{name:'Edit Beta',exact:true}).count(),0);
    await page.close();
  });
  console.log(`${checks} browser scenarios passed against real HTTP + SQLite.`);
} finally {
  await browser.close(); server.kill('SIGTERM');
  await new Promise(resolve=>server.once('exit',resolve));
  await rm(temp,{recursive:true,force:true});
}

import asyncio
import unittest
from reference import CacheAside, Clock, LinkDelivery, OriginBudget, Unavailable, sticky_read, strict_read


class CacheTests(unittest.IsolatedAsyncioTestCase):
    async def test_baseline_200_misses_and_fixed_one_flight(self):
        clock = Clock()
        calls = 0
        ready = asyncio.Event()
        async def load(key):
            nonlocal calls
            calls += 1
            await ready.wait()
            return 8
        baseline = [asyncio.create_task(load("hot")) for _ in range(200)]
        await asyncio.sleep(0)
        self.assertEqual(calls, 200)
        ready.set()
        await asyncio.gather(*baseline)
        calls = 0
        cache = CacheAside(clock, OriginBudget(clock))
        self.assertEqual(await asyncio.gather(*(cache.get("hot", load) for _ in range(200))), [8] * 200)
        self.assertEqual(calls, 1)
        clock.advance(6)
        await asyncio.gather(*(cache.get("hot", load) for _ in range(200)))
        self.assertEqual(calls, 2)
        self.assertFalse(cache.flights)

    async def test_ten_instances_and_many_keys(self):
        clock = Clock()
        budget = OriginBudget(clock, per_second=100, concurrent=100)
        async def load(key):
            await asyncio.sleep(0)
            return key
        caches = [CacheAside(clock, budget) for _ in range(10)]
        await asyncio.gather(*(caches[i % 10].get("hot", load) for i in range(200)))
        self.assertEqual(budget.total, 10)
        await asyncio.gather(*(caches[0].get(str(i), load) for i in range(20)))
        self.assertEqual(budget.total, 30)  # Coalescing distinct keys changes semantics.
        clock.advance(6)
        await asyncio.gather(*(caches[0].get(str(i), load) for i in range(20)))
        self.assertEqual(budget.total, 50)  # Twenty keys expire together: twenty loads.
        for i in range(20):
            caches[0].values[str(i)] = (clock.now + 5 + i / 10, str(i))
        clock.advance(5)
        await asyncio.gather(*(caches[0].get(str(i), load) for i in range(20)))
        self.assertEqual(budget.total, 51)  # Staggered expiry: one key expired so far.

    async def test_loader_exception_releases_every_waiter_and_retry(self):
        clock = Clock()
        cache = CacheAside(clock, OriginBudget(clock))
        calls = 0
        async def broken(key):
            nonlocal calls
            calls += 1
            await asyncio.sleep(0)
            raise ValueError("origin failed")
        results = await asyncio.gather(*(cache.get("hot", broken) for _ in range(30)), return_exceptions=True)
        self.assertTrue(all(isinstance(value, ValueError) for value in results))
        self.assertEqual(calls, 1)
        self.assertEqual(cache.budget.active, 0)
        self.assertFalse(cache.flights)
        async def recovered(key):
            return 9
        self.assertEqual(await cache.get("hot", recovered), 9)

    async def test_waiter_and_loader_cancellation_have_different_scope(self):
        clock = Clock()
        cache = CacheAside(clock, OriginBudget(clock))
        entered, release = asyncio.Event(), asyncio.Event()
        async def load(key):
            entered.set()
            await release.wait()
            return 8
        first = asyncio.create_task(cache.get("hot", load))
        await entered.wait()
        second = asyncio.create_task(cache.get("hot", load))
        first.cancel()
        with self.assertRaises(asyncio.CancelledError):
            await first
        release.set()
        self.assertEqual(await second, 8)
        clock.advance(6)
        entered.clear()
        release.clear()
        waiters = [asyncio.create_task(cache.get("hot", load)) for _ in range(2)]
        await entered.wait()
        cache.flights["hot"].cancel()
        results = await asyncio.gather(*waiters, return_exceptions=True)
        self.assertTrue(all(isinstance(value, asyncio.CancelledError) for value in results))
        self.assertFalse(cache.flights)
        self.assertEqual(cache.budget.active, 0)
        release.set()
        self.assertEqual(await cache.get("hot", load), 8)

    async def test_outage_caps_concurrency_and_rate_across_instances(self):
        clock = Clock()
        budget = OriginBudget(clock, per_second=100, concurrent=10)
        caches = [CacheAside(clock, budget) for _ in range(10)]
        for cache in caches:
            cache.cache_available = False
        async def load(key):
            await asyncio.sleep(0)
            return key
        results = []
        for wave in range(10):
            results += await asyncio.gather(*(caches[i % 10].get(f"{wave}:{i}", load) for i in range(100)), return_exceptions=True)
        self.assertEqual(budget.total, 100)
        self.assertEqual(budget.peak, 10)
        self.assertEqual(sum(isinstance(r, Unavailable) for r in results), 900)
        with self.assertRaises(Unavailable):
            await caches[0].get("extra", load)
        clock.advance(1)
        self.assertEqual(await caches[0].get("next-second", load), "next-second")


class ConsistencyTests(unittest.TestCase):
    def test_lag_outlives_pin_and_watermark_fallback(self):
        self.assertEqual(sticky_read(8, 7, now=6, pin_until=5), 7)
        self.assertEqual(strict_read(8, 7, minimum_version=8), 8)
        self.assertEqual(strict_read(None, 8, minimum_version=8), 8)
        with self.assertRaises(Unavailable):
            strict_read(None, 7, minimum_version=8)

    def test_warm_same_token_and_cold_revocation(self):
        delivery = LinkDelivery(Clock())
        self.assertEqual(delivery.get("A"), (200, "Ana 09:00"))
        del delivery.tokens["A"]
        self.assertEqual(delivery.get("A"), (403, None))
        delivery.cache.clear()
        self.assertEqual(delivery.get("A"), (403, None))
        self.assertEqual(delivery.origin_calls, 1)

    def test_token_tenant_isolation_and_origin_outage(self):
        delivery = LinkDelivery(Clock())
        self.assertNotEqual(delivery.get("A")[1], delivery.get("B")[1])
        delivery.origin_available = False
        self.assertEqual(delivery.get("A")[0], 200)
        del delivery.tokens["A"]
        self.assertEqual(delivery.get("A")[0], 403)
        delivery.cache.clear()
        self.assertEqual(delivery.get("B")[0], 503)
        delivery.auth_available = False
        self.assertEqual(delivery.get("B")[0], 503)

    def test_bounded_policy_expires_without_sliding_or_fail_open(self):
        clock = Clock()
        delivery = LinkDelivery(clock, policy="bounded", auth_ttl=5)
        self.assertEqual(delivery.get("A")[0], 200)
        del delivery.tokens["A"]
        clock.advance(4.999)
        self.assertEqual(delivery.get("A")[0], 200)
        clock.advance(.001)
        self.assertEqual(delivery.get("A")[0], 403)
        delivery.auth_available = False
        self.assertEqual(delivery.get("A")[0], 503)


if __name__ == "__main__":
    unittest.main()

import unittest
from reference import ChangeSource, Conflict, FencedResult, HistoryExpired, IntentStore, Provider, Replica, Row, RoutingRegistry, apply, rollback_ready


class RecoveryTests(unittest.TestCase):
    def test_external_effect_before_or_after_marker_has_a_crash_gap(self):
        effects, markers = [], set()
        effects.append("email")  # Send succeeds; crash before marker.
        if "a" not in markers:
            effects.append("email")  # Naive replay repeats the external effect.
            markers.add("a")
        self.assertEqual(len(effects), 2)
        effects, markers = [], {"a"}  # Marker commits; crash before send.
        if "a" not in markers:
            effects.append("email")
        self.assertEqual(len(effects), 0)

    def test_paused_owner_can_overwrite_without_fence_but_is_rejected_with_it(self):
        naive = {"title": "new"}
        naive["title"] = "old"  # A resumes after B.
        self.assertEqual(naive["title"], "old")
        store = FencedResult()
        a = store.claim(0)
        b = store.claim(6)
        store.complete(b, 7, "new")
        with self.assertRaises(Conflict):
            store.complete(a, 8, "old")
        self.assertEqual(store.result, "new")
        with self.assertRaises(Conflict):
            store.complete(b, 11, "expired")

    def test_atomic_intent_outbox_and_changed_payload_quarantine(self):
        store = IntentStore()
        self.addCleanup(store.close)
        with self.assertRaises(RuntimeError):
            store.accept("a", {"title": "hello"}, crash_before_commit=True)
        self.assertEqual(store.db.execute("SELECT count(*) FROM intent").fetchone()[0], 0)
        self.assertEqual(store.db.execute("SELECT count(*) FROM outbox").fetchone()[0], 0)
        self.assertEqual(store.accept("a", {"title": "hello"}), "accepted")
        self.assertEqual(store.accept("a", {"title": "hello"}), "duplicate")
        with self.assertRaises(Conflict):
            store.accept("a", {"title": "different"})
        self.assertEqual(store.db.execute("SELECT count(*) FROM outbox").fetchone()[0], 1)

    def test_external_success_with_lost_response_and_reconciliation(self):
        store, provider = IntentStore(), Provider()
        self.addCleanup(store.close)
        store.accept("a", {"email": "hello"})
        provider.lose_next_response = True
        self.assertEqual(store.dispatch("a", provider), "uncertain")
        self.assertEqual(provider.effects, 1)
        provider.available = False
        self.assertEqual(store.dispatch("a", provider), "uncertain")
        provider.available = True
        self.assertEqual(store.dispatch("a", provider), "done")
        self.assertEqual(provider.effects, 1)
        with self.assertRaises(Conflict):
            provider.send("a", {"email": "changed"})

    def test_external_crash_after_receipt_before_local_commit(self):
        store, provider = IntentStore(), Provider()
        self.addCleanup(store.close)
        store.accept("a", [1, 2])
        with self.assertRaises(RuntimeError):
            store.dispatch("a", provider, crash_after_send=True)
        self.assertEqual(store.dispatch("a", provider), "done")
        self.assertEqual(provider.effects, 1)

    def test_expired_provider_dedup_window_permits_another_effect(self):
        provider = Provider()
        provider.send("a", [1])
        provider.receipts.clear()  # Explicit counterexample: retention contract expired.
        provider.send("a", [1])
        self.assertEqual(provider.effects, 2)


class MigrationTests(unittest.TestCase):
    def test_partial_write_stale_backfill_delete_and_repair(self):
        source, target = ChangeSource(), Replica()
        source.write("a", "v1")
        source.write("deleted", "old")
        baseline, position = source.snapshot()
        source.write("a", "v2")  # Independent target write fails here.
        source.write("deleted", deleted=True)
        self.assertEqual(set(target.differences(source)), {"a", "deleted"})
        target.checkpoint = position
        target.resume(source)  # Live mutations arrive before slow backfill.
        for key, row in baseline.items():
            apply(target.rows, key, row)
        self.assertEqual(target.rows["a"], Row(2, "v2"))
        self.assertTrue(target.rows["deleted"].deleted)
        self.assertFalse(target.differences(source))
        with self.assertRaises(Conflict):
            apply(target.rows, "a", Row(2, "different"))

    def test_crash_between_apply_and_checkpoint_replays_idempotently(self):
        source, target = ChangeSource(), Replica()
        source.write("a", "v1")
        with self.assertRaises(RuntimeError):
            target.resume(source, crash_after_apply=True)
        self.assertEqual(target.checkpoint, 0)
        target.resume(source)
        self.assertEqual(target.checkpoint, 1)
        self.assertFalse(target.differences(source))

    def test_history_expiry_demands_resnapshot_then_new_live_replay(self):
        source, target = ChangeSource(), Replica()
        source.write("a", "old")
        source.write("a", deleted=True)
        source.expire_through(2)
        with self.assertRaises(HistoryExpired):
            target.resume(source)
        target.resnapshot(source)
        source.write("b", "new")
        target.resume(source)
        self.assertFalse(target.differences(source))
        self.assertTrue(target.rows["a"].deleted)

    def test_rebalance_blocks_stale_route_and_incomplete_target(self):
        source, target, registry = ChangeSource(), Replica(), RoutingRegistry()
        source.write("tenant-a", "v1")
        with self.assertRaises(Conflict):
            registry.cutover(target, source)
        target.resnapshot(source)
        source.write("tenant-a", "v2")
        with self.assertRaises(Conflict):
            registry.cutover(target, source)
        target.resume(source)
        registry.cutover(target, source)
        with self.assertRaises(Conflict):
            registry.check_write("old", 1)
        registry.check_write("new", 2)

    def test_rollback_after_new_only_write_needs_reverse_repair(self):
        old = {"a": Row(1, "old")}
        new = dict(old)
        apply(new, "a", Row(2, "new"))
        apply(new, "b", Row(1, None, True))
        self.assertFalse(rollback_ready(old, new))
        for key, row in new.items():
            apply(old, key, row)
        self.assertTrue(rollback_ready(old, new))
        self.assertEqual(old["a"].value, "new")

    def test_dns_cached_route_and_async_region_data_loss_are_distinct(self):
        resolver = {"answer": "old", "until": 60}
        registry_answer = "new"
        self.assertEqual(resolver["answer"] if 10 < resolver["until"] else registry_answer, "old")
        existing_connection = "old"  # A new DNS answer doesn't move this socket.
        self.assertEqual(existing_connection, "old")
        primary, replica = {"a": Row(9, "acknowledged")}, {"a": Row(8, "older")}
        lost = {key for key in primary if primary[key] != replica.get(key)}
        self.assertEqual(lost, {"a"})
        remote_available = False
        sync_ack = remote_available  # Fail closed if policy requires remote commit.
        self.assertFalse(sync_ack)


if __name__ == "__main__":
    unittest.main()

import unittest
import csv
from pathlib import Path
from fractions import Fraction as F
from model import sli, burn, alarm, latched, leaf_attempts, retry_allowed, admit, capacity, drain


class ReliabilityTests(unittest.TestCase):
    def test_incident_csv_windows_and_observed_queue(self):
        with (Path(__file__).parent/'incident.csv').open() as f:
            rows=[{k:int(v) for k,v in row.items()} for row in csv.DictReader(f)]
        def window(start,end):
            return [(r['good_reads'],r['eligible_reads']) for r in rows[start:end+1]]
        self.assertEqual(burn(window(3,4)),45)
        self.assertEqual(burn(window(0,4)),22)
        self.assertTrue(alarm(burn(window(3,4)),burn(window(0,4))))
        self.assertEqual(burn(window(5,6)),0)
        self.assertEqual(burn(window(2,6)),22)
        self.assertFalse(alarm(burn(window(5,6)),burn(window(2,6))))
        for previous,current in zip(rows,rows[1:]):
            growth=(current['offered_attempts_per_s']-current['completed_attempts_per_s'])*60
            self.assertEqual(current['queue_items']-previous['queue_items'],growth)
        self.assertEqual(drain(rows[5]['queue_items'],80,100),1020)

    def test_variable_traffic_budget_not_minutes(self):
        rows = [(990000, 990000), (0, 10000)]
        self.assertEqual(sli(rows), F(99,100))
        self.assertEqual(burn(rows), 10)
        self.assertNotEqual(sli(rows), F(1,2))  # averaging interval ratios is wrong
        self.assertEqual(F(30*24*60,1000), F(216,5))  # 43.2 time-based minutes only

    def test_no_eligible_requests_and_invalid_counters(self):
        self.assertIsNone(sli([(0,0)]))
        self.assertIsNone(burn([]))
        self.assertIsNone(alarm(None,20))
        self.assertFalse(alarm(None,0))
        with self.assertRaises(ValueError):
            sli([(2,1)])

    def test_boolean_recovery_both_directions_and_latch(self):
        self.assertTrue(alarm(20,20))
        self.assertFalse(alarm(0,20))
        self.assertFalse(alarm(20,0))
        self.assertFalse(alarm(0,0))
        self.assertTrue(latched(True,False,True))
        self.assertTrue(latched(True,True,False))
        self.assertFalse(latched(True,False,False))
        self.assertFalse(latched(False,False,True))

    def test_attempt_units_and_layer_scope(self):
        self.assertEqual(leaf_attempts([3,3,3]),27)
        self.assertEqual(leaf_attempts([3,3,3],counts='retries'),64)
        self.assertEqual(leaf_attempts([1,3,1]),3)
        self.assertEqual(100+20*20,500)  # per-process is not global retry budget

    def test_priority_cannot_create_capacity(self):
        a=admit(120,30,100)
        self.assertEqual((a.critical,a.bulk,a.critical_refused,a.bulk_refused),(100,0,20,30))
        a=admit(80,40,100)
        self.assertEqual((a.critical,a.bulk,a.critical_refused,a.bulk_refused),(80,20,0,20))

    def test_retry_policy_has_countermodels(self):
        p=dict(safe=True,remaining_ms=1500,wait_ms=1000,attempt_ms=500,tokens=1)
        self.assertTrue(retry_allowed(429,**p))
        for status in (400,401,403,404,422):
            self.assertFalse(retry_allowed(status,**p))
        for changed in ({'safe':False},{'tokens':0},{'remaining_ms':1499}):
            self.assertFalse(retry_allowed(429,**(p|changed)))

    def test_queue_growth_recovery_and_closed_loop(self):
        self.assertEqual(capacity(20,.2),100)
        self.assertEqual(capacity(20,2),10)
        backlog=(80-10)*60
        self.assertEqual(backlog,4200)
        self.assertEqual(drain(backlog,80,100),210)
        self.assertIsNone(drain(backlog,100,100))
        self.assertEqual(capacity(20,2),10)  # 20 closed-loop clients self-throttle to 10/s
        self.assertEqual((80-capacity(20,2))*60,4200)  # open arrivals still accumulate


if __name__=='__main__':
    unittest.main(verbosity=2)

import unittest
from boundary import application, edge_request, envelope, proxy_passthrough


class ContractTests(unittest.TestCase):
    def test_response_model_is_not_a_runtime_body_validator(self):
        invalid = envelope(200, {"total_cents": "200"})
        self.assertEqual(proxy_passthrough(invalid), (200, {"total_cents": "200"}))
        fixed = application('{"quantity":2}', "application/json", {"limit": "20"}, lambda _: {"total_cents": "200"})
        self.assertEqual(proxy_passthrough(fixed), (502, {"error": "invalid_provider_response"}))

    def test_valid_contract_and_invalid_200_variants(self):
        self.assertEqual(proxy_passthrough(application('{"quantity":2}', "application/json", {"limit": "20"}, lambda q: {"total_cents": q*100})), (200, {"total_cents": 200}))
        for invalid in ({}, {"total_cents": True}, {"total_cents": -1}, {"total_cents": 2, "internal_secret": "x"}, None):
            with self.subTest(invalid=invalid):
                self.assertEqual(application('{"quantity":2}', "application/json", {"limit": "20"}, lambda _, value=invalid: value)["statusCode"], 502)

    def test_basic_parameter_presence_is_not_format_validation(self):
        self.assertIsNone(edge_request('{"quantity":2}', "application/json", {"limit": "banana"}))
        result = application('{"quantity":2}', "application/json", {"limit": "banana"}, lambda _: self.fail("provider must not run"))
        self.assertEqual(result["statusCode"], 400)
        self.assertEqual(edge_request('{"quantity":2}', "application/json", {}), 400)

    def test_unmatched_content_type_and_default_model_are_separate_choices(self):
        self.assertIsNone(edge_request('{"quantity":"bad"}', "text/plain", {"limit": "20"}))
        self.assertEqual(edge_request('{"quantity":"bad"}', "text/plain", {"limit": "20"}, default_model=True), 400)
        self.assertEqual(application('{"quantity":2}', "text/plain", {"limit": "20"}, lambda _: self.fail("no provider call"))["statusCode"], 415)

    def test_invalid_requests_and_proxy_envelope(self):
        for body in ('not-json', '{"quantity":0}', '{"quantity":true}', '{}'):
            self.assertEqual(edge_request(body, "application/json", {"limit": "20"}), 400)
            self.assertEqual(application(body, "application/json", {"limit": "20"}, lambda _: self.fail("no provider call"))["statusCode"], 400)
        self.assertEqual(proxy_passthrough({"statusCode": 200, "body": {"total_cents": 200}})[0], 502)


if __name__ == "__main__":
    unittest.main()

import json
from pathlib import Path
import unittest
from product import suggest, confusion, run_task, Outcome

BASE=Path(__file__).parent
CASES=json.loads((BASE/'regression.json').read_text())


def mismatches(implementation):
    return [case['id'] for case in CASES if implementation(case['text'],allowed=case['allowed'],authorized=case['authorized']) != case['expected']]


class EvaluationTests(unittest.TestCase):
    def test_all_required_regressions_pass(self):
        self.assertEqual(len(CASES),20)
        self.assertEqual(mismatches(suggest),[])

    def test_seeded_defects_are_detected(self):
        def no_auth(text,**kwargs):
            return suggest(text,**(kwargs|{'authorized':True}))
        def ignore_allowlist(text,**kwargs):
            return suggest(text,**(kwargs|{'allowed':['database','frontend','reliability']}))
        def constant(text,**kwargs):
            return ['database']
        self.assertEqual(mismatches(no_auth),['r20'])
        self.assertEqual(mismatches(ignore_allowlist),['r18','r19'])
        self.assertGreater(len(mismatches(constant)),10)

    def test_majority_judge_counterexample(self):
        m=confusion(['pass']*99+['fail'],['pass']*100)
        self.assertEqual((m['tp'],m['tn'],m['fp'],m['fn']),(0,99,0,1))
        self.assertEqual(m['agreement'],.99)
        self.assertEqual(m['failure_recall'],0)
        self.assertIsNone(m['failure_precision'])
        self.assertIsNone(confusion([],[])['agreement'])

    def test_heldout_matrix_and_severity(self):
        row=json.loads((BASE/'heldout-labels.json').read_text())
        m=confusion(row['labels'],row['predictions'])
        self.assertEqual((m['tp'],m['tn'],m['fp'],m['fn']),(3,14,2,1))
        self.assertEqual(m['failure_recall'],.75)
        self.assertEqual(m['failure_precision'],.6)
        missed=sum(w for label,pred,w in zip(row['labels'],row['predictions'],row['severity']) if label=='fail' and pred=='pass')
        self.assertEqual(missed,10)  # the only miss is the most severe failure

    def test_budget_and_outage_fallback(self):
        r=run_task([Outcome('transient',100)]*10,allowed=['database'])
        self.assertEqual(r,dict(status='manual',tags=[],spent_cents=8,elapsed_ms=200,attempts=2))
        r=run_task([Outcome('ok',100,('database',))],allowed=['database'],budget_cents=3)
        self.assertEqual((r['attempts'],r['spent_cents']),(0,0))

    def test_whole_task_deadline_and_output_boundary(self):
        r=run_task([Outcome('transient',700),Outcome('ok',700,('database',))],allowed=['database'])
        self.assertEqual((r['status'],r['elapsed_ms'],r['spent_cents']),('manual',1000,8))
        r=run_task([Outcome('ok',10,('private-tenant-tag',))],allowed=['database'])
        self.assertEqual((r['status'],r['tags']),('manual',[]))
        r=run_task([Outcome('transient',100),Outcome('ok',100,('database',))],allowed=['database'])
        self.assertEqual((r['status'],r['tags'],r['spent_cents']),('suggestions',['database'],8))


if __name__=='__main__':
    unittest.main(verbosity=2)

"""Structural provenance/local-link checks only; not a truth or freshness oracle."""
from pathlib import Path
import re

BASE=Path(__file__).resolve().parent
ledger=(BASE/'claim-ledger.md').read_text()
for expected in ('Publication','event','Sample','primary','2025-09-22','2026-03-22','2026-09-22'):
    assert expected in ledger, expected
for n in range(1,14):
    assert f'C{n:02}:' in ledger, n
checked=0
for path in BASE.glob('*.md'):
    text=path.read_text()
    for target in re.findall(r'\]\(([^)]+)\)',text):
        if re.match(r'https?://',target) or target.startswith('#'):
            continue
        assert (path.parent/target.split('#')[0]).exists(), (path,target)
        checked+=1
print(f'13 claim IDs and {checked} local links checked; remote support/date quality requires human/source review.')