check_learned_contexts_independent.py

#!/usr/bin/env python3
"""Independent finite semantics audit for learned-context contraction.

Python strings, containers and arithmetic are semantic oracles, not native
operations or timing measurements. No external dependencies.
"""
from collections import deque
from itertools import product
from random import Random


def words(depth):
    return [''.join(bits) for n in range(depth + 1)
            for bits in product('01', repeat=n)]


def rolling_cut(v, w):
    remaining = list(v) + ['!']
    window, prefix = deque(), []
    for bit in w:
        if remaining:
            remaining.pop()
        else:
            prefix.append(window.popleft())
        window.append(bit)
    if remaining:
        return None
    return ''.join(prefix), ''.join(window)


def incremental(records, cap, degree, span):
    contexts, rows = [], {}
    for v, w in records:
        if not v or not w:
            return None
        matches = [j for j, p in enumerate(contexts) if w.startswith(p)]
        if matches:
            j = matches[0]
            y = w[len(contexts[j]):]
        else:
            cut = rolling_cut(v, w)
            if cut is None or len(contexts) == cap:
                return None
            p, y = cut
            j = len(contexts)
            contexts.append(p)
        if not y or abs(len(v).bit_length() - len(y).bit_length()) > span:
            return None
        receipt = rows.setdefault(v, [])
        if (j, y) not in receipt:
            receipt.append((j, y))
            if len(receipt) > degree:
                return None
    return contexts, rows


def reference(records, cap, degree, span):
    """Direct definition with slicing and literal row sets, no receipt state."""
    contexts, rows = [], {}
    for v, w in records:
        if not v or not w:
            return None
        p = next((p for p in contexts if w.startswith(p)), None)
        if p is None:
            if len(contexts) >= cap or len(w) <= len(v):
                return None
            p = w[:len(w) - len(v) - 1]
            contexts.append(p)
        residual = w[len(p):]
        if not residual:
            return None
        lo, hi = sorted((len(v), len(residual)))
        # Compare dyadic classes by repeated halving independently of bit_length.
        a = b = 0
        while lo > 1:
            lo //= 2
            a += 1
        while hi > 1:
            hi //= 2
            b += 1
        if b - a > span:
            return None
        rows.setdefault(v, set()).add(w)
        if len(rows[v]) > degree:
            return None
    return contexts, rows


def main():
    rng = Random(207)
    pool = [('0', '000'), ('1', '100'), ('0', '001'),
            ('00', '1011'), ('1', '00'), ('01', '1111')]
    sequences = [list(xs) for n in range(4) for xs in product(pool, repeat=n)]
    sequences += [[], [('0', '0')], [('', '00')], [('0', '')],
                  [('0', '000'), ('1', '0')],
                  [('0', '000'), ('1', '101'), ('00', '100')],
                  [('0', '0010'), ('1', '00011')],
                  [('1', '00011'), ('0', '0010')]]
    for _ in range(300):
        aliases = ['0', '1', '00', '01']
        seq = [(rng.choice(aliases), rng.choice(words(5)[1:]))
               for _ in range(rng.randrange(1, 10))]
        if seq:
            seq += [seq[0]]  # A later context must not reclassify a duplicate.
        sequences.append(seq)
    checks = accepted = rejected = snapshots = query_checks = 0
    queries = words(6)
    for records in sequences:
        for cap in (1, 2, 3):
            for degree in (1, 2, 3):
                for span in (1, 2):
                    checks += 1
                    got = incremental(records, cap, degree, span)
                    expected = reference(records, cap, degree, span)
                    assert (got is None) == (expected is None), (records, cap, degree, span)
                    if got is None:
                        rejected += 1
                        continue
                    accepted += 1
                    contexts, receipts = got
                    assert contexts == expected[0]
                    for v, entries in receipts.items():
                        assert {contexts[j] + y for j, y in entries} == expected[1][v]
                    names = sorted({v for v, _ in records}) + ['111111']
                    versions = []
                    for n in range(len(names) + 1):
                        selected = frozenset(names[:n])
                        lanes = [set() for _ in contexts]
                        for v in selected:
                            for j, y in receipts.get(v, []):
                                lanes[j].add(y)
                        versions.append((selected, tuple(frozenset(lane) for lane in lanes)))
                    for selected, lanes in reversed(versions):
                        snapshots += 1
                        literal = {w for v, w in records if v in selected}
                        for q in queries:
                            answer = any(q.startswith(p) and q[len(p):] in lane
                                         for p, lane in zip(contexts, lanes))
                            assert answer == (q in literal), (records, selected, q)
                            query_checks += 1
    # A late shorter prefix can overlap an old context but cannot steal old words.
    overlap = [('0', '0000'), ('00', '0110'), ('0', '0000')]
    contexts, rows = incremental(overlap, 2, 2, 1)
    assert contexts == ['00', '0'] and rows['0'] == [(0, '00')]
    # The first-record cut is an order-sensitive checked contract.
    assert incremental([('0', '0010'), ('1', '00011')], 1, 1, 1) is not None
    assert incremental([('1', '00011'), ('0', '0010')], 1, 1, 1) is None
    print({'contract_cases': checks, 'accepted': accepted, 'rejected': rejected,
           'retained_selection_snapshots': snapshots, 'query_checks': query_checks,
           'late_overlap_control': 1, 'order_sensitive_control': 1})


if __name__ == '__main__':
    main()