check_overlap_owner_directory.py

#!/usr/bin/env python3
"""Finite ownership-directory and restricted cold-reader cone audit.

Host maps, sets, strings and graph traversal are semantic oracles, not a native
implementation. The cone cases explicitly exclude cached terminals, history
shortcuts, prepared source/target aggregates and identity side channels. They
are not native-language lower bounds.
"""
from collections import Counter, deque
from itertools import product
from random import Random
import json


class Rejected(ValueError):
    pass


class Owners:
    def __init__(self, cap):
        self.cap = cap
        self.current = None
        self.closed = set()
        self.local = set()
        self.directory = {}
        self.records = []
        self.snapshots = []
        self.births = self.duplicates = self.owner_name_comparisons = 0

    def insert(self, store, target):
        if store != self.current:
            if store in self.closed:
                raise Rejected('regroup')
            if self.current is not None:
                self.closed.add(self.current)
            self.current, self.local = store, set()
        # This flag is the local producer's support-birth flag. In particular,
        # there is no comparison of store against the earlier owner names.
        birth = target not in self.local
        if birth:
            old = self.directory.get(target, ())
            if len(old) == self.cap:
                raise Rejected('owner cap')
            changed = dict(self.directory)
            changed[target] = old + (store,)
            self.directory = changed
            self.local.add(target)
            self.births += 1
        else:
            self.duplicates += 1
        self.records.append((store, target))
        # Exact-literal oracle is independent of birth-flag append logic.
        expected = {}
        for b, w in self.records:
            expected.setdefault(w, set()).add(b)
        assert {w: set(bs) for w, bs in self.directory.items()} == expected
        assert all(len(bs) == len(set(bs)) <= self.cap for bs in self.directory.values())
        for snapshot, literal in self.snapshots:
            assert {w: set(bs) for w, bs in snapshot.items()} == literal
        if len(self.snapshots) < 8:
            self.snapshots.append((self.directory, expected))


def ball(graph, roots, radius):
    """Ordered rooted local shape, with opaque addresses canonically renamed."""
    ids, depths, queue = {}, {}, deque()
    for root in roots:
        if root not in ids:
            ids[root], depths[root] = len(ids), 0
            queue.append(root)
    result = []
    while queue:
        vertex = queue.popleft()
        label, edges = graph[vertex]
        links = []
        for child in edges:
            if child is None:
                links.append(None)
            elif depths[vertex] == radius:
                # Boundary pointers have no observable contents in this ball.
                links.append('outside')
            else:
                if child not in ids:
                    ids[child], depths[child] = len(ids), depths[vertex] + 1
                    queue.append(child)
                links.append(ids[child])
        result.append((ids[vertex], label, tuple(links)))
    return tuple(result)


def cold_graph(depth, activation, shortcut=False):
    # Two initial roots: source-lane trie and owner spelling. Owner information
    # is available from the start, a stronger grant than late endpoint release.
    graph = {}
    for j in range(depth):
        graph[('lane', j)] = ('branch', (('lane', j + 1), None))
        graph[('word', j)] = ('zero', (('word', j + 1), None))
    graph[('lane', depth)] = (('activation', activation), (None, None))
    graph[('word', depth)] = ('end', (None, None))
    if shortcut:
        graph[('lane', 0)] = ('branch', (('lane', 1), ('lane', depth)))
    return graph, [('lane', 0), ('word', 0)]


def reversal_checks():
    counts = Counter()
    # Enumerate the exact cut geometry independently of a reader strategy.
    for n in range(2, 10):
        for t in range(1, n):
            for prefix_bits in product('01', repeat=n - t):
                prefix = ''.join(prefix_bits)
                owner_keys = [(prefix + ''.join(suffix))[::-1]
                              for suffix in product('01', repeat=t)]
                assert len({key[:t] for key in owner_keys}) == 1 << t
                assert all(key[t:] == prefix[::-1] for key in owner_keys)
                counts['bit_reversal_cut_families'] += 1
    # A generous previous-observation budget still leaves an untouched tail.
    # This checks the numerical/cut premises, not all possible native machines.
    for rate in (1, 2, 4):
        for radius in (1, 2, 3):
            n, overhead = 32, 4
            while True:
                budget = rate * (n + overhead)
                t = budget.bit_length()  # 2^t > budget, including powers of 2.
                if n - t > radius * rate * (t + overhead):
                    break
                n *= 2
            observed_subtrees = set(range(budget))
            chosen = next(i for i in range(1 << t) if i not in observed_subtrees)
            prefix = '01' * ((n - t) // 2) + '0' * ((n - t) % 2)
            suffix = format(chosen, '0%db' % t)[::-1]
            owner = (prefix + suffix)[::-1]
            assert int(owner[:t], 2) == chosen
            assert n - t > radius * rate * (t + overhead)
            counts['uncovered_tail_budget_cases'] += 1
    # Positive control: singleton stores b=reverse(q), aliases 0=present and
    # 1=absent. Each next ordinary E(b) funds one previous pending q insertion
    # into a source-specific target index. No code reconciliation is required.
    for n in range(1, 7):
        rng = Random(700 + n)
        for _ in range(40):
            committed, pending, literal = set(), None, set()
            for step in range(30):
                b = format(rng.randrange(1 << n), '0%db' % n)
                alias = rng.randrange(2)
                if pending is not None:
                    assert len(pending) == n <= 2 * len(b) + 2
                    committed.add(pending)
                pending = b[::-1] if alias == 0 else None
                if pending is not None:
                    literal.add(pending)
                for q_value in range(1 << n):
                    q = format(q_value, '0%db' % n)
                    found = q in committed or q == pending
                    assert found == (q in literal)
                    counts['singleton_materialization_queries'] += 1
            counts['singleton_materialization_histories'] += 1
    return counts


def audit():
    totals = Counter()
    # Every per-store sequence of length 1..3 on a two-name target universe,
    # for three once-only groups. Duplicates model different row incidences
    # as well as repetitions within a row: ownership depends only on support.
    sequences = [seq for length in range(1, 4) for seq in product(('0', '01'), repeat=length)]
    for cap in (1, 2, 3):
        for groups in product(sequences, repeat=3):
            owner = Owners(cap)
            rejected = False
            for b, seq in zip(('0', '1111111111111111', '001'), groups):
                for w in seq:
                    try:
                        owner.insert(b, w)
                    except Rejected as error:
                        assert str(error) == 'owner cap'
                        existing = {old_b for old_b, old_w in owner.records if old_w == w}
                        assert b not in existing and len(existing) == cap
                        totals['cap_rejections'] += 1
                        rejected = True
                        break
                if rejected:
                    break
            totals['histories'] += 1
            totals['records'] += len(owner.records)
            totals['birth_appends'] += owner.births
            totals['duplicates_skipped'] += owner.duplicates
            totals['snapshots_checked'] += len(owner.snapshots)
            assert owner.owner_name_comparisons == 0
    # Larger valid directories with at most two owners, heterogeneously long
    # names, many repeated positive incidences and entirely disjoint targets.
    for seed in range(40):
        rng = Random(seed)
        owner = Owners(2)
        for j in range(30):
            b = format(j, '06b') + '0' * (j % 7)
            targets = [format(j, '08b'), format(100 + j // 2, '08b')]
            for _ in range(12):
                owner.insert(b, rng.choice(targets))
        totals['histories'] += 1
        totals['records'] += len(owner.records)
        totals['birth_appends'] += owner.births
        totals['duplicates_skipped'] += owner.duplicates
        totals['snapshots_checked'] += len(owner.snapshots)
    regroup = Owners(2)
    regroup.insert('0', '0')
    regroup.insert('1', '0')
    try:
        regroup.insert('0', '1')
    except Rejected as error:
        assert str(error) == 'regroup'
        totals['regroup_rejections'] += 1
    else:
        raise AssertionError('reopened group accepted')
    # Pure graph-local condition: every initial visible root/edge is included.
    # An r-radius transition for T steps stays within this generous cone if
    # new pointers can only reuse already locally exposed old vertices.
    for radius in range(1, 5):
        for steps in range(1, 13):
            cone = radius * (steps + 1)
            for excess in (1, 2, 9):
                depth = cone + excess
                zero, roots = cold_graph(depth, 0)
                one, _ = cold_graph(depth, 1)
                assert ball(zero, roots, cone) == ball(one, roots, cone)
                assert ball(zero, roots, depth) != ball(one, roots, depth)
                # The membership answers differ despite equal permitted views.
                assert zero[('lane', depth)][0][1] != one[('lane', depth)][0][1]
                totals['equal_cone_opposite_answer_pairs'] += 1
                # A prepared shortcut intentionally escapes this contract.
                zero_short, roots = cold_graph(depth, 0, shortcut=True)
                one_short, _ = cold_graph(depth, 1, shortcut=True)
                assert ball(zero_short, roots, radius) != ball(one_short, roots, radius)
                totals['shortcut_escape_pairs'] += 1
    totals.update(reversal_checks())
    return dict(totals)


if __name__ == '__main__':
    print(json.dumps({'status': 'pass',
                      'scope': 'finite ownership semantics and restricted graph cones; no native lower bound',
                      'checks': audit()}, indent=2, sort_keys=True))