check_scanned_omission_reuse.py

#!/usr/bin/env python3
"""Finite audit of scanned omission descriptors; not a native execution test.

Python sets, integer indices, dictionaries and copied snapshots are oracle
machinery. The written construction must separately pay for literal tries,
streaming words, sequential transactions, initialization and strict deadlines.
Extraction below uses only descriptor codes and window order, not birth indices.
"""
from collections import Counter
from copy import deepcopy
from itertools import product
from random import Random
import json


class Store:
    def __init__(self, k, aliases, target_width, alias_width=None):
        self.k, self.K, self.aliases = k, k + 1, aliases
        self.ell = alias_width or max(1, (aliases - 1).bit_length())
        self.n = target_width
        assert self.n >= 1 and aliases <= 1 << self.ell
        # Enough space for all retained references, a replacement descriptor,
        # the global window, one garbage reference and spare candidates.
        coefficient = 8 * (k + 1) + 16
        self.d = (coefficient - 1).bit_length()
        self.m = min(self.ell, self.n) + self.d
        self.capacity = 1 << self.m
        assert self.m <= self.ell + self.d and self.m <= self.n + self.d
        self.births, self.rows, self.literal = [], {}, {}
        self.pool, self.inverse = {}, {}
        self.window, self.garbage = [], None
        self.ever, self.reuses = {}, 0
        self.dirty_recoveries = self.duplicates = self.late_births = 0
        self.peak_references = 0

    def addref(self, target):
        if target in self.pool:
            code, count = self.pool[target]
            self.pool[target] = (code, count + 1)
        else:
            code = next(c for c in range(self.capacity) if c not in self.inverse)
            if code in self.ever and self.ever[code] != target:
                self.reuses += 1
            self.ever[code] = target
            self.inverse[code] = target
            self.pool[target] = (code, 1)
        self.peak_references = max(self.peak_references,
                                   sum(c for _, c in self.pool.values()))
        return code

    def release(self, code):
        target = self.inverse[code]
        _, count = self.pool[target]
        if count == 1:
            del self.pool[target]
            del self.inverse[code]
        else:
            self.pool[target] = (code, count - 1)

    def step(self, alias, target):
        assert 0 <= alias < self.aliases
        assert 0 <= target < 1 << self.n
        if self.garbage is not None:
            self.release(self.garbage)
            self.garbage = None
        old = self.rows.get(alias)
        row = deepcopy(old) if old else dict(R=set(), p=None, i=0,
                                            H=set(), P=set(), refs=[])
        old_eligible = old is not None and old['hc'] is not None and old['pc'] is not None
        # All operations up to incoming-target publication use saved keys.
        if row['p'] is not None:
            p = row['p']
            if p not in row['R']:
                row['R'].add(p)
                if self.births.index(p) < row['i']:
                    row['H'].remove(p)
                else:
                    row['P'].add(p)
        if row['i'] < len(self.births):
            scanned = self.births[row['i']]
            if scanned in row['R']:
                row['P'].remove(scanned)
            else:
                row['H'].add(scanned)
            row['i'] += 1
        newrefs = []
        eligible = len(row['H']) <= self.K and len(row['P']) <= self.K
        for field, coded in [('H', 'hc'), ('P', 'pc')]:
            row[coded] = None
            if eligible:
                row[coded] = tuple(self.addref(t) for t in sorted(row[field]))
                newrefs.extend(row[coded])
        row['boundary'] = None
        if row['i']:
            row['boundary'] = self.addref(self.births[row['i'] - 1])
            newrefs.append(row['boundary'])
        # Add ready replacement references before dropping old descriptor refs.
        if old:
            for code in old['refs']:
                self.release(code)
        # The real incoming target now supplies one pending reference and,
        # exactly on global birth, one window reference. A window eviction is
        # retained as garbage until the next actual record.
        row['p'], row['pending'] = target, self.addref(target)
        newrefs.append(row['pending'])
        row['refs'] = newrefs
        if target not in self.births:
            if self.births:
                self.late_births += 1
            self.births.append(target)
            self.window.append(self.addref(target))
            if len(self.window) > self.K + 1:
                self.garbage = self.window.pop(0)
        if target in self.literal.get(alias, set()):
            self.duplicates += 1
        self.literal.setdefault(alias, set()).add(target)
        self.rows[alias] = row
        new_eligible = row['hc'] is not None and row['pc'] is not None
        if old and not old_eligible and new_eligible:
            self.dirty_recoveries += 1
        self.check()

    def gate(self):
        return bool(self.rows) and min(map(len, self.literal.values())) >= len(self.births) - self.k

    def holes(self, alias):
        """Code-only extraction: no literal keys, indices or cardinalities."""
        row = self.rows[alias]
        assert row['hc'] is not None and row['pc'] is not None
        boundary = row['boundary']
        start = 0 if boundary is None else self.window.index(boundary) + 1
        tail = set(self.window[start:])
        return (set(row['hc']) | (tail - set(row['pc']))) - {row['pending']}

    def check(self):
        support = set(self.births)
        expected = Counter(self.window)
        if self.garbage is not None:
            expected[self.garbage] += 1
        for alias, row in self.rows.items():
            expected.update(row['refs'])
            prefix = set(self.births[:row['i']])
            assert row['H'] == prefix - row['R']
            assert row['P'] == row['R'] - prefix
            assert row['i'] >= len(row['R'])
            assert row['R'] | {row['p']} == self.literal[alias]
            if row['i'] == 0:
                assert row['boundary'] is None
            else:
                assert self.inverse[row['boundary']] == self.births[row['i'] - 1]
            assert self.inverse[row['pending']] == row['p']
            for raw, coded in [('H', 'hc'), ('P', 'pc')]:
                assert (row[coded] is not None) == (len(row['H']) <= self.K and len(row['P']) <= self.K)
                if row[coded] is not None:
                    assert {self.inverse[c] for c in row[coded]} == row[raw]
        assert len(self.inverse) == len(self.pool)
        assert expected == Counter({code: count for code, count in self.pool.values()})
        assert all(self.inverse[code] == target for target, (code, _) in self.pool.items())
        assert len(self.pool) + 1 < self.capacity
        assert [self.inverse[c] for c in self.window] == self.births[-(self.K + 1):]
        assert self.gate() == (bool(self.rows) and all(len(support - X) <= self.k for X in self.literal.values()))
        if self.gate():
            for alias in self.rows:
                decoded = {self.inverse[c] for c in self.holes(alias)}
                assert decoded == support - self.literal[alias]

    def selections(self, names, queries):
        """Three-plus-pending when k=2; arbitrary fixed k otherwise."""
        assert self.gate()
        basis, committed, pending, actual = [], None, None, set()
        checks = extensions = direct_checks = 0
        for alias in names:
            if pending is not None:
                hp = self.holes(pending)
                if committed is None:
                    basis.append(pending)
                    committed = hp
                elif committed - hp:
                    basis.append(pending)
                    committed &= hp
                    extensions += 1
            pending = alias if alias in self.rows else None
            if pending is not None:
                actual |= self.literal[pending]
            assert len(basis) <= self.k + 1
            # Immediate readout retains original literal readers, not a
            # decoded missing-name oracle or an extra completion operation.
            readers = basis + ([] if pending is None else [pending])
            for q in queries:
                found = any(q in self.literal[a] for a in readers)
                assert (not found) == (q not in actual)
                # Independently evaluate the theorem's direct consumer:
                # global literal support minus committed coded omissions,
                # OR the delayed pending row's R and literal p.
                committed_yes = committed is not None and q in self.births and all(
                    q != self.inverse[code] for code in committed)
                pending_yes = pending is not None and (
                    q in self.rows[pending]['R'] or q == self.rows[pending]['p'])
                assert (not (committed_yes or pending_yes)) == (q not in actual)
                direct_checks += 1
                checks += 1
        return checks, extensions, direct_checks


def audit():
    totals = Counter()
    def run(k, aliases, records, rng, alias_width=None, target_width=None):
        n = target_width or max(1, max(t for _, t in records).bit_length())
        store = Store(k, aliases, n, alias_width)
        snapshots = []
        for a, t in records:
            store.step(a, t)
            totals['records'] += 1
            if store.gate():
                totals['accepted_prefixes'] += 1
                names = [rng.randrange(aliases + 1) for _ in range(7)]
                q, e, direct = store.selections(names, range(-1, max(store.births) + 2))
                totals['negative_queries'] += q
                totals['direct_omission_negative_queries'] += direct
                totals['strict_extensions'] += e
            else:
                totals['rejected_prefixes'] += 1
            if len(snapshots) < 8 or rng.randrange(40) == 0:
                snapshots.append(deepcopy(store))
        for old in snapshots:
            old.check()
            if old.gate():
                q, e, direct = old.selections(list(range(aliases + 1)) * 2,
                                      range(-1, max(old.births) + 2))
                totals['negative_queries'] += q
                totals['direct_omission_negative_queries'] += direct
            totals['retained_snapshots'] += 1
        totals['histories'] += 1
        totals['code_reuses_for_different_targets'] += store.reuses
        totals['dirty_descriptor_recoveries'] += store.dirty_recoveries
        totals['duplicate_records'] += store.duplicates
        totals['late_births'] += store.late_births
        totals['peak_references'] = max(totals['peak_references'], store.peak_references)
    # Every history of up to four records on two aliases and three targets.
    alphabet = tuple(product(range(2), range(3)))
    for k in (1, 2, 3):
        rng = Random(1000 + k)
        for length in range(1, 5):
            for records in product(alphabet, repeat=length):
                run(k, 2, records, rng)
        # Tight k+1 singleton basis, including k=3's four-row case.
        tight = Store(k, k + 1, max(1, k.bit_length()))
        for a in range(k + 1):
            tight.step(a, a)
        q, e, direct = tight.selections(list(range(k + 2)), range(-1, k + 2))
        assert e == k
        totals['tight_basis_checks'] += 1
        totals['negative_queries'] += q
        totals['direct_omission_negative_queries'] += direct
        # Both extreme width orientations use the identical ordinary records.
        # Leading zeros preserve these words when their encoded width grows.
        width_records = [(a, t) for a in range(2) for t in range(2)] * 4
        for ell, n in ((1, 1), (1000, 1), (1, 1000), (73, 91)):
            run(k, 2, width_records, rng, alias_width=ell, target_width=n)
            totals['independent_width_histories'] += 1
        # Force large dirty descriptors, late support births, duplicates,
        # descriptor recovery and repeated code reuse.
        records = [(0, 0)] + [(1, t) for t in range(1, 24)]
        records += [(0, 0)] * 24
        records += [(a, t) for a in range(2) for t in range(24)]
        records += [(a, t) for a in range(2) for t in range(24)]
        run(k, 2, records, rng)
        for seed in range(30):
            rng = Random(10000 * k + seed)
            records = [(rng.randrange(8), rng.randrange(24)) for _ in range(100)]
            records += [(a, t) for a in range(8) for t in range(24)]
            run(k, 8, records, rng)
    for field in ('rejected_prefixes', 'strict_extensions', 'code_reuses_for_different_targets',
                  'dirty_descriptor_recoveries', 'duplicate_records', 'late_births'):
        assert totals[field] > 0, field
    return dict(totals)


if __name__ == '__main__':
    print(json.dumps({'status': 'pass', 'scope': 'finite semantic and retained-reference audit; not native execution',
                      'checks': audit()}, indent=2, sort_keys=True))