check_deferred_filler_index.py

#!/usr/bin/env python3
"""Independent finite audit of the deferred shared filler index.

Sets and strings are semantic oracles, not a native graph implementation.
Service tests simulate unit key jobs and their releases, not wall-clock costs.
"""
from itertools import product
import json
import random


def words(n):
    return [''.join(x) for x in product('01', repeat=n)]


def enc(word):
    return ''.join('0' + bit for bit in word) + '11'


def split(beta):
    out = ''
    for i in range(0, len(beta) - 1, 2):
        pair = beta[i:i + 2]
        if pair == '11':
            if not out:
                raise ValueError('empty v')
            return out, beta[i + 2:]
        if pair[0] != '0':
            raise ValueError('bad pair')
        out += pair[1]
    raise ValueError('missing delimiter')


def receipt(b, n):
    assert len(b) == 2 * n + 1 and b[0] == '1'
    v, w = split(b[n + 1:])
    return b[1:n + 1], v, w


def index_key(p):
    alpha, v, w = p
    return enc(w) + alpha + enc(v)


def empty():
    return frozenset(), frozenset(), None


def update(state, b, alias, n, known):
    if b not in known:
        raise ValueError('unknown store')
    if alias not in (0, 1):
        raise ValueError('bad alias')
    literal, installed, pending = state
    # The drain is speculative until alias commits the entire new triple.
    drained = installed | ({index_key(pending)} if pending else set())
    proposed = (literal | {b}, frozenset(drained),
                receipt(b, n) if b[0] == '1' else None)
    return state if alias else proposed


def mapped(program, t):
    return ''.join(program[t[:j]] for j in range(1, len(t) + 1))


def direct_owner(view, z):
    alpha, ell, w, program = view
    n = len(alpha)
    beta = enc(mapped(program, z[-ell:])) + w if z[:n - ell] == '0' * (n - ell) else z
    return '1' + alpha + beta


def read(state, view, z, supported):
    if not supported:
        return False
    literal, installed, pending = state
    alpha, ell, w, program = view
    if z[:len(alpha) - ell] != '0' * (len(alpha) - ell):
        return '1' + alpha + z in literal
    v = mapped(program, z[-ell:])
    # Separate three exact stream equalities; no slot or reconstructed full lookup.
    p_equal = pending is not None and pending[0] == alpha and pending[1] == v and pending[2] == w
    return enc(w) + alpha + enc(v) in installed or p_equal


def invariant(state, known, n):
    literal, installed, pending = state
    for b in known:
        if b[0] == '1':
            p = receipt(b, n)
            assert (b in literal) == (index_key(p) in installed or p == pending)


def audit_injectivity():
    count = 0
    bare_collision = None
    # All valid suffixes, including every v length and arbitrary delimiter-like fillers.
    for n in range(4, 10):
        seen, bare = {}, {}
        for alpha in words(min(n, 3)):
            alpha = alpha.ljust(n, '0')
            for beta in words(n):
                try:
                    v, w = split(beta)
                except ValueError:
                    continue
                p = alpha, v, w
                key = index_key(p)
                assert key not in seen or seen[key] == p
                seen[key] = p
                assert len(key) == 3 * n - 2 * len(v)
                bad_key = alpha + w + enc(v)
                if bad_key in bare and bare[bad_key] != p:
                    bare_collision = (bare[bad_key], p)
                bare[bad_key] = p
                count += 1
    assert bare_collision is not None
    return count, bare_collision


def audit_updates():
    rng = random.Random(242)
    reads = updates = snapshots = 0
    for n in (8, 12, 16):
        views = []
        known = {'0' + '0' * (2 * n)}
        # Shared alpha and same v with many distinct fillers deliberately produce large eta fibres.
        alpha = '01' * (n // 2)
        for ell in range(1, (n - 4) // 4 + 1):
            width = n - 2 * ell - 2
            for k in range(min(20, 2 ** width)):
                w = format(k, f'0{width}b')
                program = {prefix: rng.choice('01') for j in range(1, ell + 1) for prefix in words(j)}
                view = alpha, ell, w, program
                views.append(view)
                for t in words(ell):
                    known.add(direct_owner(view, '0' * (n - ell) + t))
        # Include every sampled ordinary foreign owner only if its suffix parses.
        ordinary = []
        for _ in range(60):
            ell = rng.randint(1, (n - 2) // 2)
            v = ''.join(rng.choice('01') for _ in range(ell))
            w = ''.join(rng.choice('01') for _ in range(n - 2 * ell - 2))
            b = '1' + alpha + enc(v) + w
            known.add(b)
            ordinary.append(b[n + 1:])
        states = {s: empty() for s in range(5)}
        history = []
        population = sorted(known)
        for step in range(180):
            s = rng.randrange(5)
            b = rng.choice(population)
            alias = rng.randrange(2)
            old = states[s]
            states[s] = update(old, b, alias, n, known)
            if alias:
                assert states[s] == old
            history.append((states[s], frozenset(states[s][0])))
            updates += 1
            invariant(states[s], known, n)
            for state, direct in rng.sample(history, min(4, len(history))):
                assert state[0] == direct
                invariant(state, known, n)
                snapshots += 1
                for view in rng.sample(views, min(5, len(views))):
                    ell = view[1]
                    zs = ['0' * (n - ell) + t for t in words(ell)] + rng.sample(ordinary, 3)
                    for z in zs:
                        bq = direct_owner(view, z)
                        for support in (False, bq in known):
                            assert read(state, view, z, support) == (support and bq in direct)
                            reads += 1
        for bad in ('1', 'x' * (2 * n + 1)):
            try:
                update(empty(), bad, 0, n, known)
            except ValueError:
                pass
            else:
                raise AssertionError('unknown accepted')
    return updates, snapshots, reads


def audit_schedule():
    checks = 0
    # Fixed setup/terminal observation are separate fixed physical sub-batches.
    for n in range(8, 161):
        for ell in range(1, (n - 4) // 4 + 1):
            wlen = n - 2 * ell - 2
            # All filler-route jobs available after c; alpha after a; two late code jobs/output.
            backlog = 2 * wlen + 2
            for _ in range(n):
                backlog = max(0, backlog - 4)
            assert backlog == 0
            backlog = n + 2  # inherited alpha route plus bounded routing jobs
            for _ in range(n - ell):
                backlog = max(0, backlog - 4)
            assert backlog == 0
            for j in range(ell):
                backlog += 4 if j == ell - 1 else 2
                backlog = max(0, backlog - 4)
                assert backlog == 0
            # Independent comparator lanes consume the pending words with exact exhaustion.
            assert wlen <= n and n <= 4 * (n - ell)
            checks += 1
        # Worst possible pending v has length one; literal E(b) is 4n+4 symbols.
        backlog = 3 * n - 2
        for _ in range(4 * n + 4):
            backlog = max(0, backlog - 4)
        assert backlog == 0
        checks += 1
    return checks


def audit_unbounded_fibre():
    largest = 0
    for n in (8, 10, 12):
        alpha, v = '0' * n, '0'
        owners = ['1' + alpha + enc(v) + w for w in words(n - 4)]
        assert len({receipt(b, n)[0] + enc(receipt(b, n)[1]) for b in owners}) == 1
        assert len({index_key(receipt(b, n)) for b in owners}) == len(owners)
        known = set(owners)
        state = update(empty(), owners[-1], 0, n, known)
        invariant(state, known, n)
        assert state[1] == frozenset() and state[2] is not None
        # Last activation is readable even without a later source record.
        for b in owners:
            p = receipt(b, n)
            view = p[0], 1, p[2], {'0': '0', '1': '1'}
            assert read(state, view, '0' * n, True) == (b == owners[-1])
        largest = len(owners)
    return largest


def main():
    keys, collision = audit_injectivity()
    updates, snapshots, reads = audit_updates()
    result = {'seed': 242, 'injective_keys_checked': keys,
              'bare_key_collision': collision, 'updates': updates,
              'immutable_snapshot_checks': snapshots, 'semantic_reads': reads,
              'deadline_cases': audit_schedule(),
              'largest_single_fibre': audit_unbounded_fibre(), 'status': 'passed'}
    print(json.dumps(result, indent=2))


if __name__ == '__main__':
    main()