Jev production reliability checklist

"""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()