check_pending_substitution.py

"""Finite semantic/scheduling corroboration for assessment126.

Capsule tries are immutable. Python tuples model the established strict FIFO;
Python dict directories and integer clocks are semantic models, NOT native
instructions or timing evidence. Installer progress and reader replay are
counted explicitly in trie-bit/closure macrosteps. Independent set joins are
used only by the oracle. The written proof realizes directories/clocks locally.
"""
from dataclasses import dataclass
from itertools import product
import json
import random
from check_frozen_row_composition import B, view, close, lookup


def step(builder, bit):
    frames, root = builder
    value, left, right = view(root)
    return frames + ((value, int(bit), left if bit == '1' else right),), (right if bit == '1' else left)


def walk(root, word):
    builder = (), root
    for bit in word:
        builder = step(builder, bit)
    return builder


def cls(word):
    assert word
    return len(word).bit_length() - 1


@dataclass(frozen=True)
class RightRow:
    seen: object = B
    targets: tuple = ()  # (immutable spelling tuple, finite class offset)


def add_right(root, v, w, d, s):
    outer = walk(root, v)
    row = view(outer[1])[0] or RightRow()
    inner = walk(row.seen, w)
    offset = cls(w) - cls(v)
    if abs(offset) > s:
        return None
    if view(inner[1])[0]:
        return root
    if len(row.targets) == d:
        return None
    new = RightRow(close(inner, True), row.targets + ((tuple(w), offset),))
    return close(outer, new)


@dataclass(frozen=True)
class Bucket:
    committed: object = B
    pending: tuple = ()


class Installer:
    """One macrostep per inserted key bit or capsule closure."""
    def __init__(self, bucket):
        self.root = bucket.committed
        self.jobs = bucket.pending
        self.job = self.pos = self.steps = 0
        self.builder = (), self.root

    @property
    def done(self):
        return self.job == len(self.jobs)

    def tick(self):
        if self.done:
            return
        word = self.jobs[self.job]
        if self.pos < len(word):
            self.builder = step(self.builder, word[self.pos])
            self.pos += 1
        else:
            self.root = close(self.builder, True)
            self.job += 1
            self.pos = 0
            self.builder = (), self.root
        self.steps += 1

    def finish(self):
        while not self.done:
            self.tick()
        return Bucket(self.root, ())


STATS = {'serviced_batches': 0, 'discarded_partial_jobs': 0,
         'pending_answers': 0, 'reader_promotions': 0, 'max_pending': 0}


def service_for_name(index, word, d, s):
    """No new target pointers are passed here: they are revealed afterwards."""
    current = dict(index)
    start_cost = 0
    for j in range(s + 1):
        job = Installer(current.get(j, Bucket()))
        current[j] = job.finish()
        start_cost += job.steps
        STATS['serviced_batches'] += bool(job.steps)
    assert start_cost <= (s + 1) * d * (2 ** (s + 1))
    j, boundary = 0, 2
    next_bucket = s + 1
    worker = Installer(current.get(next_bucket, Bucket()))
    rate = d * (2 ** (s + 2) + 1)
    for length in range(2, len(word) + 1):
        for _ in range(rate):
            worker.tick()
        if length == boundary:
            assert worker.done, (length, next_bucket, worker.steps)
            current[next_bucket] = Bucket(worker.root, ())
            STATS['serviced_batches'] += bool(worker.steps)
            j += 1
            boundary *= 2
            next_bucket = j + s + 1
            worker = Installer(current.get(next_bucket, Bucket()))
    assert j == cls(word)
    if worker.steps and not worker.done:
        STATS['discarded_partial_jobs'] += 1
    # Even if speculative work finished early, no unpromoted bucket is changed.
    for bucket in range(max(0, j - s), j + s + 1):
        assert not current.get(bucket, Bucket()).pending
    return current, j


def add_left(root, frozen, u, v, d, s):
    outer = walk(root, u)
    index = view(outer[1])[0] or {}
    serviced, j = service_for_name(index, v, d, s)
    receipt = lookup(frozen, v) or RightRow()
    groups = {}
    for word, offset in receipt.targets:
        assert cls(word) == j + offset
        assert abs(offset) <= s
        groups.setdefault(j + offset, []).append(word)
    for bucket, words in groups.items():
        old = serviced.get(bucket, Bucket())
        assert not old.pending
        assert len(words) <= d
        serviced[bucket] = Bucket(old.committed, tuple(words))
        STATS['max_pending'] = max(STATS['max_pending'], len(words))
    return close(outer, serviced)


class Reader:
    def __init__(self, bucket):
        self.cursor = bucket.committed
        self.pending = [tuple(word) for word in bucket.pending]
        self.viable = [True] * len(self.pending)
        self.consumed = 0

    def feed(self, bit):
        _, left, right = view(self.cursor)
        self.cursor = right if bit == '1' else left
        for i, word in enumerate(self.pending):
            if self.viable[i]:
                if not word or word[0] != bit:
                    self.viable[i] = False
                else:
                    self.pending[i] = word[1:]
        self.consumed += 1

    def contains(self):
        pending = any(ok and not word for ok, word in zip(self.viable, self.pending))
        STATS['pending_answers'] += bool(pending and not view(self.cursor)[0])
        return bool(view(self.cursor)[0] or pending)


def index_contains(index, word):
    current = Reader(index.get(0, Bucket()))
    current.feed(word[0])
    prefix = (word[0],)
    queue = prefix
    ahead = Reader(index.get(1, Bucket()))
    j, boundary = 0, 2
    for length, bit in enumerate(word[1:], 2):
        prefix += (bit,)
        current.feed(bit)
        queue += (bit,)
        for _ in range(2):
            if queue:
                ahead.feed(queue[0])
                queue = queue[1:]
        if length == boundary:
            assert not queue and ahead.consumed == length
            current = ahead
            STATS['reader_promotions'] += 1
            j += 1
            boundary *= 2
            ahead = Reader(index.get(j + 1, Bucket()))
            queue = prefix
    assert current.consumed == len(word) and j == cls(word)
    return current.contains()


def answer(root, a, b):
    return not index_contains(lookup(root, a) or {}, b)


def main():
    names = ['0', '1']
    pairs = list(product(names, repeat=2))
    rels = [{edge for i, edge in enumerate(pairs) if mask >> i & 1} for mask in range(16)]
    exhaustive = 0
    for srel in rels:
        frozen = B
        for v, w in sorted(srel):
            frozen = add_right(frozen, v, w, 2, 0)
            assert frozen is not None
        for r in rels:
            root = B
            for u, v in sorted(r):
                root = add_left(root, frozen, u, v, 2, 0)
            expected = {(a, b) for a, v in r for z, b in srel if z == v}
            for a, b in product(names + ['00'], repeat=2):
                assert answer(root, a, b) == ((a, b) not in expected)
                exhaustive += 1
    rng = random.Random(126)
    words = ['0', '1', '00', '01', '10', '000', '111']
    words += ['0' * (n - 1) + '1' for n in [4, 7, 8, 15, 16, 31, 32, 65]]
    sources = ['0', '00', '1' * 97]
    snapshots = queries = checked = 0
    for trial in range(18):
        d, span = 1 + trial % 3, trial % 2
        frozen, srel = B, set()
        for _ in range(30):
            v, w = rng.choice(words), rng.choice(words)
            new = add_right(frozen, v, w, d, span)
            row = {b for a, b in srel if a == v}
            valid = abs(cls(v) - cls(w)) <= span and (w in row or len(row) < d)
            assert (new is not None) == valid
            checked += 1
            if valid:
                frozen = new
                srel.add((v, w))
        # Explicit long-then-short source variations with arbitrary row reuse.
        root, r, saved = B, set(), []
        for _ in range(28):
            u, v = rng.choice(sources), rng.choice(words)
            root = add_left(root, frozen, u, v, d, span)
            r.add((u, v))
            expected = {(a, b) for a, v in r for z, b in srel if z == v}
            saved.append((root, expected.copy()))
            snapshots += 1
        for old, expected in saved:
            for a, b in product(sources + ['101'], words):
                assert answer(old, a, b) == ((a, b) not in expected)
                queries += 1
    # Force a speculative long bucket installation to be discarded, while
    # short queries still see the exact earlier pending word.
    frozen = B
    for v in ['0' * 7, '0' * 4, '0' * 15, '0']:
        frozen = add_right(frozen, v, v[::-1], 1, 0)
    root = B
    for v in ['0' * 15, '0' * 7, '0' * 4, '0']:
        root = add_left(root, frozen, '0', v, 1, 0)
        assert not answer(root, '0', v[::-1])
    # Arbitrary equal-length permutations, not fixed streaming maps.
    permutation_queries = 0
    for n in range(1, 6):
        names = [''.join(x) for x in product('01', repeat=n)]
        targets = names.copy()
        rng.shuffle(targets)
        frozen, root = B, B
        for v, w in zip(names, targets):
            frozen = add_right(frozen, v, w, 1, 0)
        chosen = set(names[::2])
        for v in chosen:
            root = add_left(root, frozen, '0', v, 1, 0)
        expected = {w for v, w in zip(names, targets) if v in chosen}
        for w in names:
            assert answer(root, '0', w) == (w not in expected)
            permutation_queries += 1
    assert STATS['pending_answers'] and STATS['discarded_partial_jobs']
    print(json.dumps(dict(status='pass', exhaustive_queries=exhaustive,
        right_condition_checks=checked, retained_versions=snapshots,
        retained_queries=queries, arbitrary_permutation_queries=permutation_queries,
        scheduling=STATS, native_timing_evidence=False)))


if __name__ == '__main__':
    main()