check_metric_action_receipts.py

#!/usr/bin/env python3
"""Finite semantics for metric copy/reset actions and checkpoint suffixes.
Uses full Python action logs as an oracle, not a native implementation.
"""
from collections import Counter
from itertools import product, permutations
import json, random
from check_bounded_type_dynamics import initial, step, columns, lookup


def identity(k): return tuple(range(k))
def apply(f, b): return tuple(False if i is None else b[i] for i in f)
def compose(f, g):  # apply f, then g
    return tuple(None if i is None else f[i] for i in g)
def total(fs, k):
    out = identity(k)
    for f in fs: out = compose(out, f)
    return out


def action(old, new, v, w):
    k = len(old.predicates)
    label = lookup(old.targets, w)
    src = None if label is None else label - 1
    dst = lookup(new.targets, w) - 1
    bits = [bool(lookup(p, v)) for p in old.predicates]
    active = [j for j in range(k) if old.populations[j] is not None]
    # Determine predicate creation from the same finite metric guards, not sets.
    duplicate = src is not None and bits[src]
    if src is None:
        matches = [j for j in active if old.sizes[j] is not None
                   and old.sizes[j].tail is None and bits[j]]
    else:
        matches = [j for j in active if j != src and old.distances[src][j] is not None
                   and old.distances[src][j].tail is None and bits[j]]
    created = not duplicate and not matches
    f = list(identity(k))
    if created: f[dst] = src
    for j in range(k):
        if new.populations[j] is None: f[j] = None
    return tuple(f), dst


def mask(st, v):
    return tuple(c is not None and bool(lookup(p, v))
                 for p, c in zip(st.predicates, st.populations))


def audit_receipts(st, edges, fs, cache, stats):
    cs = columns(edges)
    for v, (at, b) in cache.items():
        resolved = apply(total(fs[at:], len(b)), b)
        assert resolved == mask(st, v)
        for w, col in cs.items():
            label = lookup(st.targets, w) - 1
            assert resolved[label] == (v in col)
            stats['incidence_comparisons'] += 1
        stats['suffix_resolutions'] += 1
    # Semantic compression at all live checkpoints. Equal boundaries coalesce.
    cuts = sorted({at for at, b in cache.values()} | {len(fs)})
    assert len(cuts) <= len(cache) + 1
    for at, b in cache.values():
        intervals = [total(fs[a:c], len(b)) for a, c in zip(cuts, cuts[1:]) if a >= at]
        assert apply(total(intervals, len(b)), b) == apply(total(fs[at:], len(b)), b)
        stats['compressed_resolutions'] += 1


def run(seq, k, stats):
    st = initial(k); edges = set(); fs = []; cache = {}; saved = []
    for v, w in seq:
        new = step(st, v, w, Counter())
        if new is None:
            stats['width_rejections'] += 1
            break
        f, dst = action(st, new, v, w)
        universe = {x for x, y in edges} | {v, '111111111111111111111111'}
        for x in universe:
            actual = apply(f, mask(st, x))
            if x == v:
                actual = tuple(a or j == dst for j, a in enumerate(actual))
            assert actual == mask(new, x)
            stats['point_override_checks'] += 1
        b = mask(st, v) if v not in cache else apply(total(fs[cache[v][0]:], k), cache[v][1])
        updated = apply(f, b)
        updated = tuple(a or j == dst for j, a in enumerate(updated))
        fs.append(f); cache = dict(cache); cache[v] = (len(fs), updated)
        edges.add((v, w)); st = new
        audit_receipts(st, edges, fs, cache, stats)
        if len(saved) < 3: saved.append((st, edges.copy(), fs[:], cache.copy()))
        stats['prefixes'] += 1
    for args in saved:
        audit_receipts(*args, stats); stats['retained_versions'] += 1
    stats['streams'] += 1


def deferred_checks(stats, rng):
    # A late-key normalizer: skip paired target identifier then 11; emit suffix.
    # Separate per-key logs below are semantic oracles, never native handles.
    def target(tag, z): return ''.join('0' + b for b in tag) + '11' + z
    for groups in range(1, 7):
        stores = {}; receipts = {}; edges = set(); jobs = {}
        for key in range(groups):
            z = format(key + 1, '04b')
            rows = [format(16 * key + r + 1, '08b') for r in range(4)]
            ws = [target(format(t + 1, '03b'), z) for t in range(4)]
            jobs[z] = [(v, w) for v in rows for w in rng.sample(ws, len(ws))]
            stores[z] = (initial(2), [])
        while jobs:
            z = rng.choice(list(jobs)); v, w = jobs[z].pop(0)
            if not jobs[z]: del jobs[z]
            old, fs = stores[z]; new = step(old, v, w, Counter()); assert new is not None
            f, dst = action(old, new, v, w)
            b = (False,) * 2 if v not in receipts else apply(total(fs[receipts[v][1]:], 2), receipts[v][2])
            b = tuple(a or i == dst for i, a in enumerate(apply(f, b)))
            stores[z] = (new, fs + [f]); receipts[v] = (z, len(fs) + 1, b)
            edges.add((v, w)); stats['keyed_action_records'] += 1
        def resolve(job):
            z, at, b = job
            return z, apply(total(stores[z][1][at:], 2), b)
        # One pending finite-mask job per key-length class for each source.
        sources = {}; left = set(); snapshots = []
        for _ in range(100):
            u = rng.choice(['0', '10', '111111111111']); v = rng.choice(list(receipts) + ['absent'])
            committed, pending = sources.get(u, ({}, {}))
            committed, pending = dict(committed), dict(pending)
            if v in receipts:
                job = receipts[v]; bucket = len(job[0]).bit_length() - 1
                if bucket in pending:
                    z, b = resolve(pending[bucket]); prev = committed.get(z, (False,) * 2)
                    committed[z] = tuple(x or y for x, y in zip(prev, b))
                pending[bucket] = job
            sources = dict(sources); sources[u] = (committed, pending)
            left.add((u, v))
            if len(snapshots) < 4: snapshots.append((sources, left.copy()))
        for srcs, rel in snapshots + [(sources, left)]:
            for u in ['0', '10', '111111111111', 'absent']:
                committed, pending = srcs.get(u, ({}, {}))
                for z, (st, fs) in stores.items():
                    b = committed.get(z, (False,) * 2)
                    bucket = len(z).bit_length() - 1
                    if bucket in pending:
                        zp, bp = resolve(pending[bucket])
                        if zp == z: b = tuple(x or y for x, y in zip(b, bp))
                    for w in [t for v, t in edges if t.endswith('11' + z)] + [target('111111', z)]:
                        label = lookup(st.targets, w)
                        got = label is not None and b[label - 1]
                        expect = any(a == u and (v, w) in edges for a, v in rel)
                        assert got == expect
                        stats['deferred_product_queries'] += 1
            stats['deferred_source_versions'] += 1


def main():
    stats = Counter(); rng = random.Random(143)
    for k in range(1, 4):
        maps = list(product([None] + list(range(k)), repeat=k))
        masks = list(product((False, True), repeat=k))
        for f, g in product(maps, repeat=2):
            assert compose(f, g) in maps
            for b in masks:
                assert apply(compose(f, g), b) == apply(g, apply(f, b))
                stats['monoid_comparisons'] += 1
    alphabet = list(product(['0', '10'], ['0', '01', '11']))
    for k in (1, 2, 3):
        for length in range(4):
            for seq in product(alphabet, repeat=length): run(seq, k, stats)
        for seq in permutations(alphabet): run(seq, k, stats)
    names = ['0', '00', '01', '1', '10', '11111111111111110']
    for k in (2, 3, 4):
        for _ in range(200):
            run([(rng.choice(names), rng.choice(names)) for _ in range(25)], k, stats)
    for rows in range(2, 16):
        targets = ['0', '01', '11']
        seq = []
        for r in range(rows):
            rng.shuffle(targets)
            seq.extend((bin(r + 1)[2:], w) for w in targets)
        run(seq, 2, stats)
    deferred_checks(stats, rng)
    # Prefix aggregates cannot recover a suffix, already for {identity, reset}.
    I, Z = (0,), (None,)
    assert total([Z], 1) == total([Z, I], 1) == total([Z, Z], 1)
    assert apply(I, (True,)) != apply(Z, (True,))
    stats['prefix_quotient_counterexamples'] = 1
    # A pending chain for unselected groups can grow while prefix diversity <=2.
    for m in range(2, 40):
        seq = [(bin(i + 1)[2:], '0') for i in range(m)]
        st = initial(2)
        for v, w in seq: st = step(st, v, w, Counter()); assert st is not None
        assert len(columns(seq)['0']) == m
        stats['unbounded_pending_controls'] += 1
    print(json.dumps({'checks': dict(sorted(stats.items())),
          'native_suffix_oracle_implemented': False,
          'finite_timing_evidence': False}, indent=2))

if __name__ == '__main__': main()