check_single_hole_reuse.py

#!/usr/bin/env python3
"""Finite audit of the single-hole export and pending union (section 11).

Python integers model unary stack lengths and are not native arithmetic costs.
The counter operations below act on run heads; the proof implements their
length changes by unary stack push/pop. No production code is claimed.
"""
from dataclasses import dataclass, replace
from random import Random
import json
from check_name_funded_ball_reuse import Receipt, step, Q, B, deficit


def encode(n):
    result = []
    for bit in map(int, bin(n)[:1:-1] if n else ''):
        if result and result[-1][0] == bit:
            result[-1] = (bit, result[-1][1]+1)
        else:
            result.append((bit, 1))
    return tuple(result)


def decode(runs):
    out = offset = 0
    for bit, length in runs:
        assert bit in (0, 1) and length > 0
        if bit:
            out += ((1 << length)-1) << offset
        offset += length
    assert not runs or runs[-1][0] == 1
    assert all(x[0] != y[0] for x, y in zip(runs, runs[1:]))
    return out


def inc(r):
    def zero_start(t):
        (_, k), *tail = t
        tail = tuple(tail)
        if k > 1:
            return ((1, 1), (0, k-1)) + tail
        assert tail and tail[0][0] == 1
        return ((1, tail[0][1]+1),) + tail[1:]
    if not r:
        return ((1, 1),)
    if r[0][0] == 0:
        return zero_start(r)
    return ((0, r[0][1]),) + (zero_start(r[1:]) if r[1:] else ((1, 1),))


def dec(r):
    assert r
    def one_start(t):
        (_, k), *tail = t
        tail = tuple(tail)
        if k > 1:
            return ((0, 1), (1, k-1)) + tail
        if not tail:
            return ()
        assert tail[0][0] == 0
        return ((0, tail[0][1]+1),) + tail[1:]
    if r[0][0] == 1:
        return one_start(r)
    head = ((1, r[0][1]),)
    return head if r[1:] == ((1, 1),) else head + one_start(r[1:])


@dataclass(frozen=True)
class Signed:
    sign: int = 0
    magnitude: int = 0
    runs: tuple = ()

    def unit(self, direction):
        if not self.sign or self.sign == direction:
            result = Signed(direction, self.magnitude+1, inc(self.runs))
        elif self.magnitude == 1:
            result = Signed()
        else:
            result = Signed(self.sign, self.magnitude-1, dec(self.runs))
        assert result.sign*result.magnitude == self.sign*self.magnitude+direction
        assert decode(result.runs) == result.magnitude
        return result


@dataclass(frozen=True)
class Row:
    widths: frozenset = frozenset()
    h: int = 0
    gap: int = 0
    total: Signed = Signed()
    delta: Signed = Signed()
    outer: int = 0
    inner: int = 0
    emitted: int = 0
    done: bool = True


def record_width(old, n):
    a = b = old.delta
    candidate_sum = old.total
    gap = old.gap
    outer, inner, emitted, done = old.outer, old.inner, old.emitted, old.done
    worker_steps = 0
    for position in range(1, n+1):
        for _ in range(8):
            if done:
                break
            worker_steps += 1
            if inner:
                inner -= 1
                emitted += 1
                a = a.unit(1); b = b.unit(1)
            elif outer:
                outer -= 1
                inner = outer
            else:
                done = True
        b = b.unit(-1)
        candidate_sum = candidate_sum.unit(1)
        if position > old.h:
            gap += 1
    fresh = n not in old.widths  # inspected only at the record delimiter
    total, delta = (candidate_sum, b) if fresh else (old.total, a)
    if fresh:
        assert gap > 0
        gap -= 1
    if n > old.h:
        outer = inner = n
        emitted = 0
        done = False
        delta = Signed(-1, total.magnitude, total.runs)
    new = Row(old.widths | {n}, max(old.h, n), gap, total, delta,
              outer, inner, emitted, done)
    assert worker_steps <= 8*n
    assert total.magnitude == sum(new.widths)
    assert gap == new.h-len(new.widths)
    assert delta.sign*delta.magnitude == emitted-total.magnitude
    if done:
        assert emitted == new.h*(new.h+1)//2
        missing = set(range(1, new.h+1))-new.widths
        assert delta.sign*delta.magnitude == sum(missing)
    return new


def service(s, full, budget):
    for _ in range(budget):
        if full or (not s.queue and s.active is None):
            break
        active = s.active
        completion = active is not None and not (active.left and active.right)
        unequal = False
        if completion:
            sign = 1 if active.right else -1 if active.left else active.last
            unequal = bool(sign)
        s, _ = step(s)
        if unequal:
            full = True
            s = replace(s, queue=(), active=None)
    return s, full


def select(s, full, name, hole):
    s, full = service(s, full, Q*len(name))
    if name in s.seen:
        return s, full
    s = replace(s, seen=s.seen | {name})
    if full or hole is None:
        return s, full
    if hole == 0:
        return replace(s, queue=(), active=None), True
    if not s.candidate:
        return replace(s, candidate=hole), False
    return replace(s, queue=s.queue+(hole,)), False


def query(s, full, h, selected, count):
    saved_hole = s.candidate
    answers = 0
    for n in range(1, count+1):
        s, full = service(s, full, 4*B)
        selected_any = bool(saved_hole) or full
        answer = selected_any and n <= h and (full or n != saved_hole)
        oracle = any(n <= h and (hole == 0 or n != hole) for hole in selected)
        assert answer == oracle
        if saved_hole and n >= saved_hole:
            assert full or (not s.queue and s.active is None)
        answers += 1
    return answers


def main():
    rng = Random(18311)
    counter_cases = 0
    for n in list(range(8192)) + [2**k+d for k in (64, 256, 1024, 4096)
                                 for d in (-1, 0, 1)]:
        assert decode(inc(encode(n))) == n+1
        if n:
            assert decode(dec(encode(n))) == n-1
        counter_cases += 1
    signed = Signed()
    for _ in range(20000):
        signed = signed.unit(rng.choice((-1, 1)))
    # Arbitrary intermediate gaps, duplicates, order and maximum resets.
    width_records = 0
    for _ in range(200):
        row = Row()
        for _ in range(50):
            row = record_width(row, rng.randrange(1, 30))
            width_records += 1
    literal_records = exported = 0
    for h in range(1, 8):
        for trial in range(6):
            holes = [rng.randrange(h) for _ in range(4)]  # zero means full
            records = [(u, format(i, '0%db' % j))
                       for u, hole in enumerate(holes)
                       for j in range(1, h+1) if j != hole
                       for i in range(2**j)]
            rng.shuffle(records)
            records += records[:20]
            rows = [Row() for _ in range(4)]
            seen = [set() for _ in range(4)]
            counts = [{} for _ in range(4)]
            for u, word in records:
                rows[u] = record_width(rows[u], len(word))
                if word not in seen[u]:
                    seen[u].add(word)
                    n = len(word)
                    counts[u][n] = counts[u].get(n, 0)+1
                row = rows[u]
                if counts[u].get(row.h, 0) == 2**row.h:
                    assert row.done  # forced own-row future service, at first Full
                literal_records += 1
            for u, row in enumerate(rows):
                assert row.done and row.h == h and row.gap <= 1
                assert row.delta.sign*row.delta.magnitude == holes[u]
                exported += 1
    queries = snapshots = 0
    for _ in range(150):
        h = rng.randrange(2, 80)
        table = {bin(i)[2:]: rng.choice([None, 0]+list(range(1, h)))
                 for i in range(1, 20)}
        s, full, chosen = Receipt(), False, set()
        for _ in range(40):
            name = rng.choice(list(table))
            hole = table[name]
            s, full = select(s, full, name, hole)
            if hole is not None:
                chosen.add(hole)
            queries += query(s, full, h, chosen, h+2)
            snapshots += 1
    unfinished = 0
    for exponent in (64, 256, 1024, 4096):
        m = 1 << exponent
        for distinct_last in (False, True):
            s, full = Receipt(), False
            chosen = set()
            for i in range(1, 160):
                hole = m+int(distinct_last and i == 159)
                chosen.add(hole)
                s, full = select(s, full, bin(i)[2:], hole)
            unfinished += bool(s.queue or s.active)
            work, probe, settled = 0, s, full
            while not settled and (probe.queue or probe.active):
                probe, settled = service(probe, settled, 1)
                work += 1
            assert work <= deficit(m) <= B*m
            assert (work+4*B-1)//(4*B) <= m
            queries += query(s, full, m+3, chosen, 1200)
            snapshots += 1
    assert unfinished > 0
    # First moment alone cannot export two arbitrary holes.
    assert sum((1, 4)) == sum((2, 3)) and set((1, 4)) != set((2, 3))
    print(json.dumps(dict(status='passed', unsigned_counter_cases=counter_cases,
                         signed_unit_updates=20000, width_records=width_records,
                         literal_records=literal_records, completed_row_exports=exported,
                         retained_query_receipts=snapshots, query_prefix_answers=queries,
                         unfinished_large_receipts=unfinished), indent=2))


if __name__ == '__main__':
    main()