#!/usr/bin/env python3
"""Bounded reliability models and local SQLite transaction experiments.

No Kafka, CDC, ZooKeeper, network faults or physical clock is simulated as real.
SQLite uses real temporary files/transactions; its locks are not InnoDB locks.
"""
import argparse
import hashlib
import copy
import json
import sqlite3
import tempfile
import uuid
from contextlib import closing
from pathlib import Path


class Ledger:
    """Public operations reopen a file connection; durable effects share a tx."""
    def __init__(self, path):
        self.path = path
        with closing(self.connect()) as db:
            db.executescript('''
              CREATE TABLE account(balance INTEGER NOT NULL);
              INSERT INTO account VALUES(0);
              CREATE TABLE inbox(operation_id TEXT PRIMARY KEY, payload TEXT NOT NULL,
                                 state TEXT NOT NULL, result INTEGER NOT NULL);
              CREATE TABLE effects(operation_id TEXT PRIMARY KEY, amount INTEGER NOT NULL);
            ''')

    def connect(self):
        return sqlite3.connect(self.path, isolation_level=None)

    def apply(self, operation_id, amount, crash=None):
        payload = json.dumps({'amount': amount}, sort_keys=True)
        with closing(self.connect()) as db:
            db.execute('BEGIN IMMEDIATE')
            prior = db.execute('SELECT payload,state,result FROM inbox WHERE operation_id=?', (operation_id,)).fetchone()
            if prior:
                db.execute('ROLLBACK')
                return {'status': 'DUPLICATE' if prior[0] == payload else 'CONFLICT', 'result': prior[2]}
            db.execute('INSERT INTO effects VALUES(?,?)', (operation_id, amount))
            db.execute('UPDATE account SET balance=balance+?', (amount,))
            result = db.execute('SELECT balance FROM account').fetchone()[0]
            db.execute('INSERT INTO inbox VALUES(?,?,?,?)', (operation_id, payload, 'SUCCEEDED', result))
            if crash == 'before-commit':
                # Connection loss before COMMIT rolls back all three writes.
                db.execute('ROLLBACK')
                return {'status': 'ROLLED_BACK', 'result': None}
            db.execute('COMMIT')
            return {'status': 'COMMITTED_RESPONSE_LOST' if crash == 'after-commit' else 'SUCCEEDED', 'result': result}

    def status(self):
        with closing(self.connect()) as db:
            return {'balance': db.execute('SELECT balance FROM account').fetchone()[0],
                    'inbox': db.execute('SELECT count(*) FROM inbox').fetchone()[0],
                    'effects': db.execute('SELECT count(*) FROM effects').fetchone()[0]}


def inbox(case):
    with tempfile.TemporaryDirectory(prefix='study_t20_inbox_') as directory:
        ledger = Ledger(Path(directory) / 'ledger.sqlite')
        if case == 'conflict':
            results = [ledger.apply('op-001', amount) for amount in (100, 200, 100)]
        elif case == 'setnx':
            # A persistent reservation in another resource cannot atomically
            # decide whether this ledger transaction has committed.
            with closing(sqlite3.connect(Path(directory) / 'reservations.sqlite')) as db:
                db.execute('CREATE TABLE reservations(operation_id TEXT PRIMARY KEY)')
                db.execute('INSERT INTO reservations VALUES(?)', ('op-001',))
                db.commit()  # injected stop here: no ledger effect occurred
                reserved = db.execute('SELECT 1 FROM reservations WHERE operation_id=?', ('op-001',)).fetchone()
            return {'case': case, 'status': 'DROPPED_AFTER_RESERVATION' if reserved else 'RETRY', 'persisted': ledger.status()}
        elif case == 'external':
            return external_payment(directory)
        elif case in ('before-commit', 'after-commit'):
            results = [ledger.apply('op-001', 100, case), ledger.apply('op-001', 100)]
            if case == 'before-commit':
                results.append(ledger.apply('op-001', 100))
        else:
            raise ValueError('unsupported inbox case: ' + case)
        return {'case': case, 'storage': 'temporary-sqlite-file',
                'statuses': [result['status'] for result in results],
                'replayed_result': results[-1]['result'], 'persisted': ledger.status()}


def external_payment(directory):
    """A declared provider model, with a distinct durable commit boundary."""
    local_path, provider_path = Path(directory) / 'operations.sqlite', Path(directory) / 'provider.sqlite'
    with closing(sqlite3.connect(local_path)) as db:
        db.execute('CREATE TABLE operations(client_key TEXT PRIMARY KEY, state TEXT, receipt TEXT)')
        db.executemany('INSERT INTO operations VALUES(?,?,NULL)', [('client-7', 'PROCESSING'), ('receipt-only-8', 'PROCESSING')])
        db.commit()  # both client identities are known before any request
    with closing(sqlite3.connect(provider_path)) as db:
        db.execute('CREATE TABLE charges(receipt TEXT PRIMARY KEY, client_key TEXT UNIQUE, amount INTEGER)')
        # Provider commits, then response is lost. Only the first contract allows
        # querying by the pre-known client key; the second exposes receipt only.
        db.executemany('INSERT INTO charges VALUES(?,?,?)', [('receipt-7', 'client-7', 100), ('receipt-8', None, 100)])
        db.commit()
    with closing(sqlite3.connect(local_path)) as db:
        db.execute("UPDATE operations SET state='UNKNOWN'")
        db.commit()
    with closing(sqlite3.connect(provider_path)) as db:
        found = db.execute('SELECT receipt FROM charges WHERE client_key=?', ('client-7',)).fetchone()
        # Reissue the same provider idempotency key: no second charge.
        db.execute('INSERT OR IGNORE INTO charges VALUES(?,?,?)', ('retry-receipt-7', 'client-7', 100))
        db.commit()
        counts = [db.execute('SELECT count(*) FROM charges WHERE client_key=?', ('client-7',)).fetchone()[0],
                  db.execute('SELECT count(*) FROM charges WHERE client_key IS NULL').fetchone()[0]]
    with closing(sqlite3.connect(local_path)) as db:
        db.execute('UPDATE operations SET state=?,receipt=? WHERE client_key=? AND state=?',
                   ('SUCCEEDED', found[0], 'client-7', 'UNKNOWN'))
        db.commit()
        states = dict(db.execute('SELECT client_key,state FROM operations'))
    return {'case': 'external', 'provider': 'separate-sqlite-model-not-network',
            'client_key': {'initial': 'UNKNOWN', 'recovered': states['client-7'], 'attempts': 2, 'charges': counts[0]},
            'receipt_only': {'initial': 'UNKNOWN', 'recovered': states['receipt-only-8'], 'charges': counts[1]}}


class Outbox:
    def __init__(self, path):
        self.path = path
        with closing(self.connect()) as db:
            db.executescript('''
              CREATE TABLE orders(id INTEGER PRIMARY KEY, status TEXT NOT NULL);
              CREATE TABLE outbox(event_id TEXT PRIMARY KEY, order_id INTEGER NOT NULL,
                 payload INTEGER NOT NULL, state TEXT NOT NULL, available_at INTEGER NOT NULL,
                 claim_version INTEGER NOT NULL DEFAULT 0, lease_until INTEGER,
                 owner TEXT, claim_id TEXT);
            ''')

    def connect(self):
        return sqlite3.connect(self.path, isolation_level=None)

    def create_order(self, order_id=7, rollback=False):
        with closing(self.connect()) as db:
            db.execute('BEGIN IMMEDIATE')
            db.execute('INSERT INTO orders VALUES(?,?)', (order_id, 'CREATED'))
            db.execute('INSERT INTO outbox(event_id,order_id,payload,state,available_at) VALUES(?,?,?,?,?)',
                       (f'e-{order_id}-1', order_id, 100, 'PENDING', 0))
            db.execute('ROLLBACK' if rollback else 'COMMIT')

    def claim(self, owner, now, event_id='e-7-1'):
        # SQLite serializes writers here. This is NOT SKIP LOCKED or a simulation
        # of InnoDB locks. Read the new token before releasing the same tx.
        with closing(self.connect()) as db:
            db.execute('BEGIN IMMEDIATE')
            claim_id = uuid.uuid4().hex
            changed = db.execute('''UPDATE outbox SET state='CLAIMED', owner=?,
                claim_version=claim_version+1, lease_until=?, claim_id=?
                WHERE event_id=? AND available_at<=? AND
                (state='PENDING' OR (state='CLAIMED' AND lease_until<=?))''',
                (owner, now + 10, claim_id, event_id, now, now)).rowcount
            row = db.execute('SELECT claim_version,payload FROM outbox WHERE event_id=?', (event_id,)).fetchone() if changed else None
            db.execute('COMMIT')
            return {'event_id': event_id, 'version': row[0], 'payload': row[1], 'owner': owner, 'claim_id': claim_id} if row else None

    def mark_sent(self, claim):
        with closing(self.connect()) as db:
            return db.execute('''UPDATE outbox SET state='SENT',lease_until=NULL
                WHERE event_id=? AND state='CLAIMED' AND claim_version=? AND owner=? AND claim_id=?''',
                (claim['event_id'], claim['version'], claim['owner'], claim['claim_id'])).rowcount

    def status(self):
        with closing(self.connect()) as db:
            return {'orders': db.execute('SELECT count(*) FROM orders').fetchone()[0],
                    'events': [{'event_id': row[0], 'state': row[1], 'claim_version': row[2]}
                               for row in db.execute('SELECT event_id,state,claim_version FROM outbox ORDER BY event_id')]}


def outbox(case):
    with tempfile.TemporaryDirectory(prefix='study_t20_outbox_') as directory:
        sender = Outbox(Path(directory) / 'sender.sqlite')
        receiver = Ledger(Path(directory) / 'receiver.sqlite')
        sender.create_order(rollback=case == 'rollback')
        result = {'case': case, 'storage': 'temporary-sqlite-file', 'send_attempts': 0}
        if case in ('rollback', 'committed'):
            pass
        elif case == 'send-before-mark':
            first = sender.claim('A', 0)
            receiver.apply(first['event_id'], first['payload'])
            # Provider ack arrived, but local SENT update did not happen.
            second = sender.claim('B', 11)
            receiver.apply(second['event_id'], second['payload'])
            sender.mark_sent(second)
            result['send_attempts'] = 2
        elif case == 'mark-before-send':
            first = sender.claim('A', 0)
            sender.mark_sent(first)  # intentional defect, then injected stop
            result['lost_intent'] = sender.claim('B', 11) is None and receiver.status()['effects'] == 0
        elif case == 'stale-claim':
            first = sender.claim('A', 0)
            second = sender.claim('B', 11)
            # A resumes with its OWN returned token, not SELECT latest version.
            # Neither CAS can retract an already emitted external request.
            receiver.apply(first['event_id'], first['payload'])
            result['old_mark_rows'] = sender.mark_sent(first)
            receiver.apply(second['event_id'], second['payload'])
            result['new_mark_rows'] = sender.mark_sent(second)
            result.update(claim_tokens=[first['version'], second['version']], send_attempts=2)
        elif case == 'batch':
            sender.create_order(8)
            first = sender.claim('A', 0)
            sender.claim('A', 0, 'e-8-1')
            receiver.apply(first['event_id'], first['payload'])
            sender.mark_sent(first)  # only this item's response was confirmed
            result['send_attempts'] = 1
        else:
            raise ValueError('unsupported outbox case: ' + case)
        result.update(persisted=sender.status(), receiver=receiver.status())
        return result


class Reservation:
    """Finite participant model: each transition is declared locally atomic.

    Not an XA/TCC coordinator implementation, and not a network experiment.
    """
    def __init__(self):
        self.state = None
        self.quantity = None
        self.available, self.reserved, self.sold = 1, 0, 0

    def apply(self, action, quantity=1):
        if self.quantity is not None and quantity != self.quantity:
            return 'CONFLICT'
        if self.state is None:
            if action == 'confirm':
                return 'REJECTED'
            self.quantity = quantity
            if action == 'cancel':
                self.state = 'CANCELLED'  # durable tombstone is an assumption
            elif quantity <= self.available:
                self.available -= quantity
                self.reserved += quantity
                self.state = 'RESERVED'
            else:
                return 'INSUFFICIENT'
        elif self.state == 'RESERVED':
            if action == 'confirm':
                self.reserved -= quantity
                self.sold += quantity
                self.state = 'CONFIRMED'
            elif action == 'cancel':
                self.reserved -= quantity
                self.available += quantity
                self.state = 'CANCELLED'
        elif ((self.state == 'CANCELLED' and action != 'cancel') or
              (self.state == 'CONFIRMED' and action == 'cancel')):
            return 'REJECTED'
        return self.state

    def counts(self):
        return {'available': self.available, 'reserved': self.reserved, 'sold': self.sold}


def workflow():
    empty = Reservation()
    empty_statuses = [empty.apply(action) for action in ('cancel', 'try', 'cancel')]
    conflict = Reservation()
    conflicting = [conflict.apply('try'), conflict.apply('try', 2)]
    races = []
    for actions in (('confirm', 'cancel'), ('cancel', 'confirm')):
        participant = Reservation()
        statuses = [participant.apply('try')] + [participant.apply(action) for action in actions]
        races.append({'order': actions, 'statuses': statuses, 'final': participant.state, **participant.counts()})
    participants = [Reservation(), Reservation()]
    for participant in participants:
        participant.apply('try')
    with tempfile.TemporaryDirectory(prefix='study_t20_decision_') as directory:
        path = Path(directory) / 'coordinator.sqlite'
        with closing(sqlite3.connect(path)) as db:
            db.execute('CREATE TABLE decisions(workflow_id TEXT PRIMARY KEY, decision TEXT NOT NULL)')
            db.execute('INSERT INTO decisions VALUES(?,?)', ('order-7', 'CONFIRM'))
            db.commit()  # decision is durable before notifying either participant
        participants[0].apply('confirm')
        # Coordinator resumes with a new connection, not an in-memory choice.
        with closing(sqlite3.connect(path)) as db:
            decision = db.execute('SELECT decision FROM decisions WHERE workflow_id=?', ('order-7',)).fetchone()[0]
            # Try to choose a contradictory decision through the same unique identity.
            db.execute('INSERT OR IGNORE INTO decisions VALUES(?,?)', ('order-7', 'CANCEL'))
            db.commit()
            recovered = db.execute('SELECT decision FROM decisions WHERE workflow_id=?', ('order-7',)).fetchone()[0]
            rejected_reverse = recovered == decision == 'CONFIRM'
        participants[1].apply(decision.lower())
    split = [Reservation(), Reservation()]
    for participant in split:
        participant.apply('try')
    split[0].apply('confirm')
    split[1].apply('cancel')
    return {'kind': 'finite-participant-model-with-sqlite-decision',
            'empty_cancel': {'statuses': empty_statuses, **empty.counts()},
            'conflict': {'statuses': conflicting, **conflict.counts()},
            'terminal_races': races,
            'recovery': {'decision': decision, 'states': [p.state for p in participants],
                         'rejected_reverse': rejected_reverse},
            'split_decision': {'states': [p.state for p in split], 'violation': 'GLOBAL_DECISION_SPLIT'}}


def schedules(dependencies, prefix=()):
    """Enumerate every topological ordering, not just one hand-picked trace."""
    if len(prefix) == len(dependencies):
        yield prefix
        return
    done = set(prefix)
    for event, required in dependencies.items():
        if event not in done and required <= done:
            yield from schedules(dependencies, prefix + (event,))


def execute(events, variant):
    state = {'owner': None, 'generation': 0, 'issuer': 40,
             'tickets': {}, 'resource': {'fence': 0, 'op_seq': 0, 'value': ''},
             'resource_generation': 0, 'business': {}, 'unbound': set()}
    trace = []
    violation = None
    for event in events:
        accepted = None
        if event.startswith('grant-'):
            client = event[-1]
            assert state['owner'] is None
            state['generation'] += 1
            state['issuer'] += 1
            state['owner'] = client
            # This transition is one atomic authority operation.
            state['tickets'][client] = (state['issuer'], state['generation'])
        elif event == 'expire-A':
            state['owner'] = None  # Expiry never changes resource state.
        elif event == 'restore-issuer':
            # A correct restart reads the durable high water. Fault injection
            # deliberately substitutes a stale snapshot while resources survive.
            if variant == 'rollback-issuer':
                state['issuer'] = 40
        elif event == 'mint-A':
            # Deliberate defect: an independent issuer trusts an old local check.
            state['issuer'] += 1
            state['tickets']['A'] = (state['issuer'], 1)
            state['unbound'].add(state['issuer'])
        elif event.startswith('write-'):
            client = event[-1]
            fence, generation = state['tickets'][client]
            seq = 2 if client == 'A' and event == 'write-A' else 1
            key = f'{client}-{seq}'
            resource = state['resource']
            # Comparison, idempotency record and effect are ONE atomic transition.
            newer = (fence, seq) > (resource['fence'], resource['op_seq'])
            accepted = key not in state['business'] and (newer or variant == 'no-fence')
            if accepted:
                if fence in state['unbound']:
                    violation = 'unauthorized_accept'
                elif generation < state['resource_generation']:
                    violation = 'revoked_generation_overwrite'
                state['resource'] = {'fence': fence, 'op_seq': seq, 'value': client}
                state['resource_generation'] = generation
                state['business'][key] = client
        trace.append({'event': event, 'accepted': accepted, 'resource': copy.deepcopy(state['resource'])})
    return {'events': list(events), 'trace': trace, 'violation': violation}


def fencing(variant):
    deps = {'grant-A': set(), 'pause-A': {'grant-A'}, 'expire-A': {'pause-A'},
            'restore-issuer': {'expire-A'}, 'grant-B': {'restore-issuer'}, 'resume-A': {'grant-B'},
            'write-A': {'resume-A'}, 'write-B': {'grant-B'}}
    if variant == 'split-authority':
        deps['mint-A'] = {'resume-A', 'write-B'}
        deps['write-A'] = {'mint-A'}
    histories = [execute(events, variant) for events in schedules(deps)]
    examples = {}
    for history in histories:
        old = next(step for step in history['trace'] if step['event'] == 'write-A')
        after = history['events'].index('write-A') > history['events'].index('write-B')
        examples['oldAfterNew' if after else 'oldBeforeNew'] = old
    return {'variant': variant, 'histories': len(histories),
            'violation': next((h['violation'] for h in histories if h['violation']), None),
            'examples': examples, 'counterexample': next((h for h in histories if h['violation']), None)}


def operations():
    """Resource-only transaction model; granted tickets are input assumptions."""
    resource = {'fence': 0, 'op_seq': 0, 'value': ''}
    business, results = {}, []
    inputs = [(41, 1, 'invoice-1', 'first'), (41, 2, 'invoice-2', 'second'),
              (42, 1, 'invoice-1', 'first'), (42, 2, 'invoice-1', 'changed'),
              (41, 1, 'invoice-3', 'late')]
    for fence, seq, key, payload in inputs:
        # Entire block is one transaction, including the identity check.
        if key in business:
            status = 'duplicate' if business[key] == payload else 'conflict'
        elif (fence, seq) <= (resource['fence'], resource['op_seq']):
            status = 'stale'
        else:
            status = 'applied'
            business[key] = payload
            resource = {'fence': fence, 'op_seq': seq, 'value': payload}
        results.append({'fence': fence, 'op_seq': seq, 'key': key, 'payload': payload, 'status': status})
    return {'case': 'operations', 'results': results, 'effects': len(business), 'resource': resource}


def replication():
    """Fixed stable-ISR snapshots; no elections, expansion or disk simulation."""
    hw, required, min_isr = 103, 105, 2
    stages = []
    for leos, isr in (([105, 103, 103], [0, 1, 2]),
                      ([105, 105, 105], [0, 1, 2]),
                      ([105, 105, 105], [0])):
        # An already established HW does not roll back on ISR shrink. Kafka's
        # real maximalIsr/reconfiguration paths are outside these snapshots.
        if len(isr) >= min_isr:
            hw = max(hw, min(leos[replica] for replica in isr))
        status = 'ERROR_AFTER_APPEND' if len(isr) < min_isr else ('ACK' if hw >= required else 'WAIT')
        stages.append(dict(leos=leos, isr=isr, min_isr=min_isr,
                           required_offset=required, hw=hw, status=status, appended=True))
    visible_hw, first_open = 110, 104
    lso = min(visible_hw, first_open)
    return dict(model='stable-isr-boundary-not-kafka-cluster', observations='independent-snapshots-no-ack-revocation', stages=stages,
                visibility=dict(hw=visible_hw, open_transaction=first_open, lso=lso,
                                visible_107=107 < lso, settled_lso=visible_hw))


def offsets(case):
    """One partition's delivered prefix; ownership is an explicit input event."""
    delivered, queue, completed, safe = [100, 103, 108], [100, 103, 108], set(), 100
    epoch, owned, trace = 1, True, []

    def complete(offset, notification_epoch):
        nonlocal safe
        reason = 'not-owner' if not owned else ('old-assignment' if notification_epoch != epoch else 'completed')
        accepted = reason == 'completed'
        if accepted:
            completed.add(offset)
            while queue and queue[0] in completed:
                safe = queue.pop(0) + 1
        trace.append(dict(offset=offset, notification_epoch=notification_epoch,
                          assignment_epoch=epoch, accepted=accepted, reason=reason, safe_offset=safe))

    # Execute the deliberately unsafe rule against the same arrival history.
    unsafe_completed, counterexample = set(), None
    for offset in (108, 100, 103):
        unsafe_completed.add(offset)
        unsafe_offset = max(unsafe_completed) + 1
        skipped = [value for value in delivered if value < unsafe_offset and value not in unsafe_completed]
        if skipped and counterexample is None:
            counterexample = dict(unsafe_offset=unsafe_offset, skipped=skipped)
    replayed = []
    if case == 'handoff':
        complete(100, 1)
        owned = False
        complete(103, 1)  # Late effect may exist, but this owner cannot commit.
        replayed = [offset for offset in delivered if offset >= safe]
        epoch, owned, queue, completed = 2, True, replayed.copy(), set()
        complete(108, 1)  # Old worker cannot complete a new assignment's work.
        complete(108, 2)
        complete(103, 2)
        complete(108, 2)  # Repeated current notification is harmless.
    else:
        if case == 'stale':
            epoch = 2
        for offset in (108, 100, 103):
            complete(offset, 1)
    return dict(case=case, delivered=delivered, trace=trace, replayed=replayed,
                counterexample=counterexample,
                model='delivered-prefix-not-group-protocol')


def backlog(case):
    """Fluid capacity calculation and one object's durable deferred-event model."""
    capacity = []
    for service_rate in (10000, 7000, min(12 * 1000, 9000)):
        backlog_size, arrivals = 1200000, 8000
        net = service_rate - arrivals
        capacity.append(dict(backlog=backlog_size, arrivals_per_second=arrivals,
                             completions_per_second=service_rate, net_per_second=net,
                             drain_seconds=backlog_size / net if net > 0 else None))
    with tempfile.TemporaryDirectory(prefix='study_t21_order_') as directory:
        filename = Path(directory) / 'projection.sqlite'
        with closing(sqlite3.connect(filename)) as db:
            db.executescript('''CREATE TABLE projection(version INTEGER);
                INSERT INTO projection VALUES(10);
                CREATE TABLE events(version INTEGER PRIMARY KEY, state TEXT, due INTEGER);
                CREATE TABLE effects(sequence INTEGER PRIMARY KEY AUTOINCREMENT, version INTEGER UNIQUE);''')
        statuses = []
        deferred_durable = False
        inputs = {'normal': (12, 11, 11, 13), 'stale': (9, 12, 11, 11, 13),
                  'deferred-repeat': (12, 12, 11, 11, 13)}[case]
        for version in inputs:
            # Reopen every time: DEFERRED event version/state survives a
            # process-local cache loss. All effect/state changes share one tx.
            with closing(sqlite3.connect(filename)) as db:
                current = db.execute('SELECT version FROM projection').fetchone()[0]
                known = db.execute('SELECT state FROM events WHERE version=?', (version,)).fetchone()
                if known:
                    statuses.append('DEFERRED' if known[0] == 'DEFERRED' else 'DUPLICATE')
                elif version <= current:
                    statuses.append('STALE')
                elif version > current + 1:
                    db.execute("INSERT INTO events VALUES(?,'DEFERRED',10)", (version,))
                    statuses.append('DEFERRED')
                else:
                    db.execute("INSERT INTO events VALUES(?,'APPLIED',NULL)", (version,))
                    db.execute('INSERT INTO effects(version) VALUES(?)', (version,))
                    current = version
                    while db.execute("SELECT 1 FROM events WHERE version=? AND state='DEFERRED'", (current + 1,)).fetchone():
                        current += 1
                        db.execute("UPDATE events SET state='APPLIED' WHERE version=?", (current,))
                        db.execute('INSERT INTO effects(version) VALUES(?)', (current,))
                    db.execute('UPDATE projection SET version=?', (current,))
                    statuses.append('APPLIED')
                db.commit()
            if version == 12:
                with closing(sqlite3.connect(filename)) as db:
                    deferred_durable = db.execute('SELECT state FROM events WHERE version=12').fetchone()[0] == 'DEFERRED'
        with closing(sqlite3.connect(filename)) as db:
            applied = [row[0] for row in db.execute('SELECT version FROM effects ORDER BY sequence')]
            final = db.execute('SELECT version FROM projection').fetchone()[0]
            # A later 15 waits for absent 14; explicit time 11 exceeds due 10.
            db.execute("INSERT INTO events VALUES(15,'DEFERRED',10)")
            db.execute("UPDATE events SET state='ISOLATE_AND_RECONCILE' WHERE state='DEFERRED' AND due<=11")
            db.commit()
        with closing(sqlite3.connect(filename)) as db:
            gap_timeout = db.execute('SELECT state FROM events WHERE version=15').fetchone()[0]
    unsafe = []
    for version in (12, 11):
        unsafe.append(version)  # Intentional bypass: apply every arrival.
    return dict(case=case, capacity=capacity, model='fluid-calculation-and-sqlite-version-model',
                order=dict(unsafe=dict(applied=unsafe, final_version=unsafe[-1]),
                           guarded=dict(statuses=statuses, applied=applied, final_version=final,
                                        deferred_durable=deferred_durable, gap_timeout=gap_timeout)))


def config_model():
    """Application parse/validate/swap contract, not Nacos client internals."""
    def parse(raw):
        candidate = json.loads(raw)
        if (not isinstance(candidate.get('version'), int) or candidate['version'] <= 0 or
                not isinstance(candidate.get('rules'), str) or not candidate['rules'] or
                not isinstance(candidate.get('limit'), int) or isinstance(candidate['limit'], bool) or candidate['limit'] <= 0):
            raise ValueError('invalid version/rules/limit schema')
        return candidate

    initial = '{"version":1,"rules":"old","limit":10}'
    invalid = '{"version":2,"rules":"new","limit":"oops"}'
    final = '{"version":3,"rules":"next","limit":30}'
    runtime, revisions = None, []
    for raw in (initial, invalid, final):
        try:
            candidate = parse(raw)  # all validation happens before the swap
            runtime = candidate
            status = 'APPLIED'
        except (ValueError, TypeError):
            status = 'REJECTED'
            safe_at_invalid = runtime['version']
        revisions.append({'status': status, 'runtime_version': runtime['version']})
    bad_runtime = json.loads(initial)
    bad_candidate = json.loads(invalid)
    bad_runtime['rules'] = bad_candidate['rules']  # intentional partial mutation
    try:
        bad_runtime['limit'] = int(bad_candidate['limit'])
    except ValueError:
        pass  # an exception does not undo already changed runtime objects
    docs = {'rules': {'version': 1, 'value': 'old'}, 'limits': {'version': 1, 'value': 10}}
    docs['rules'] = {'version': 2, 'value': 'new'}
    mixed = {key: value['version'] for key, value in docs.items()}
    targets = {'rules-v2': '{"version":2,"rules":"new"}',
               'limits-v2': '{"version":2,"limit":20}'}
    digest = lambda raw: hashlib.sha256(raw.encode()).hexdigest()
    manifest = {'version': 2, 'refs': {key: digest(raw) for key, raw in targets.items()}}
    available = {'rules-v2': targets['rules-v2']}
    active, steps = parse(initial), []
    for arrival in (None, '{"version":2,"limit":999}', targets['limits-v2']):
        if arrival is not None:
            available['limits-v2'] = arrival
        if any(key not in available for key in manifest['refs']):
            status = 'WAITING'
        elif any(digest(available[key]) != wanted for key, wanted in manifest['refs'].items()):
            status = 'HASH_CONFLICT'
        else:
            pieces = [json.loads(available[key]) for key in manifest['refs']]
            candidate = {'version': manifest['version']}
            if any(part['version'] != manifest['version'] for part in pieces):
                status = 'VERSION_CONFLICT'
            else:
                for part in pieces:
                    candidate.update(part)
                active = parse(json.dumps(candidate))
                status = 'APPLIED'
        steps.append({'status': status, 'runtime_version': active['version']})
    return {'model': 'finite-application-snapshot-not-nacos-server', 'revisions': revisions,
            'split_data_ids': {'mixed': mixed},
            'manifest': {'steps': steps, 'snapshot': active},
            'partial_mutation': {'bad_rules': bad_runtime['rules'], 'bad_limit': bad_runtime['limit'],
                                 'safe_version': safe_at_invalid}}


def session_model():
    """Declared session/watch transitions, not a ZooKeeper process or clock."""
    sessions = {'A': 'CONNECTED'}
    owner, trace = 'A', []
    for event in ('disconnect-A', 'reconnect-A', 'disconnect-A', 'expire-A', 'create-B', 'resume-A'):
        action, actor = event.split('-')
        if action == 'disconnect':
            sessions[actor] = 'DISCONNECTED'
        elif action == 'reconnect' and sessions[actor] != 'EXPIRED':
            sessions[actor] = 'CONNECTED'
        elif action == 'expire':
            sessions[actor] = 'EXPIRED'
            if owner == actor:
                owner = None
        elif action == 'create':
            sessions[actor] = 'CONNECTED'
            if owner is None:
                owner = actor
        elif action == 'resume':
            # A cannot reuse an expired session; its process is still alive.
            old_rejected = sessions[actor] == 'EXPIRED'
        trace.append({'event': event, 'actor': actor, 'client': sessions[actor], 'ephemeral': owner})
    armed, notifications, current = True, 0, 1
    versions = []
    for current in (2, 3):
        versions.append(current)
        if armed:
            notifications += 1
            armed = False  # ordinary one-shot watch, not an event journal
    blind = ['guid-A-0001']  # creation committed; its response was lost
    recovered = blind.copy()
    blind.append('guid-A-0002')
    matches = [name for name in recovered if name.startswith('guid-A-')]
    if not matches:
        recovered.append('guid-A-0002')
    return {'model': 'finite-session-watch-not-zookeeper-cluster', 'trace': trace,
            'old_session_rejected': old_rejected,
            'watch': {'notifications': notifications, 'reread': current,
                      'missed_intermediate': next(v for v in versions if v != current)},
            'create_unknown': {'blind_candidates': len(blind), 'recovered_candidates': len(recovered)}}


def budget(total=800, fixed=100, attempt=150, backoff=80, first=None, max_attempts=3):
    """Conservative admission: a retry reserves backoff AND full attempt cost.

    Costs are explicit deterministic inputs, not a real monotonic clock or RPC.
    """
    first = attempt if first is None else first
    remaining = total - fixed
    attempts = []
    for index in range(max_attempts):
        cost = first if index == 0 else attempt
        delay = 0 if index == 0 else backoff
        if remaining < delay + cost:
            stop = 'INSUFFICIENT_DEADLINE'
            break
        remaining -= delay + cost
        attempts.append({'attempt': index + 1, 'backoff_ms': delay,
                         'elapsed_ms': cost, 'remaining_ms': remaining})
    else:
        stop = 'ATTEMPT_LIMIT'
    naive_cost = fixed + first + (max_attempts - 1) * (backoff + attempt)
    return {'model': 'deterministic-deadline-admission-not-network',
            'inputs': {'total_ms': total, 'fixed_ms': fixed, 'attempt_ms': attempt,
                       'backoff_ms': backoff, 'first_attempt_ms': first, 'max_attempts': max_attempts},
            'attempts': attempts, 'remaining_ms': remaining, 'stop': stop,
            'amplification': 3 ** 3,
            'counterexample': {'policy': 'count-only-no-deadline', 'attempts': max_attempts,
                               'remaining_ms': total - naive_cost}}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('scenario', choices=['fencing', 'inbox', 'outbox', 'workflow', 'replication', 'offsets', 'backlog', 'budget', 'session', 'config'])
    parser.add_argument('--variant', choices=['correct', 'split-authority', 'rollback-issuer', 'no-fence'], default='correct')
    parser.add_argument('--case', default=None)
    parser.add_argument('--json', action='store_true')
    parser.add_argument('--total-ms', type=int, default=800)
    parser.add_argument('--fixed-ms', type=int, default=100)
    parser.add_argument('--attempt-ms', type=int, default=150)
    parser.add_argument('--backoff-ms', type=int, default=80)
    parser.add_argument('--first-attempt-ms', type=int)
    parser.add_argument('--max-attempts', type=int, default=3)
    args = parser.parse_args()
    if min(args.total_ms, args.attempt_ms, args.max_attempts) <= 0 or min(args.fixed_ms, args.backoff_ms) < 0 or (args.first_attempt_ms is not None and args.first_attempt_ms <= 0):
        parser.error('budget costs require positive total/attempt/first/max-attempts and nonnegative fixed/backoff')
    if args.scenario == 'config':
        if args.case is not None:
            parser.error('config has one fixed scenario; omit --case')
        result = config_model()
    elif args.scenario == 'session':
        if args.case is not None:
            parser.error('session has one fixed history; omit --case')
        result = session_model()
    elif args.scenario == 'budget':
        if args.case is not None:
            parser.error('budget uses cost flags, not --case')
        result = budget(args.total_ms, args.fixed_ms, args.attempt_ms, args.backoff_ms, args.first_attempt_ms, args.max_attempts)
    elif args.scenario == 'backlog':
        case = args.case or 'normal'
        if case not in ('normal', 'stale', 'deferred-repeat'):
            parser.error('backlog --case must be normal, stale or deferred-repeat')
        result = backlog(case)
    elif args.scenario == 'offsets':
        case = args.case or 'normal'
        if case not in ('normal', 'stale', 'handoff'):
            parser.error('offsets --case must be normal, stale or handoff')
        result = offsets(case)
    elif args.scenario == 'replication':
        if args.case is not None:
            parser.error('replication has one fixed scenario; omit --case')
        result = replication()
    elif args.scenario == 'fencing':
        case = args.case or 'takeover'
        if case not in ('takeover', 'operations'):
            parser.error('fencing --case must be takeover or operations')
        result = fencing(args.variant) if case == 'takeover' else operations()
    elif args.scenario == 'workflow':
        if args.case not in (None, 'all'):
            parser.error('workflow --case must be all')
        result = workflow()
    elif args.scenario == 'inbox':
        cases = ('before-commit', 'after-commit', 'conflict', 'setnx', 'external')
        case = args.case or 'all'
        if case not in cases + ('all',):
            parser.error('inbox --case must be ' + ', '.join(cases) + ', all')
        result = {name: inbox(name) for name in cases} if case == 'all' else inbox(case)
    else:
        cases = ('rollback', 'committed', 'send-before-mark', 'mark-before-send', 'stale-claim', 'batch')
        case = args.case or 'all'
        if case not in cases + ('all',):
            parser.error('outbox --case must be ' + ', '.join(cases) + ', all')
        result = {name: outbox(name) for name in cases} if case == 'all' else outbox(case)
    print(json.dumps(result, ensure_ascii=False, indent=None if args.json else 2))


if __name__ == '__main__':
    main()