check_mixed_budget_producer.py

#!/usr/bin/env python3
"""Finite immutable audit of mixed-budget preparation and reuse.

Tuple deques, integer phase counters and Python dictionaries model inherited
native primitives; they are not claimed as constant-time native operations.
Point/merge frames and budget changes are explicit. Direct leaf sets are the
independent semantic oracle. Actual word bits alone provide producer service.
"""
from dataclasses import dataclass, replace
from collections import deque
from random import Random
import json


@dataclass(frozen=True)
class Branch:
    left: object
    right: object


def branch(t):
    return isinstance(t, Branch)


def join(a, b):
    return a if isinstance(a, bool) and a == b else Branch(a, b)


def bits(t, h):
    if not branch(t):
        return (t,) * (1 << h)
    return bits(t.left, h - 1) + bits(t.right, h - 1)


def mixed(t):
    return 0 if not branch(t) else 1 + mixed(t.left) + mixed(t.right)


@dataclass(frozen=True)
class Budget:
    credit: int
    debt: int = 0

    def adjust(self, delta):
        c, d = self.credit, self.debt
        if delta == 1:
            if c: c -= 1
            else: d += 1
        elif delta == -1:
            if d: d -= 1
            else: c += 1
        assert not (c and d) and min(c, d) >= 0
        return Budget(c, d)


@dataclass(frozen=True)
class Point:
    node: object
    route: tuple
    stack: tuple = ()
    phase: str = 'down'
    result: object = None


def point_step(s):
    delta = 0
    if s.phase == 'down':
        if not s.route or s.node is True:
            return replace(s, phase='up', result=True), delta
        direction, rest = s.route[0], s.route[1:]
        a, b = (s.node.left, s.node.right) if branch(s.node) else (False, False)
        frame = (branch(s.node), b if direction == 0 else a, direction)
        return Point(a if direction == 0 else b, rest, s.stack + (frame,)), delta
    if not s.stack:
        return replace(s, phase='done'), delta
    old_mixed, sibling, direction = s.stack[-1]
    result = join(s.result, sibling) if direction == 0 else join(sibling, s.result)
    delta = int(branch(result)) - int(old_mixed)
    return replace(s, stack=s.stack[:-1], result=result), delta


@dataclass(frozen=True)
class Job:
    width: int
    phase: str = 'ruler'
    cursor: int = 0
    focus: int = 0
    point: object = None
    steps: int = 0


@dataclass(frozen=True)
class Producer:
    seen: frozenset = frozenset()
    queue: tuple = ()
    active: object = None
    trees: tuple = ()
    budgets: tuple = ()
    bad: int = 0
    # Budgets include active deltas; trees commit only on job completion.


def producer_step(p, r):
    if p.active is None:
        if not p.queue: return p
        return replace(p, active=Job(p.queue[0]), queue=p.queue[1:])
    s = replace(p.active, steps=p.active.steps + 1)
    j = s.width.bit_length()
    if s.phase == 'ruler':
        if s.cursor < s.width:
            return replace(p, active=replace(s, cursor=s.cursor + 1))
        return replace(p, active=replace(s, phase='focus'))
    if s.phase == 'focus':
        i = s.focus + 1
        trees, budgets = p.trees, p.budgets
        if i > len(trees): trees += (False,)
        if i & (i - 1) == 0 and i.bit_length() > len(budgets):
            # r*i units prepared incrementally on prior/current focus steps.
            budgets += (Budget(r * i),)
        if i == j:
            route = tuple(int(x) for x in bin(s.width)[3:])
            s = replace(s, phase='point', focus=i, point=Point(trees[i-1], route))
        else:
            s = replace(s, focus=i)
        return replace(p, active=s, trees=trees, budgets=budgets)
    if s.phase == 'point':
        point, delta = point_step(s.point)
        b = j.bit_length() - 1
        old = p.budgets[b]
        new = old.adjust(delta)
        budgets = p.budgets[:b] + (new,) + p.budgets[b+1:]
        bad = p.bad + int(bool(new.debt)) - int(bool(old.debt))
        if point.phase == 'done':
            trees = p.trees[:j-1] + (point.result,) + p.trees[j:]
            assert s.steps + 1 <= 16 * (s.width + 1)
            return replace(p, active=None, trees=trees, budgets=budgets, bad=bad)
        return replace(p, active=replace(s, point=point), budgets=budgets, bad=bad)
    raise AssertionError(s.phase)


def word(p, r, n, value, service=64):
    for _ in range(n):
        for _ in range(service): p = producer_step(p, r)
    first_width = not any(k == n for k, _ in p.seen)
    seen = p.seen | {(n, value)}
    p = replace(p, seen=seen, queue=p.queue + ((n,) if first_width else ()))
    return p


def complete(p):
    widths = {n for n, _ in p.seen}
    return all(sum(k == n for k, _ in p.seen) == 1 << n for n in widths)


def check_quiet(p, r):
    assert p.active is None and not p.queue
    widths = {n for n, _ in p.seen}
    for j, t in enumerate(p.trees, 1):
        start = 1 << (j - 1)
        assert bits(t, j-1) == tuple(n in widths for n in range(start, 2*start))
    bad = 0
    for ell, budget in enumerate(p.budgets):
        start = 1 << ell
        m = sum(mixed(t) for t in p.trees[start-1:2*start-1])
        assert budget.credit - budget.debt == r*start - m
        assert bool(budget.debt) == (m > r*start)
        bad += bool(budget.debt)
    assert bad == p.bad


@dataclass(frozen=True)
class Merge:
    a: object
    b: object
    stack: tuple = ()
    phase: str = 'call'
    result: object = None


def merge_step(s):
    if s.phase == 'call':
        if s.b is False: out = s.a
        elif s.b is True or s.a is True: out = True
        elif s.a is False: out = s.b
        else:
            f = ('right', s.a.right, s.b.right)
            return Merge(s.a.left, s.b.left, s.stack + (f,))
        return replace(s, phase='return', result=out)
    if not s.stack: return replace(s, phase='done')
    f = s.stack[-1]
    if f[0] == 'right':
        return Merge(f[1], f[2], s.stack[:-1] + (('join', s.result),))
    return replace(s, stack=s.stack[:-1], result=join(f[1], s.result))


@dataclass(frozen=True)
class Bucket:
    tree: object = False
    jobs: tuple = ()
    active: object = None
    continuation: tuple = ()


@dataclass(frozen=True)
class Worker:
    prefix: tuple = ()
    suffix: tuple = ()


def release(w, row):
    if not row: return w
    seq = w.prefix + w.suffix
    b = seq[0] if seq else Bucket()
    return Worker((), (replace(b, jobs=(row,) + b.jobs),) + seq[1:])


def service(w):
    if not w.suffix: return w
    b, tail = w.suffix[0], w.suffix[1:]
    if b.active is not None:
        active = merge_step(b.active)
        if active.phase != 'done':
            return Worker(w.prefix, (replace(b, active=active),) + tail)
        if b.continuation:
            nxt = tail[0] if tail else Bucket()
            tail = (replace(nxt, jobs=(b.continuation,) + nxt.jobs),) + tail[1:]
        b = replace(b, tree=active.result, active=None, continuation=())
    elif b.jobs:
        row = b.jobs[0]
        b = replace(b, jobs=b.jobs[1:], active=Merge(b.tree, row[0]), continuation=row[1:])
    else:
        return Worker(w.prefix + (b,), tail)
    return Worker(w.prefix, (b,) + tail)


def drain(w, k):
    steps = 0
    while w.suffix and len(w.prefix) < k:
        w = service(w); steps += 1
        assert steps < 100000
    return w, steps


def query(w, length, rate):
    # Explicit local DFS emission; tuple FIFO models the inherited strict FIFO.
    ready, w = w.prefix, Worker((), w.suffix)
    stack, out, answer, j, ended = (), deque(), [], 1, False
    for _ in range(length):
        for _ in range(rate):
            if not w.suffix: break
            w = service(w)
        ready += w.prefix; w = Worker((), w.suffix)
        for _ in range(1024):
            if not stack:
                if ready:
                    stack = ((ready[0].tree, j-1),); ready = ready[1:]; j += 1
                elif not w.suffix: ended = True; break
                else: break
            t, h = stack[-1]; stack = stack[:-1]
            if h == 0: out.append(t is True)
            else:
                a, b = (t.left, t.right) if branch(t) else (t, t)
                stack += ((b, h-1), (a, h-1))
        assert out or ended
        answer.append(out.popleft() if out else False)
    return answer


def main():
    rng = Random(201660)
    counts = dict(point_cases=0, producer_words=0, complete_boundaries=0,
                  retained_versions=0, consumer_releases=0, residual_checks=0,
                  query_prefixes=0, budget_cutoffs=0)
    # Exhaustive small trees, plus deep final-hole collapses.
    for h in range(4):
        width = 1 << h
        for mask in range(1 << width):
            def build(lo, size):
                vals = [(mask >> x) & 1 for x in range(lo, lo+size)]
                if all(vals): return True
                if not any(vals): return False
                return join(build(lo, size//2), build(lo+size//2, size//2))
            old = build(0, width)
            for x in range(width):
                route = tuple((x >> b) & 1 for b in reversed(range(h)))
                s, d = Point(old, route), 0
                while s.phase != 'done': s, change = point_step(s); d += change
                expect = tuple(bool((mask >> i) & 1) or i == x for i in range(width))
                assert bits(s.result, h) == expect
                assert mixed(s.result) == mixed(old) + d
                counts['point_cases'] += 1
    # Real complete word sets, arbitrary row interleaving and duplicates.
    for _ in range(120):
        r = rng.randrange(5); rows = [Producer() for _ in range(3)]
        records = []
        for row in range(3):
            for n in rng.sample(range(1, 7), rng.randrange(1, 5)):
                records += [(row, n, value) for value in range(1 << n)]
        records += rng.choices(records, k=25); rng.shuffle(records)
        retained = []
        for row, n, value in records:
            if rng.randrange(30) == 0: retained.append(rows[row])
            rows[row] = word(rows[row], r, n, value)
            counts['producer_words'] += 1
            if complete(rows[row]):
                check_quiet(rows[row], r); counts['complete_boundaries'] += 1
        for p in rows: check_quiet(p, r)
        for p in retained:
            frozen = repr(p)
            q = p
            for _ in range(1000):
                if q.active is None and not q.queue: break
                q = producer_step(q, r)
            check_quiet(q, r); assert repr(p) == frozen
            counts['retained_versions'] += 1
    # Many runs within one band, with complete producer input conceptually
    # represented here by independent canonical semantic masks.
    pool = []
    for _ in range(100):
        row = []
        for j in range(1, rng.randrange(2, 8)):
            vals = [rng.randrange(2) for _ in range(1 << (j-1))]
            def make(v):
                if all(v): return True
                if not any(v): return False
                m = len(v)//2; return join(make(v[:m]), make(v[m:]))
            row.append(make(vals))
        r = max((sum(mixed(t) for t in row[(1 << ell)-1:(1 << (ell+1))-1])
                 + (1 << ell)-1)//(1 << ell) for ell in range(len(row).bit_length()))
        for k in range(1, len(row)+2):
            assert sum(mixed(t) for t in row[:k]) <= 2*r*k
            counts['budget_cutoffs'] += 1
        pool.append((tuple(row), r))
    for _ in range(60):
        w, selected = Worker(), set()
        r = max(v for _, v in pool); c = 64*(2*r+2)
        for _ in range(12):
            row, _ = rng.choice(pool)
            for _ in range(rng.randrange(60)): w = service(w)
            old = w; new = release(old, row)
            for k in range(1, 8):
                _, before = drain(old, k); _, after = drain(new, k)
                assert after-before <= c*k
                counts['residual_checks'] += 1
            for j, t in enumerate(row, 1):
                start = 1 << (j-1)
                selected |= {start+i for i, v in enumerate(bits(t, j-1)) if v}
            w = new; frozen = repr(w)
            got = query(w, 100, 1024*c)
            assert got == [n in selected for n in range(1, 101)]
            assert repr(w) == frozen
            counts['consumer_releases'] += 1; counts['query_prefixes'] += len(got)
    print(json.dumps(counts, sort_keys=True))


if __name__ == '__main__': main()