check_learned_spine_independent.py

#!/usr/bin/env python3
"""Independent finite semantic audit of learned-spine profile reuse.

No imports from the primary model. Strings, sets, exhaustive products and integer
mixed counts below are semantic oracles, not proposed native machine primitives.
The audit checks content/assignment invariants and finite query-prefix answers;
the maintained proof supplies persistence compilation and scheduling bounds.
"""
from itertools import product, permutations, combinations
import json
import random


def words(maximum, empty=False):
    return ([""] if empty else []) + ["".join(bits)
        for width in range(1, maximum + 1)
        for bits in product("01", repeat=width)]


def delayed_parent(word):
    # Separately implement the draft's one-bit delay, including width one.
    pending = None
    parent = []
    for bit in word:
        if pending is not None:
            parent.append(pending)
        pending = bit
    assert pending is not None
    return "".join(parent)


def compatible_stream(parent, guide):
    mismatch = False
    extended = False
    cursor = 0
    for bit in parent:
        if cursor < len(guide):
            mismatch = mismatch or bit != guide[cursor]
            cursor += 1
        else:
            extended = True
    return not mismatch, extended


def advance_cover(guides, parent, capacity):
    updated = list(guides)
    for slot, guide in enumerate(guides):
        match, extended = compatible_stream(parent, guide)
        if match:
            if extended:
                updated[slot] = parent
            return tuple(updated), slot
    if len(guides) == capacity:
        return None, None
    updated.append(parent)
    return tuple(updated), len(guides)


def maximal_parents(parents):
    return {p for p in parents
            if not any(q != p and q.startswith(p) for q in parents)}


def mixed_count(occupied, lo, hi):
    population = sum(n in occupied for n in range(lo, hi + 1))
    if population in (0, hi - lo + 1):
        return 0
    middle = (lo + hi) // 2
    return (1 + mixed_count(occupied, lo, middle)
            + mixed_count(occupied, middle + 1, hi))


def profile_gate(depths, budget):
    counts = {}
    for j in range(1, max(depths, default=0).bit_length() + 1):
        block = 1 << (j.bit_length() - 1)
        counts[block] = counts.get(block, 0) + mixed_count(
            depths, 1 << (j - 1), (1 << j) - 1)
    return all(count <= budget * block for block, count in counts.items())


class History:
    def __init__(self, capacity):
        self.capacity = capacity
        self.guides = ()
        self.literal = {}
        self.assignment = {}
        self.profiles = {}
        self.debt = 0
        self.observed = set()

    def add(self, row, word):
        parent = delayed_parent(word)
        assert parent == word[:-1]
        new_guides, candidate = advance_cover(self.guides, parent, self.capacity)
        if new_guides is None:
            assert len(maximal_parents(self.observed | {parent})) > self.capacity
            return False
        previous_guides = self.guides
        self.guides = new_guides
        assert all(new.startswith(old)
                   for old, new in zip(previous_guides, self.guides))
        self.observed.add(parent)
        assert set(self.guides) == maximal_parents(self.observed)
        present = self.literal.setdefault(row, set())
        pair = (row, parent)
        first = pair not in self.assignment
        if first:
            self.assignment[pair] = candidate
            self.profiles.setdefault((row, candidate), set()).add(len(word))
        else:
            # Stronger invariant: least-compatible choice cannot change.
            assert candidate == self.assignment[pair]
        if word not in present:
            mate = parent + ("1" if word[-1] == "0" else "0")
            self.debt += -1 if mate in present else 1
            present.add(word)
        assert self.debt == sum(
            int((p + "0" in self.literal[r]) != (p + "1" in self.literal[r]))
            for r, p in self.assignment)
        for (r, p), slot in self.assignment.items():
            assert self.guides[slot].startswith(p)
            assert len(p) + 1 in self.profiles[r, slot]
        return True

    def snapshot(self):
        return (self.guides,
                tuple(sorted((r, tuple(sorted(v))) for r, v in self.literal.items())),
                tuple(sorted(self.assignment.items())),
                tuple(sorted((key, tuple(sorted(v))) for key, v in self.profiles.items())),
                self.debt)

    def closed_oracle(self, selected):
        return {p + bit for r, p in self.assignment if r in selected for bit in "01"}

    def query(self, selected, query):
        # Every prefix is checked, even though only final target is output.
        active_depths = [set().union(*(values for (row, lane), values
                          in self.profiles.items() if row in selected and lane == slot))
                         for slot in range(len(self.guides))]
        cursors = [0] * len(self.guides)
        matches = [True] * len(self.guides)
        buffered = None
        results = []
        for width, bit in enumerate(query, 1):
            if buffered is not None:
                for slot, guide in enumerate(self.guides):
                    cursor = cursors[slot]
                    if cursor == len(guide):
                        matches[slot] = False
                    else:
                        matches[slot] = matches[slot] and buffered == guide[cursor]
                        cursors[slot] += 1
            buffered = bit
            results.append(any(matches[slot] and width in active_depths[slot]
                               for slot in range(len(self.guides))))
        return results


def run():
    counts = dict(cover_histories=0, cover_prefixes=0, interleavings=0,
                  record_prefixes=0, query_prefix_checks=0,
                  negative_query_prefixes=0, passing_boundaries=0,
                  retained_snapshots=0, randomized_histories=0,
                  order_sensitive_gates=0)
    parents = words(2, empty=True)
    for size in range(5):
        for sequence in product(parents, repeat=size):
            counts["cover_histories"] += 1
            guide = ()
            seen = set()
            assigned = {}
            for p in sequence:
                before = guide
                guide, slot = advance_cover(guide, p, 4)
                seen.add(p)
                assert set(guide) == maximal_parents(seen)
                assert all(q.startswith(g) for g, q in zip(before, guide))
                assigned.setdefault(p, slot)
                assert assigned[p] == slot
                assert all(guide[s].startswith(q) for q, s in assigned.items())
                counts["cover_prefixes"] += 1

    queries = words(4)
    records = (("a", "0"), ("a", "1"), ("a", "00"),
               ("a", "01"), ("b", "10"), ("b", "11"))

    def inspect(history, saved, query_list):
        snapshot = history.snapshot()
        saved.append(snapshot)
        for size in range(3):
            for subset in combinations(("a", "b"), size):
                oracle = history.closed_oracle(subset)
                if history.debt == 0:
                    literal = set().union(*(history.literal.get(r, set()) for r in subset))
                    assert literal == oracle
                for query in query_list:
                    answers = history.query(subset, query)
                    for n, answer in enumerate(answers, 1):
                        assert answer == (query[:n] in oracle)
                        counts["query_prefix_checks"] += 1
                        counts["negative_query_prefixes"] += int(not answer)
        # Deeply immutable snapshots are independent of later subject mutation.
        assert snapshot == history.snapshot()
        counts["retained_snapshots"] += 1
        counts["passing_boundaries"] += int(history.debt == 0)

    for ordered in permutations(records):
        counts["interleavings"] += 1
        history = History(2)
        saved = []
        for row, word in ordered:
            before = tuple(saved)
            assert history.add(row, word)
            counts["record_prefixes"] += 1
            inspect(history, saved, queries)
            assert tuple(saved[:-1]) == before
        assert history.debt == 0
        frozen = history.snapshot()
        for row, word in reversed(ordered):
            assert history.add(row, word)
        assert history.snapshot() == frozen

    rng = random.Random(2046701)
    for _ in range(240):
        capacity = rng.randint(1, 4)
        guides = ["".join(rng.choice("01") for _ in range(rng.randint(1, 10)))
                  for _ in range(capacity)]
        records = []
        for _ in range(rng.randint(2, 10)):
            guide = rng.choice(guides)
            p = guide[:rng.randrange(len(guide) + 1)]
            row = rng.choice(("a", "b"))
            records.extend(((row, p + "0"), (row, p + "1")))
        rng.shuffle(records)
        records.extend(rng.sample(records, min(3, len(records))))
        history = History(capacity)
        saved = []
        for row, word in records:
            assert history.add(row, word)
            counts["record_prefixes"] += 1
            targeted = [word, word + "0", word + "1",
                        "".join(rng.choice("01") for _ in range(rng.randint(1, 12)))]
            inspect(history, saved, targeted)
        assert history.debt == 0
        counts["randomized_histories"] += 1

    # Same literal rows, different order changes the GLOBAL r=1 gate.
    # In band [4,7], {4,6} has three mixed coordinates, {4,6,7}
    # and {7} have two. The index-block capacity is two.
    shared = (("a", "0000"), ("a", "0001"),
              ("a", "000000"), ("a", "000001"))
    qa = (("a", "0000000"), ("a", "0000001"))
    qb = (("b", "0000010"), ("b", "0000011"))
    first, second = History(2), History(2)
    for record in qa + shared + qb:
        assert first.add(*record)
    for record in qb + shared + qa:
        assert second.add(*record)
    assert first.literal == second.literal
    assert first.debt == second.debt == 0
    assert all(profile_gate(d, 1) for d in first.profiles.values())
    assert not all(profile_gate(d, 1) for d in second.profiles.values())
    counts["order_sensitive_gates"] = 1
    # Overflow cannot subsequently be repaired by a later parent.
    h = History(1)
    assert h.add("a", "00")
    assert not h.add("a", "10")
    assert len(maximal_parents({"0", "1", "0000"})) == 2
    print(json.dumps({"status": "passed", **counts}, sort_keys=True))


if __name__ == "__main__":
    run()