"""Offline reliability lab: bounded evaluation + SQLite outbox + local effect. Python 3.11+, standard library only. No provider, key, payment or email call. The deterministic fixtures teach application behavior, not model accuracy. """ import argparse import asyncio from contextlib import contextmanager import hashlib import json import os import sqlite3 import time import sys from math import isfinite if sys.version_info < (3, 11): raise SystemExit('This lab requires Python 3.11 or newer; check python3 --version before running it.') class ProviderError(Exception): def __init__(self, status): self.status = status @contextmanager def connect(path): db = sqlite3.connect(path, timeout=5, isolation_level=None) db.row_factory = sqlite3.Row db.executescript(""" CREATE TABLE IF NOT EXISTS decisions ( request_id TEXT PRIMARY KEY, input_hash TEXT NOT NULL, result TEXT NOT NULL); CREATE TABLE IF NOT EXISTS outbox ( event_id TEXT PRIMARY KEY, ticket_id TEXT NOT NULL, version INTEGER NOT NULL, team TEXT NOT NULL, delivered INTEGER NOT NULL DEFAULT 0); CREATE TABLE IF NOT EXISTS applied_events (event_id TEXT PRIMARY KEY); CREATE TABLE IF NOT EXISTS assignments (ticket_id TEXT PRIMARY KEY, version INTEGER NOT NULL, team TEXT NOT NULL); """) try: with db: yield db finally: db.close() def fingerprint(ticket_id, version, message): body = json.dumps([ticket_id, version, message], ensure_ascii=False).encode() return hashlib.sha256(body).hexdigest() def cached(db, request_id, input_hash): row = db.execute("SELECT * FROM decisions WHERE request_id=?", (request_id,)).fetchone() if row and row["input_hash"] != input_hash: raise ValueError("Idempotency key was reused for a different input") return json.loads(row["result"]) if row else None def valid_route(answer): c = answer.get("confidence") if isinstance(answer, dict) else None return (isinstance(answer, dict) and answer.get("type") == "choice" and answer.get("choice") in {"technical", "billing", "human"} and isinstance(c, (int, float)) and not isinstance(c, bool) and isfinite(c) and 0 <= c <= 1) async def evaluate(provider, *, attempt_timeout=0.02, total_timeout=0.07, max_attempts=2): """Provider must be an async callable with no application side effects.""" deadline = time.monotonic() + total_timeout attempts = 0 reason = "budget_exhausted" while attempts < max_attempts: remaining = deadline - time.monotonic() if remaining <= 0: break attempts += 1 try: answer = await asyncio.wait_for(provider(attempts), min(attempt_timeout, remaining)) except TimeoutError: reason = "timeout" except ProviderError as error: reason = "provider_failure" if error.status not in {429, 500, 502, 503, 504}: break except Exception: reason = "provider_failure" break else: if not valid_route(answer): reason = "invalid_response" break if answer["choice"] == "human" or answer["confidence"] < 0.75: return {"action": "review", "reason": "policy", "attempts": attempts} return {"action": "queue_assignment", "team": answer["choice"], "attempts": attempts} if attempts < max_attempts: # Bounded demonstration backoff; a live adapter must also bound Retry-After. await asyncio.sleep(min(0.005, max(0, deadline - time.monotonic()))) return {"action": "review", "reason": reason, "attempts": attempts} async def ingest(path, ticket_id, version, message, provider, **budget): request_id = f"{ticket_id}:{version}" digest = fingerprint(ticket_id, version, message) with connect(path) as db: existing = cached(db, request_id, digest) if existing is not None: return existing result = await evaluate(provider, **budget) with connect(path) as db: db.execute("BEGIN IMMEDIATE") # Another worker may have saved a decision while we awaited the provider. existing = cached(db, request_id, digest) if existing is not None: return existing db.execute("INSERT INTO decisions VALUES (?, ?, ?)", (request_id, digest, json.dumps(result))) if result["action"] == "queue_assignment": db.execute("INSERT INTO outbox(event_id,ticket_id,version,team) VALUES (?,?,?,?)", (request_id, ticket_id, version, result["team"])) return result def deliver(path, *, crash_before_commit=False, hard_crash_before_commit=False): """The demonstrated side effect stays INSIDE this same SQLite transaction.""" applied = 0 with connect(path) as db: db.execute("BEGIN IMMEDIATE") for row in db.execute("SELECT * FROM outbox WHERE delivered=0 ORDER BY event_id").fetchall(): fresh = db.execute("INSERT OR IGNORE INTO applied_events VALUES (?)", (row["event_id"],)).rowcount if fresh: db.execute("INSERT INTO assignments VALUES (?, ?, ?) ON CONFLICT(ticket_id) DO UPDATE SET version=excluded.version, team=excluded.team WHERE excluded.version > assignments.version", (row["ticket_id"], row["version"], row["team"])) applied += 1 if hard_crash_before_commit: # Fault drill in a disposable database: bypass Python rollback/cleanup. os._exit(86) if crash_before_commit: raise RuntimeError("Injected crash: local update and marker must roll back together") db.execute("UPDATE outbox SET delivered=1 WHERE event_id=?", (row["event_id"],)) return applied def snapshot(path): with connect(path) as db: return {name: [dict(row) for row in db.execute(f"SELECT * FROM {name}")] for name in ("decisions", "outbox", "applied_events", "assignments")} def fixture(scenario): async def provider(attempt): if scenario == "timeout": await asyncio.sleep(1) if scenario == "auth": raise ProviderError(401) if scenario == "transient" and attempt == 1: raise ProviderError(503) if scenario == "invalid": return {"type": "choice", "choice": "invented", "confidence": 0.99} return {"type": "choice", "choice": "technical", "confidence": 0.8} return provider def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("command", choices=["ingest", "deliver", "inspect"]) parser.add_argument("--db", required=True, help="Use a disposable local SQLite file") parser.add_argument("--scenario", default="ok", choices=["ok", "transient", "timeout", "auth", "invalid"]) parser.add_argument("--ticket", default="ticket-7") parser.add_argument("--version", type=int, default=1) parser.add_argument("--crash-before-commit", action="store_true") parser.add_argument("--hard-crash-before-commit", action="store_true", help="Fault drill: terminate this process with exit 86 before SQLite commit") args = parser.parse_args() if args.command == "ingest": result = asyncio.run(ingest(args.db, args.ticket, args.version, "Checkout is unavailable", fixture(args.scenario))) elif args.command == "deliver": result = {"applied": deliver(args.db, crash_before_commit=args.crash_before_commit, hard_crash_before_commit=args.hard_crash_before_commit)} else: result = snapshot(args.db) print(json.dumps({"offline_fixture": True, "result": result}, indent=2)) if __name__ == "__main__": main()