check_routed_scatter.py

#!/usr/bin/env python3
"""Finite semantics and abstract service audit for routed bounded scatter.

Host tuples/frozensets/dictionaries are independent semantic models, not native
implementations. The abstract debt is one token per retained target bit plus
one close token; constant-cost capsule/FIFO realization belongs to the proof.
Snapshots are immutable and queries never flush a pending packet.
"""
from dataclasses import dataclass
from itertools import product
import json
import random


def encode(word):
    return ''.join('00' if b == '0' else '01' for b in word) + '11'


def owner(q):
    bits = []
    for i in range(0, len(q) - 1, 2):
        pair = q[i:i + 2]
        if pair == '11':
            return ''.join(bits) if bits and q[i + 2:] else None
        if pair not in ('00', '01'):
            return None
        bits.append(pair[1])
    return None


@dataclass(frozen=True)
class Header:
    active: frozenset = frozenset()
    aggregate: frozenset = frozenset()
    pending: tuple = ()


def prepare(rows, d, c):
    assert rows
    m = len(next(iter(rows)))
    assert m > 0 and all(len(b) == m for b in rows)
    widths = {len(q) for row in rows.values() for q in row}
    assert widths and min(widths) > 0 and max(widths) <= c * m
    primary, packets = set(), {}
    for b, row in rows.items():
        assert row
        foreign = set()
        for q in row:
            assert owner(q) is not None
            (primary if owner(q) == b else foreign).add(q)
        assert len(foreign) <= d
        packets[b] = tuple(sorted(foreign))
    return frozenset(primary), packets, m


def select(header, b, alias, packets, m, d, c):
    # One source's NEXT actual store field services its OLD packet, even when
    # the new alias is absent. Interleaved sources have disjoint pending debt.
    debt = sum(len(q) + 1 for q in header.pending)
    original_debt = debt
    rate = d * (c + 1)
    for _ in b:
        debt = max(0, debt - rate)
    assert len(b) == m and debt == 0
    committed = header.aggregate | frozenset(header.pending)
    active = header.active | ({b} if alias == '0' else set())
    pending = packets[b] if alias == '0' else ()
    return Header(frozenset(active), committed, pending), original_debt


def query(header, primary, q):
    return ((q in primary and owner(q) in header.active)
            or q in header.aggregate or q in header.pending)


def horn(rows, history, source, q):
    clauses = [('01' + encode(b) + '0', '1' + target)
               for b, row in rows.items() for target in row]
    clauses += [('00' + a, '01' + encode(b) + alias)
                for a, b, alias in history]
    reached = {'00' + source}
    while True:
        additions = {y for x, y in clauses if x in reached} - reached
        if not additions:
            return '1' + q in reached
        reached |= additions


def run():
    counts = dict(right_instances=0, histories=0, selections=0, queries=0,
                  horn_checks=0, retained_snapshot_queries=0,
                  abstract_service_tokens=0, stars=0, rejected_gates=0)
    rng = random.Random(226)

    def audit(rows, histories, d=1, c=5):
        primary, packets, m = prepare(rows, d, c)
        counts['right_instances'] += 1
        targets = sorted(set().union(*rows.values()))
        probes = targets + ['0', '10', encode('111') + '0',
                            encode(next(iter(rows))) + '00000']
        for history in histories:
            counts['histories'] += 1
            state, actual, snapshots = {}, {}, []
            prefix = []
            for step, (a, b, alias) in enumerate(history):
                snapshots.append((dict(state), {s: frozenset(t) for s, t in actual.items()}))
                old = state.get(a, Header())
                state[a], tokens = select(old, b, alias, packets, m, d, c)
                counts['abstract_service_tokens'] += tokens
                counts['selections'] += 1
                actual[a] = actual.get(a, frozenset()) | (rows[b] if alias == '0' else frozenset())
                prefix.append((a, b, alias))
                for source in ('0', '10', '111'):
                    for q in probes:
                        answer = query(state.get(source, Header()), primary, q)
                        assert answer == (q in actual.get(source, ()))
                        counts['queries'] += 1
                        if step == 0 or (step == len(history) - 1 and counts['histories'] % 17 == 0):
                            assert answer == horn(rows, prefix, source, q)
                            counts['horn_checks'] += 1
                # Revisit every old snapshot after later selections. No snapshot
                # normalization or mutation is permitted by this model.
                for saved, expected in snapshots:
                    for source, header in saved.items():
                        for q in probes:
                            assert query(header, primary, q) == (q in expected[source])
                            counts['retained_snapshot_queries'] += 1

    stores = ('0', '1')
    targets = tuple(encode(b) + suffix for b in stores for suffix in ('0', '1'))
    options = {}
    for b in stores:
        options[b] = [frozenset(q for q, bit in zip(targets, bits) if bit)
                      for bits in product((0, 1), repeat=4)
                      if any(bits) and sum(bit for q, bit in zip(targets, bits)
                                           if owner(q) != b) <= 1]
    actions = tuple(product(('0', '10'), stores, ('0', '1')))
    for first, second in product(options['0'], options['1']):
        audit({'0': first, '1': second}, product(actions, repeat=2))
    for _ in range(60):
        rows = {b: rng.choice(options[b]) for b in stores}
        audit(rows, [tuple(rng.choice(actions) for _ in range(30))])

    # Unequal target lengths, proper-prefix words and foreign namespaces that
    # do not name any prepared store are all legitimate.
    varied = {
        '0': frozenset((encode('0') + '0', encode('0') + '01', encode('11') + '0')),
        '1': frozenset((encode('1') + '101', encode('0') + '01')),
    }
    for _ in range(30):
        audit(varied, [tuple(rng.choice(actions) for _ in range(30))], c=8)

    # Unbounded primary star degree, foreign scatter one at each leaf, exactly
    # two owners per target; center activation materializes no target packet.
    stars = []
    for leaves in (1, 2, 4, 8, 16, 32, 64):
        m = max(1, leaves.bit_length())
        names = [format(i, f'0{m}b') for i in range(leaves + 1)]
        words = [encode(names[0]) + format(i, f'0{m}b') for i in range(leaves)]
        rows = {names[0]: frozenset(words)}
        rows.update({b: frozenset((q,)) for b, q in zip(names[1:], words)})
        primary, packets, _ = prepare(rows, 1, 5)
        assert not packets[names[0]] and len(primary) == leaves
        assert all(sum(q in row for row in rows.values()) == 2 for q in words)
        history = [('0', names[0], '0')]
        history += [('10', b, '0') for b in names[1:]]
        history += [('0', names[0], '1'), ('10', names[0], '1')]
        audit(rows, [history])
        counts['stars'] += 1
        stars.append(dict(leaves=leaves, center_primary_degree=leaves,
                          center_pending_words=0, max_foreign_scatter=1))
    # Concrete gate failures, rather than silently broadening the contract.
    invalid = [({'0': frozenset((encode('1') + '0', encode('1') + '1'))}, 1, 5),
               ({'0': frozenset((encode('0') + '000',))}, 1, 5),
               ({'0': frozenset((encode('0') + '0',)),
                 '11': frozenset((encode('0') + '1',))}, 1, 5)]
    for args in invalid:
        try:
            prepare(*args)
        except AssertionError:
            counts['rejected_gates'] += 1
        else:
            raise AssertionError('invalid gate accepted')
    return dict(status='pass', scope='finite semantics, immutable snapshots and abstract packet service; not native timing',
                counts=counts, stars=stars)


if __name__ == '__main__':
    print(json.dumps(run(), indent=2))