check_banded_context_codes_independent.py

#!/usr/bin/env python3
"""Independent finite relation semantics and virtual-symbol schedule.

Python maps/indexes are semantic oracles, not native code. No primary imports.
"""
from itertools import product
import random
import json


def words(a, b):
    return [''.join(t) for n in range(a, b+1) for t in product('01', repeat=n)]


def code(v, slot, b):
    return ''.join('0'+bit for bit in v) + '11' + format(slot, f'0{b}b')


def produce(records, d=2, s=1):
    b = max(1, d.bit_length())
    A = 2**(s+2) + b + 2
    contexts, literal, tagged = [], {}, {}
    for v, w in records:
        prior = literal.get(v, set())
        if not v or not w or len(prior | {w}) > d:
            return None
        j = next((i for i, (p, _) in enumerate(contexts) if w.startswith(p)), None)
        new_context = j is None
        if new_context:
            if len(w) < len(v)+1:
                return None
            p = w[:len(w)-len(v)-1]
            directory = []
        else:
            p, directory = contexts[j]
        y = w[len(p):]
        if not y or abs(len(y).bit_length()-len(v).bit_length()) > s:
            return None
        h = len(y).bit_length()-1
        # Literal placeholder spine: missing interior class != end of spine.
        directory = directory[:] + [None] * max(0, h+1-len(directory))
        if directory[h] is None:
            assert w not in prior
            directory[h] = code(v, len(prior), b)
        C = directory[h]
        assert len(C) <= A * 2**h
        assert len(C)+len(y) == len(C+y)
        if new_context:
            contexts.append((p, directory))
        else:
            contexts[j] = p, directory
        literal[v] = prior | {w}
        tagged[v] = tagged.get(v, set()) | {C+y}
    codes = [C for _, D in contexts for C in D if C is not None]
    assert len(codes) == len(set(codes))
    for j, (p, _) in enumerate(contexts):
        assert all(not p.startswith(q) for q, _ in contexts[:j])
    return contexts, literal, tagged, A


class Reader:
    def __init__(self, directory, selected, A):
        self.D, self.selected, self.A = directory, selected, A
        self.Q = 2*A+4
        self.residual = ''
        self.h = 0
        self.current = self.ahead = None
        self.work = 0
        self.promotions = 0

    def job(self, h):
        C = self.D[h] if h < len(self.D) else None
        return None if C is None else [C, 0]

    def service(self, job, budget):
        if job is None:
            return
        available = len(job[0])+len(self.residual)-job[1]
        n = min(budget, available)
        job[1] += n
        self.work += n

    def ready(self, job):
        return job is None or job[1] == len(job[0])+len(self.residual)

    def bit(self, bit):
        self.residual += bit
        n = len(self.residual)
        before = self.work
        if n == 1:
            self.current = self.job(0)
            self.service(self.current, self.A+1)
            self.ahead = self.job(1)
        else:
            self.service(self.current, 1)
            self.service(self.ahead, self.Q)
            if n == 2**(self.h+1):
                assert self.ready(self.ahead), (self.h, self.D, self.ahead, n)
                self.current = self.ahead
                self.h += 1
                self.promotions += 1
                self.ahead = self.job(self.h+1)
        assert self.ready(self.current)
        assert self.work-before <= max(self.A+1, self.Q+1)
        assert self.h == n.bit_length()-1

    def answer(self):
        if not self.residual or self.current is None:
            return False
        assert self.ready(self.current)
        return self.current[0]+self.residual in self.selected


def query(contexts, selected, w, A):
    markers = dict(contexts)
    reader = Reader(markers[''], selected, A) if '' in markers else None
    work = changes = promotions = 0
    for n, bit in enumerate(w, 1):
        if reader is not None:
            previous = reader.work
            old_promotions = reader.promotions
            reader.bit(bit)
            work += reader.work-previous
            promotions += reader.promotions-old_promotions
        if w[:n] in markers:
            reader = Reader(markers[w[:n]], selected, A)
            changes += 1
    return False if reader is None else reader.answer(), work, changes, promotions


def main():
    rng = random.Random(212)
    candidates = [(v,w) for v in words(1,2) for w in words(1,3)]
    histories = [[x,y] for x in candidates for y in candidates]
    histories += [[(format(i,'04b'), '0'*(12-i)+'10000') for i in range(13)]]
    # Long creator followed by genuine shorter classes: no creator-code guard.
    histories += [[('0'*20, '000'+'1'*21), ('0','0001'), ('1','00000'), ('10','000101')]]
    # One context with holes and class creation in descending residual length.
    histories += [[('0'*20,'000'+'1'*21), ('1'*8,'000'+'1'*8), ('1','0001')]]
    for _ in range(100):
        histories.append([(rng.choice(words(1,4)), rng.choice(words(1,8))) for _ in range(rng.randrange(1,12))])
    accepted = rejected = versions = checks = switches = promotions = total_work = 0
    maximum_contexts = maximum_classes = 0
    for records in histories:
        result = produce(records)
        if result is None:
            rejected += 1
            continue
        accepted += 1
        contexts, literal, tagged, A = result
        maximum_contexts = max(maximum_contexts, len(contexts))
        maximum_classes = max(maximum_classes, max(sum(C is not None for C in D) for _,D in contexts))
        tests = set(words(1,3)) | set().union(*literal.values()) | {p for p,_ in contexts}
        tests |= {w[:-1] for w in list(tests)} | {w+'0' for w in list(tests)}
        order = list(literal)
        rng.shuffle(order)
        order += order[:2] + ['absent-alias']
        states = [({}, {})]
        for i,v in enumerate(order):
            u = 'a' if i%3 else 'b'
            old_lit,old_tag = states[-1]
            new_lit,new_tag = dict(old_lit),dict(old_tag)
            new_lit[u] = old_lit.get(u,set()) | literal.get(v,set())
            new_tag[u] = old_tag.get(u,set()) | tagged.get(v,set())
            states.append((new_lit,new_tag))
        for lit,tag in states:
            versions += 1
            for u in ('a','b','missing'):
                for w in tests:
                    actual,work,sw,pr = query(contexts,tag.get(u,set()),w,A)
                    assert actual == (w in lit.get(u,set())), (records,u,w,contexts)
                    checks += 1
                    switches += sw
                    promotions += pr
                    total_work += work
    # Worst-size code in every class, and interior holes. Check each deadline.
    schedule_checks = 0
    A = 12
    for mask in range(32):
        directory = [('1'*(A*2**h)) if mask>>h&1 else None for h in range(5)]
        reader = Reader(directory,set(),A)
        for _ in range(63):
            reader.bit('0')
            assert not reader.answer()
            schedule_checks += 1
    assert maximum_contexts >= 13 and maximum_classes >= 3
    print(json.dumps(dict(status='passed',accepted=accepted,rejected=rejected,
        retained_versions=versions,query_checks=checks,context_switches=switches,
        class_promotions=promotions,virtual_work=total_work,
        maximum_contexts=maximum_contexts,maximum_populated_classes=maximum_classes,
        worst_size_scheduler_checks=schedule_checks,
        scope='Finite relation semantics and symbol scheduler; not native implementation'),sort_keys=True))

if __name__ == '__main__':
    main()