check_release_envelope_reuse.py

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

Adapted from check_delayed_suffix_reuse.py for the unchanged relation wrapper.
Pairwise conditions and maximal-common-prefix schedule oracles are independent of
incremental chunk insertion. Python maps, sets, integers and copied snapshots are
semantic models,
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, C):
    """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)
        tokens = deque([None] * ((C-1)*r))
        for j in range(1, r+1):
            chunk = ''
            for _ in range(C):
                if tokens:
                    tokens.popleft()
                else:
                    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, C):
    """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,r-C*(r-j))] != bb[:max(0,r-C*(r-j))]:
                    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, C):
    """Actual arriving chunks; setup, alpha, switch, then queued beta work."""
    queue, phase, alpha_left = deque(), 'setup', r
    consumed, service = [], 0
    capacity = max(C,4)
    beta = tuple(range(1, r+1))
    emitted = 0
    for symbol in range(1, r+1):
        upto = max(0,r-C*(r-symbol))
        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 common_prefix_length(xs):
    xs = tuple(xs)
    for j in range(len(xs[0])):
        if any(x[j] != xs[0][j] for x in xs[1:]):
            return j
    return len(xs[0])


def earliest_safe_reader(table, z, r, C):
    # Discover the maximal common prefix afresh: an intentionally uncharged
    # semantic oracle, independent of the producer's prescribed chunks.
    served = 0
    for j in range(1,r+1):
        ready = common_prefix_length(b for zz,b in table.items()
                                     if zz[:j] == z[:j])
        served = min(ready,served+C)
    return served == r


def criterion_checks():
    cases = 0
    for r in (1,2,3,4):
        ws = words(r)
        tables = []
        if r <= 2:
            for vals in product((None,)+ws,repeat=len(ws)):
                table = {z:b for z,b in zip(ws,vals) if b is not None}
                if table:
                    tables.append(table)
        else:
            for z,zz in combinations(ws,2):
                for b,bb in product(ws,repeat=2):
                    tables.append({z:b,zz:bb})
        for table in tables:
            for C in range(1,5):
                expected = direct_condition(tuple(table.items()),r,C)
                feasible = all(earliest_safe_reader(table,z,r,C) for z in table)
                assert expected == feasible
                cases += 1
    counter_cases = 0
    for r in range(1,301):
        for C in range(1,25):
            # Explicit allocation and consumption, not packed native integers.
            tokens = deque()
            for _ in range(r):
                for _ in range(C-1):
                    tokens.append(None)
            beta = deque(range(1,r+1))
            emitted = []
            for j in range(1,r+1):
                chunk = []
                for _ in range(C):
                    if tokens:
                        tokens.popleft()
                    else:
                        chunk.append(beta.popleft())
                emitted.extend(chunk)
                assert len(emitted) == max(0,r-C*(r-j))
                assert len(chunk) <= C
                for i in chunk:
                    assert j == ((C-1)*r+i+C-1)//C
                    assert r-i+1 <= C*(r-j+1)
            assert not tokens and not beta and emitted == list(range(1,r+1))
            counter_cases += 1
    for n in range(1,7):
        r = 2*n
        table = {z:z[:n][::-1]+z[n:] for z in words(r)}
        # Grouped LCP oracle avoids a quadratic pairwise full-domain scan.
        for j in range(r+1):
            groups = {}
            for z,b in table.items():
                groups.setdefault(z[:j],[]).append(b)
            assert all(common_prefix_length(bs) >= max(0,r-2*(r-j))
                       for bs in groups.values())
        j = n//2
        outputs = {b[:j] for z,b in table.items() if z[:j] == '0'*j}
        assert len(outputs) == 2**j
    return dict(criterion_cases=cases,counter_cases=counter_cases,
                half_reversal_domains=6)


# Independent bounded-width transition-array and forked-service model.
def branch_envelope(r,C,j):return max(0,r-C*(r-j))
def branch_condition(rows,r,C,W):
    for j in range(r+1):
        groups={}
        for z,b in rows:groups.setdefault(z[:j],set()).add(b[:branch_envelope(r,C,j)])
        if any(len(v)>(1 if j==r else W) for v in groups.values()):return False
    return True

def branch_insert(rows,r,C,W):
    nodes={}
    for z,b in rows:
        parent=0
        for j in range(1,r+1):
            p=z[:j]
            slots=nodes.setdefault(p,[])
            pair=(parent,b[branch_envelope(r,C,j-1):branch_envelope(r,C,j)])
            if pair not in slots:
                if len(slots)>=(1 if j==r else W):raise ValueError
                slots.append(pair)
            parent=slots.index(pair)
    return nodes

def branch_prefixes(nodes,p):
    old=['']
    for j in range(1,len(p)+1):
        old=[old[parent]+chunk for parent,chunk in nodes[p[:j]]]
    return old

# state is (phase, alpha remaining, pending beta, consumed beta, jobs done)
def branch_service(state,chunk,K):
    phase,left,pending,done,jobs=state
    q=deque(pending+chunk)
    for _ in range(K):
        if phase=='setup':phase='alpha'
        elif phase=='alpha':
            left-=1
            if not left:phase='switch'
        elif phase=='switch':phase='beta'
        elif q:done+=q.popleft()
        else:break
        jobs+=1
    return phase,left,''.join(q),done,jobs

def branch_read(nodes,z,r,C):
    states=[('setup',r,'','',0)]
    for j in range(1,r+1):
        if z[:j] not in nodes:return None
        # Every child slot copies its parent version before branch_service, so siblings
        # receive neither another sibling's mutations nor extra branch_service credit.
        states=[branch_service(states[parent],chunk,max(C,4))
                for parent,chunk in nodes[z[:j]]]
    assert len(states)==1
    phase,left,pending,done,jobs=states[0]
    assert phase=='beta' and left==0 and not pending and jobs==2*r+2
    return done

def branch_checks():
    counts=dict(tables=0,orders=0,rejected_orders=0,prefix_sets=0,query_branches=0)
    for r in (1,2,3):
        ws=words(r);edges=tuple(product(ws,repeat=2))
        sizes=range(4) if r<=2 else range(3)
        tables=(rows for size in sizes for rows in combinations(edges,size))
        for rows in tables:
            for C,W in product(range(1,4),range(1,4)):
                counts['tables']+=1;expected=branch_condition(rows,r,C,W)
                for order in (rows,tuple(reversed(rows))):
                    counts['orders']+=1
                    try:nodes=branch_insert(order,r,C,W)
                    except ValueError:
                        assert not expected;counts['rejected_orders']+=1;continue
                    assert expected
                    for p in nodes:
                        actual=set(branch_prefixes(nodes,p))
                        want={b[:branch_envelope(r,C,len(p))] for z,b in rows if z.startswith(p)}
                        assert actual==want and len(actual)==len(nodes[p])
                        counts['prefix_sets']+=1
                    for z in ws:
                        answer=branch_read(nodes,z,r,C)
                        expect={b for zz,b in rows if zz==z}
                        assert ({answer} if answer is not None else set())==expect
                        counts['query_branches']+=1
    # Full-domain half reversal is admitted at W=1,C=2. W=2,C=1
    # additionally permits swapping each adjacent bit pair, rejected at W=1,C=1.
    for r in range(2,11,2):
        ws=words(r)
        rows=tuple((z,''.join(z[j:j+2][::-1] for j in range(0,r,2))) for z in ws)
        assert not branch_condition(rows,r,1,1)
        assert branch_condition(rows,r,1,2)
        nodes=branch_insert(rows,r,1,2)
        for z,b in rows:assert branch_read(nodes,z,r,1)==b
    for r in range(2,11):
        rows=tuple((z,z[-1]+z[:-1]) for z in words(r))
        assert branch_condition(rows,r,1,2)
        nodes=branch_insert(rows,r,1,2)
        for z,b in rows:assert branch_read(nodes,z,r,1)==b
        for C in range(1,r):assert not branch_condition(rows,r,C,1)
    counts['rotation_domains']=9
    counts['swap_domains']=5
    for r in range(2,11):
        rows=tuple((z,str(int(z[0]) ^ int(z[-1]))+z[1:]) for z in words(r))
        assert branch_condition(rows,r,1,2)
        nodes=branch_insert(rows,r,1,2)
        for z,beta in rows:
            assert branch_read(nodes,z,r,1)==beta
        for C in range(1,r):
            assert not branch_condition(rows,r,C,1)
    counts['xor_domains']=9
    return {'branch_'+key:value for key,value in counts.items()}


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 C in range(1,5):
            for pairs in sorted(tables):
                counts['tables'] += 1
                expected = direct_condition(pairs,r,C)
                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,C)
                    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, growing half-block 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
                            z[:r//2][::-1]+z[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,2)
        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
                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,1)
        except ValueError:
            counts['rejected_inputs'] += 1
        else:
            raise AssertionError('invalid producer accepted')
    for r in range(1,257):
        for C in range(1,18):
            capacity = tick_worker(r,C)
            # Independent ordered-job recurrence and suffix inequalities.
            releases = [1]*(r+2) + [((C-1)*r+i+C-1)//C 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
    counts.update(criterion_checks())
    counts.update(branch_checks())
    print(json.dumps({'status':'pass','scope':'finite semantic and normalized-service audit',
                      'counts':counts},indent=2,sort_keys=True))


if __name__ == '__main__':
    main()