check_learned_spine_profiles.py

#!/usr/bin/env python3
"""Finite transition model for theorem 67's learned-spine factorization.

Reuses the maintained theorem-66 producer and consumer transition models.
Python tuples, strings, dictionaries and integers are semantic containers, not
claims of native unit-cost primitives. Native compilation and all-length rates
are proved in learned-spine-profile-reuse.md. The independent checker imports
neither this file nor the inherited model.
"""
from dataclasses import dataclass, replace
from itertools import product
from random import Random
import json
import check_mixed_budget_producer as base


@dataclass(frozen=True)
class Row:
    words: frozenset = frozenset()
    slots: tuple = ()
    profiles: tuple = ()
    pending: tuple = ()
    active: object = None


def step(row, r):
    if row.active is None:
        if not row.pending:
            return row
        lane, n = row.pending[0]
        p = row.profiles[lane]
        assert p.active is None and not p.queue
        profiles = row.profiles[:lane] + (replace(p, queue=(n,)),) + row.profiles[lane+1:]
        return replace(row, profiles=profiles, active=lane, pending=row.pending[1:])
    lane = row.active
    p = base.producer_step(row.profiles[lane], r)
    profiles = row.profiles[:lane] + (p,) + row.profiles[lane+1:]
    return replace(row, profiles=profiles,
                   active=None if p.active is None and not p.queue else lane)


def guide_word(guides, word, K):
    # Actual one-bit-delayed parent construction, no endpoint LCP oracle.
    previous, parent = None, []
    cursors = [0 for _ in guides]
    mismatch = [False for _ in guides]
    extension = [False for _ in guides]
    for bit in word:
        if previous is not None:
            parent.append(previous)
            for lane, guide in enumerate(guides):
                pos = cursors[lane]
                if pos == len(guide):
                    extension[lane] = True
                else:
                    mismatch[lane] |= guide[pos] != previous
                    cursors[lane] += 1
        previous = bit
    assert previous is not None
    parent = ''.join(parent)
    compatible = [i for i, bad in enumerate(mismatch) if not bad]
    if not compatible:
        if len(guides) == K:
            return None
        return guides + (parent,), len(guides), parent
    lane = compatible[0]
    if extension[lane]:
        guides = guides[:lane] + (parent,) + guides[lane+1:]
    return guides, lane, parent


def insert(guides, row, word, K, r, counts):
    for _ in word:
        for _ in range(64):
            before = row
            row = step(row, r)
            counts['producer_service_steps'] += before != row
    result = guide_word(guides, word, K)
    if result is None:
        return None
    guides, choice, parent = result
    slots = dict(row.slots)
    first = parent not in slots
    if first:
        slots[parent] = choice
    else:
        assert slots[parent] == choice  # Stable least-compatible invariant.
    lane = slots[parent]
    p = row.profiles[lane]
    profiles = row.profiles[:lane] + (replace(p, seen=p.seen | {(len(word), 0)}),) + row.profiles[lane+1:]
    row = replace(row, words=row.words | {word}, slots=tuple(slots.items()),
                  profiles=profiles,
                  pending=row.pending + (((lane, len(word)),) if first else ()))
    counts['records'] += 1
    return guides, row


def closed(row):
    return all(w[:-1] + ('1' if w[-1] == '0' else '0') in row.words for w in row.words)


def read_content(guides, word, lane_bits):
    previous, cursors, failed = None, [0]*len(guides), [False]*len(guides)
    answers = []
    for n, bit in enumerate(word, 1):
        if previous is not None:
            for lane, guide in enumerate(guides):
                pos = cursors[lane]
                if pos == len(guide):
                    failed[lane] = True
                else:
                    failed[lane] |= guide[pos] != previous
                    cursors[lane] += 1
        answers.append(any(not failed[lane] and lane_bits[lane][n-1]
                           for lane in range(len(guides))))
        previous = bit
    return answers


def main():
    rng = Random(20467)
    counts = dict(histories=0, records=0, producer_service_steps=0,
                  closure_boundaries=0, budget_passes=0, retained_versions=0,
                  consumer_releases=0, query_prefixes=0, negative_prefixes=0,
                  overflow_cases=0)
    for trial in range(180):
        K = 1 + trial % 3
        r = 0 if trial % 11 == 0 else 4
        # Long shared prefix and distinct final branches; all rows remain sparse.
        stem = ''.join(rng.choice('01') for _ in range(trial % 9))
        branches = ('00', '01', '1')[:K]
        spines = tuple(stem + b + ''.join(rng.choice('01') for _ in range(12)) for b in branches)
        records = []
        row_count = 2 + trial % 7
        for name in range(row_count):
            for _ in range(2 + trial % 5):
                g = rng.choice(spines)
                n = rng.randrange(1, len(g)+2)
                parent = g[:n-1]
                records.extend(((name, parent+'0'), (name, parent+'1')))
        records += rng.sample(records, min(8, len(records)))
        rng.shuffle(records)
        rows = {name: Row(profiles=tuple(base.Producer() for _ in range(K))) for name in range(row_count)}
        guides, saved = (), []
        for name, word in records:
            old = rows[name]
            out = insert(guides, old, word, K, r, counts)
            assert out is not None
            guides, rows[name] = out
            saved.append((old, old.words, old.profiles))
            if closed(rows[name]):
                assert rows[name].active is None and not rows[name].pending
                for p in rows[name].profiles:
                    base.check_quiet(p, r)
                counts['closure_boundaries'] += 1
        assert all(closed(row) for row in rows.values())
        for old, words, profiles in saved:
            assert old.words == words and old.profiles == profiles
            counts['retained_versions'] += 1
        if any(p.bad for row in rows.values() for p in row.profiles):
            counts['histories'] += 1
            continue
        counts['budget_passes'] += 1
        workers = tuple(base.Worker() for _ in range(K))
        selected, literal = set(), set()
        # Deliberately give no pre-query service: this is stronger than paying
        # each real intermediate name. No delimiter drain is permitted.
        for name in list(rows) + list(rows)[:2]:
            if name not in selected:
                workers = tuple(base.release(w, p.trees) for w, p in zip(workers, rows[name].profiles))
                selected.add(name)
                literal.update(rows[name].words)
                counts['consumer_releases'] += 1
            length = max(max(map(len, literal)), max(map(len, guides))) + 2
            lane_bits = tuple(base.query(w, length, 1024*max(1,16*r+16)) for w in workers)
            tests = set(literal)
            tests.update(''.join(rng.choice('01') for _ in range(length)) for _ in range(20))
            tests.update(g+'00' for g in guides)
            for word in tests:
                ans = read_content(guides, word, lane_bits)
                for n, actual in enumerate(ans, 1):
                    expected = word[:n] in literal
                    assert actual == expected
                    counts['query_prefixes'] += 1
                    counts['negative_prefixes'] += not expected
        counts['histories'] += 1
    assert guide_word(('0','1'), '001', 2) is not None
    assert guide_word(('00','01'), '10', 2) is None
    counts['overflow_cases'] += 1
    print(json.dumps(counts, sort_keys=True))


if __name__ == '__main__':
    main()