check_learned_filler_reuse.py

#!/usr/bin/env python3
"""Finite audit of learned fillers with a fixed occupied-store fibre bound.

Independent direct-owner activation versus immutable normalized bit masks.
Python dictionaries/copies are oracle machinery, not a native implementation.
The service audit counts abstract jobs; it is not a native timing measurement.
Run without arguments for deterministic exhaustive and seeded checks.
"""
from itertools import product, permutations
import json
import random


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


def encode(v):
    return ''.join('0' + b for b in v)


def split(beta):
    """Pair parser independent of any namespace's learned key length."""
    v = ''
    for p in range(0, len(beta) - 1, 2):
        pair = beta[p:p + 2]
        if pair == '11':
            if not v:
                raise ValueError('empty key')
            return v, beta[p + 2:]
        if pair[0] != '0':
            raise ValueError('invalid pair')
        v += pair[1]
    raise ValueError('missing delimiter')


def key(b, n):
    assert len(b) == 2 * n + 1 and b[0] == '1'
    v, _ = split(b[n + 1:])
    return b[1:n + 1] + encode(v) + '11'


class Registry:
    def __init__(self, n, capacity):
        self.n, self.capacity = n, capacity
        self.full, self.count = {}, {}

    def add(self, b):
        # Exact spelling lookup occurs before normalized-key count changes.
        k = key(b, self.n)
        if b in self.full:
            return self.full[b]
        slot = self.count.get(k, 0)
        if slot == self.capacity:
            raise ValueError('fibre overflow')
        self.full[b] = slot
        self.count[k] = slot + 1
        return slot


def update(state, b, alias, registry):
    if b not in registry.full:
        raise ValueError('unknown owner')
    if alias not in (0, 1):
        raise ValueError('invalid alias')
    literal, normalized = state
    if alias:
        return state
    next_literal, next_normalized = set(literal), dict(normalized)
    next_literal.add(b)
    k, slot = key(b, registry.n), registry.full[b]
    next_normalized[k] = next_normalized.get(k, 0) | (1 << slot)
    return frozenset(next_literal), next_normalized


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


def owner(view, z, n):
    alpha, ell, filler, program = view
    if len(z) != n:
        return None
    if z[:n - ell] == '0' * (n - ell):
        t = z[n - ell:]
        return '1' + alpha + encode(mapped(program, t)) + '11' + filler
    return '1' + alpha + z


def query(state, view, z, support, registry):
    # The actual support lane is finite data; no reconstructed owner lookup.
    if z not in support:
        return False
    alpha, ell, _, program = view
    literal, normalized = state
    if z[:registry.n - ell] == '0' * (registry.n - ell):
        t = z[registry.n - ell:]
        k = alpha + encode(mapped(program, t)) + '11'
        return bool(normalized.get(k, 0) & (1 << support[z]))
    return '1' + alpha + z in literal


def schedule(n, ell):
    # One FIFO: initial alpha/setup jobs; two jobs per late transformed bit;
    # the last bit additionally releases the two delimiter jobs.
    pending = n + 2
    trace = []
    for p in range(1, n + 1):
        released = (2 if p > n - ell else 0) + (2 if p == n else 0)
        pending += released
        done = min(4, pending)
        pending -= done
        trace.append((released, done, pending))
    return pending, trace


def main():
    counts = dict(parser=0, registry=0, reads=0, snapshots=0, schedules=0,
                  rejected=0, counterexamples=0)
    for length in range(10):
        for beta in words(length):
            # Independent language-membership oracle enumerates decompositions.
            candidates = [(v, beta[2 * k + 2:])
                          for k in range(1, (length - 2) // 2 + 1)
                          for v in words(k)
                          if beta.startswith(encode(v) + '11')]
            try:
                result = split(beta)
                assert candidates == [result]
            except ValueError:
                assert not candidates
            counts['parser'] += 1

    n = 8
    alpha = '0' * n
    owners = ['1' + alpha + encode('1') + '11' + w for w in words(4)[:3]]
    for order in permutations(owners):
        for capacity in (1, 2, 3):
            r = Registry(n, capacity)
            for j, b in enumerate(order):
                before = (dict(r.full), dict(r.count))
                if j >= capacity:
                    try:
                        r.add(b)
                        raise AssertionError('accepted overflow')
                    except ValueError:
                        assert (r.full, r.count) == before
                        counts['rejected'] += 1
                    continue
                assert r.add(b) == j
                frozen = (dict(r.full), dict(r.count))
                assert r.add(b) == j
                assert (r.full, r.count) == frozen
                counts['registry'] += 2
            # Exhaustive activation and alias combinations within this fibre.
            stored = list(r.full)
            for statuses in product((0, 1), repeat=len(stored)):
                s = (frozenset(), {})
                history = [(s, frozenset())]
                direct = set()
                for b, alias in zip(stored, statuses):
                    s = update(s, b, alias, r)
                    if alias == 0:
                        direct.add(b)
                    history.append((s, frozenset(direct)))
                for old, expected in history:
                    for b in stored:
                        actual = bool(old[1].get(key(b, n), 0) & (1 << r.full[b]))
                        assert actual == (b in expected)
                        counts['reads'] += 1
                    counts['snapshots'] += 1

    rng = random.Random(241)
    for trial in range(24):
        n, capacity = 20, 3
        registry = Registry(n, capacity)
        views, supports = [], []
        # Multiple lengths and arbitrary independent fillers at shared keys.
        for ell in (1, 2, 3, 4):
            program = {p: rng.choice('01') for j in range(1, ell + 1) for p in words(j)}
            # Every view has an actual identity-key seed at t = 0^ell.
            for j in range(1, ell + 1):
                program['0' * j] = '0'
            alpha = format(ell, '020b')
            fillers = set()
            while len(fillers) < capacity:
                fillers.add(''.join(rng.choice('01') for _ in range(n - 2 * ell - 2)))
            for filler in sorted(fillers):
                view = (alpha, ell, filler, program)
                support = {}
                for t in words(ell):
                    z = '0' * (n - ell) + t
                    b = owner(view, z, n)
                    slot = registry.add(b)
                    if rng.randrange(4) or t == '0' * ell:
                        support[z] = slot
                    # Actual ordinary identity rows use the same known owner.
                    support[b[n + 1:]] = slot
                views.append(view)
                supports.append(support)
        names = list(registry.full)
        states = {'s0': (frozenset(), {}), 's1': (frozenset(), {})}
        direct = {'s0': set(), 's1': set()}
        history = []
        for step in range(80):
            source, b, alias = rng.choice(tuple(states)), rng.choice(names), rng.randrange(2)
            states[source] = update(states[source], b, alias, registry)
            if alias == 0:
                direct[source].add(b)
            history.append((dict(states), {s: frozenset(v) for s, v in direct.items()}))
        for states_then, direct_then in history[::5]:
            for source in states_then:
                for view, support in zip(views, supports):
                    candidates = set(support)
                    ell = view[1]
                    candidates.update('0' * (n - ell) + t for t in words(ell))
                    candidates.update(('0', '1' * (n + 1), '1' * n))
                    for z in candidates:
                        expected = z in support and owner(view, z, n) in direct_then[source]
                        assert query(states_then[source], view, z, support, registry) == expected
                        counts['reads'] += 1
            counts['snapshots'] += 1
        try:
            update(states['s0'], 'unknown', 0, registry)
            raise AssertionError('accepted unknown owner')
        except ValueError:
            counts['rejected'] += 1

    for ell in range(1, 65):
        for n in range(4 * ell + 4, 4 * ell + 37):
            pending, trace = schedule(n, ell)
            assert pending == 0 and trace[-1][0] == 4
            assert trace[n - ell - 1][2] == 0
            counts['schedules'] += 1

    # Removing the finite slot produces a false positive across two fillers.
    r = Registry(8, 2)
    for b in owners[:2]:
        r.add(b)
    state = update((frozenset(), {}), owners[0], 0, r)
    assert key(owners[0], 8) == key(owners[1], 8)
    assert state[1][key(owners[1], 8)] != 0
    assert not state[1][key(owners[1], 8)] & (1 << r.full[owners[1]])
    counts['counterexamples'] += 1
    print(json.dumps({'status': 'pass', 'seed': 241, 'checks': counts}, sort_keys=True))


if __name__ == '__main__':
    main()