check_delayed_suffix_reuse.py

#!/usr/bin/env python3
"""Finite audit of learned fixed-delay suffix programs and paid readout.

Python maps, sets, integers and copied snapshots are independent semantic oracles,
not native storage instructions. The scheduler counts normalized primitive jobs;
the native FIFO/capsule implementation and uniform batch costs require the proof.
"""
from collections import deque
from itertools import combinations, product
import json


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


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


def target(c, a, z):
    return encode(c) + a + z


def parse(q, r):
    pos, c = 0, ''
    while pos + 2 <= len(q):
        pair = q[pos:pos + 2]
        pos += 2
        if pair == '11':
            break
        if pair not in ('00', '01'):
            return None
        c += pair[1]
    else:
        return None
    tail = q[pos:]
    if len(c) != 2*r+1 or not c.startswith('0') or len(tail) != 2*r:
        return None
    if any(b not in '01' for b in tail):
        return None
    return c, tail[:r], tail[r:]


def build(records, r, d):
    """Ingest ordinary (store, target) incidences, including group checks."""
    programs, support, known = {}, {}, set()
    closed, previous = set(), None
    for b, q in records:
        if len(b) != 2*r+1 or any(x not in '01' for x in b):
            raise ValueError('store width')
        if b != previous:
            if b in closed:
                raise ValueError('reopened store')
            if previous is not None:
                closed.add(previous)
            previous = b
        known.add(b)
        parsed = parse(q, r)
        if parsed is None:
            raise ValueError('target shape')
        c, a, z = parsed
        if b[0] == '0':
            if b != c:
                raise ValueError('primary owner')
            support[q] = support.get(q, 0) | 1
            continue
        alpha, beta = b[1:r+1], b[r+1:]
        key = c, a
        if key not in programs:
            programs[key] = alpha, {}
        old_alpha, trie = programs[key]
        if old_alpha != alpha:
            raise ValueError('alpha conflict')
        stream = deque(beta)
        for j in range(1, r+1):
            chunk = ''
            if j == r:
                chunk = ''.join(stream)
                stream.clear()
            elif j > d:
                chunk = stream.popleft()
            prefix = z[:j]
            if prefix in trie and trie[prefix] != chunk:
                raise ValueError('suffix conflict')
            trie[prefix] = chunk
        assert not stream
        support[q] = support.get(q, 0) | 2
    return programs, support, known


def direct_condition(pairs, r, d):
    """Pairwise output-prefix condition, independent of trie insertion/chunks."""
    for z, beta in pairs:
        for zz, bb in pairs:
            for j in range(1, r):
                if z[:j] == zz[:j] and beta[:max(0,j-d)] != bb[:max(0,j-d)]:
                    return False
            if z == zz and beta != bb:
                return False
    return True


def query(built, active, q, r):
    programs, support, _ = built
    parsed = parse(q, r)
    if parsed is None:
        return False
    c, a, z = parsed
    mask = support.get(q, 0)
    primary = bool(mask & 1 and c in active)
    if not mask & 2:
        return primary
    alpha, trie = programs[c, a]
    beta = ''.join(trie[z[:j]] for j in range(1, r+1))
    return primary or '1'+alpha+beta in active


def tick_worker(r, d):
    """Actual arriving chunks; setup, alpha, switch, then queued beta work."""
    queue, phase, alpha_left = deque(), 'setup', r
    consumed, service = [], 0
    capacity = d+4
    beta = tuple(range(1, r+1))
    emitted = 0
    for symbol in range(1, r+1):
        upto = r if symbol == r else max(0, symbol-d)
        queue.extend(beta[emitted:upto])
        emitted = upto
        for _ in range(capacity):
            if phase == 'setup':
                phase = 'alpha'
            elif phase == 'alpha':
                alpha_left -= 1
                if alpha_left == 0:
                    phase = 'switch'
            elif phase == 'switch':
                phase = 'beta'
            elif queue:
                consumed.append(queue.popleft())
            else:
                break
            service += 1
    assert phase == 'beta' and alpha_left == 0 and not queue
    assert tuple(consumed) == beta and service == 2*r+2
    return capacity


def main():
    counts = dict(tables=0, valid_tables=0, rejected_tables=0,
                  producer_orders=0, retained_versions=0, relation_queries=0,
                  schedule_cases=0, suffix_tail_checks=0, rejected_inputs=0)
    for r in (1, 2):
        ws = words(r)
        # All partial functions, then every relation of at most three pairs.
        tables = set()
        for outputs in product((None,)+ws, repeat=len(ws)):
            tables.add(tuple((z,b) for z,b in zip(ws,outputs) if b is not None))
        edges = tuple(product(ws, repeat=2))
        for size in range(4):
            tables.update(combinations(edges, size))
        c, a, alpha = '0'*(2*r+1), '0'*r, '1'*r
        for d in range(4):
            for pairs in sorted(tables):
                counts['tables'] += 1
                expected = direct_condition(pairs,r,d)
                records = sorted(('1'+alpha+b,target(c,a,z)) for z,b in pairs)
                orders = (records, sorted(records, reverse=True))
                for order in orders:
                    counts['producer_orders'] += 1
                    try:
                        built = build(order,r,d)
                    except ValueError:
                        assert not expected
                        continue
                    assert expected
                    if order != orders[0]:
                        continue
                    # Every subset for these small tables, and retained roots.
                    stores = sorted(built[2])
                    snapshots = [frozenset(b for bit,b in zip(bits,stores) if bit)
                                 for bits in product((0,1),repeat=len(stores))]
                    for active in snapshots:
                        counts['retained_versions'] += 1
                        for z in ws:
                            q = target(c,a,z)
                            direct = any(b in active and t == q for b,t in records)
                            assert query(built,active,q,r) == direct
                            counts['relation_queries'] += 1
                    for q in ('','0', target(c,a,ws[0])+'0', target(c,a,ws[0])[:-1]):
                        assert not query(built,frozenset(stores),q,r)
                        counts['relation_queries'] += 1
                counts['valid_tables' if expected else 'rejected_tables'] += 1
    # Growing namespaces and different (c,a) programs, primary support,
    # noninjective output, genuine one-bit delay, and unchanged suffix controls.
    for r in range(1,6):
        ws = words(r)
        namespaces = ('0'*(2*r+1), '0'+'1'*(2*r))
        records = []
        for c in namespaces:
            for a in ws:
                alpha = a[::-1]
                for z in ws:
                    mode = (int(a,2)+int(c[-1])) % 3
                    beta = (z if mode == 0 else '0'*r if mode == 1 else
                            ''.join(z[j:j+2][::-1] for j in range(0,r,2)))
                    q = target(c,a,z)
                    records.append(('1'+alpha+beta,q))
                    if z[0] == '0':
                        records.append((c,q))
        records.sort()
        built = build(records,r,1)
        active = frozenset()
        snapshots = [active]
        for b in sorted(built[2]):
            active = active | {b}
            snapshots.append(active)
        # Sample many retained versions without quadratic large-table cost.
        chosen = snapshots[::max(1,len(snapshots)//9)] + [snapshots[-1]]
        owners = {}
        for b,q in records:
            owners.setdefault(q,set()).add(b)
        for active in chosen:
            counts['retained_versions'] += 1
            for q, bs in owners.items():
                assert query(built,active,q,r) == bool(bs & active)
                counts['relation_queries'] += 1
        assert all(len(bs) <= 2 for bs in owners.values())
        for alpha,trie in built[0].values():
            for prefix, chunk in trie.items():
                assert len(chunk) <= (2 if len(prefix)==r else 1)
                counts['suffix_tail_checks'] += 1
    # Explicit malformed producer cases separate from mathematical prefix test.
    c,a,z='000','0','0'
    q=target(c,a,z)
    bad_cases = [ [('10',q)], [('001',q)], [('100',q+'0')],
                  [('100',q),('110',q)],
                  [('100',q),('101',target(c,a,'1')),('100',q)] ]
    for recs in bad_cases:
        try:
            build(recs,1,0)
        except ValueError:
            counts['rejected_inputs'] += 1
        else:
            raise AssertionError('invalid producer accepted')
    for r in range(1,257):
        for d in range(17):
            capacity = tick_worker(r,d)
            # Independent ordered-job recurrence and suffix inequalities.
            releases = [1]*(r+2) + [min(i+d,r) for i in range(1,r+1)]
            finish = 0
            for release in releases:
                finish = max(finish,capacity*(release-1))+1
            assert finish <= capacity*r
            assert all(len(releases)-j <= capacity*(r-release+1)
                       for j,release in enumerate(releases))
            counts['schedule_cases'] += 1
    print(json.dumps({'status':'pass','scope':'finite semantic and normalized-service audit',
                      'counts':counts},indent=2,sort_keys=True))


if __name__ == '__main__':
    main()