check_born_hole_lanes.py

#!/usr/bin/env python3
"""Finite semantic/scheduling audit for name-funded-ball-reuse section 16.

Python integers, tuples and replacement are oracle machinery, not native
primitives. Jobs use the proof's worst-case duration, normalized to a=1.
Each query iteration is one actual target bit; no jump-to-deadline operation.
"""
from dataclasses import dataclass, replace
from itertools import combinations, product
import json
import random


@dataclass(frozen=True)
class Row:
    maximum: int
    holes: tuple[int, ...]

    def contains(self, n):
        return 1 <= n <= self.maximum and n not in self.holes


def cost(n):
    # 1 + ceil(log2(n+1)), for positive integer n.
    return 1 + n.bit_length()


@dataclass(frozen=True)
class Hole:
    position: int
    queue: tuple[Row, ...] = ()
    debt: int = 0
    deleted: bool = False

    def work(self, steps):
        lane = self
        while steps and lane.queue and not lane.deleted:
            debt = lane.debt or cost(lane.position)
            amount = min(debt, steps)
            steps -= amount
            debt -= amount
            if debt:
                lane = replace(lane, debt=debt)
            elif lane.queue[0].contains(lane.position):
                lane = replace(lane, queue=(), debt=0, deleted=True)
            else:
                lane = replace(lane, queue=lane.queue[1:], debt=0)
        return lane

    def append(self, row):
        return self if self.deleted else replace(self, queue=self.queue + (row,))

    def residual(self):
        if not self.queue or self.deleted:
            return 0
        return (self.debt or cost(self.position)) + (len(self.queue)-1)*cost(self.position)


@dataclass(frozen=True)
class Receipt:
    maximum: int = 0
    queue: tuple[Row, ...] = ()
    debt: int = 0
    holes: tuple[Hole, ...] = ()
    seen: frozenset[str] = frozenset()
    selected: tuple[Row, ...] = ()
    births: tuple[tuple[int, int, int], ...] = ()

    def global_work(self, steps):
        s = self
        while steps and s.queue:
            debt = s.debt or cost(s.maximum)
            amount = min(debt, steps)
            steps -= amount
            debt -= amount
            if debt:
                s = replace(s, debt=debt)
                continue
            row, tail = s.queue[0], s.queue[1:]
            holes, births = s.holes, s.births
            if row.maximum > s.maximum:
                for m in row.holes:
                    if m > s.maximum:
                        # Entire tail is new work; no inherited partial debt.
                        holes += (Hole(m, tail),)
                        births += ((s.maximum, m, len(tail)),)
                        assert len(tail)**4 <= 4096*(s.maximum+1)
                        assert len(tail)*cost(m) <= 64*m
                s = replace(s, maximum=row.maximum)
            s = replace(s, queue=tail, debt=0, holes=holes, births=births)
        return s

    def tick(self, global_steps=4, hole_steps=4):
        s = self.global_work(global_steps)
        return replace(s, holes=tuple(h.work(hole_steps) for h in s.holes))

    def select(self, name, row):
        assert name and set(name) <= {'0', '1'}
        s = self
        for _ in name:
            s = s.tick()
        if name in s.seen:
            return s
        s = replace(s, seen=s.seen | {name}, selected=s.selected+(row,))
        if not s.maximum:
            return replace(s, maximum=row.maximum,
                           holes=tuple(Hole(m) for m in row.holes))
        return replace(s, queue=s.queue+(row,),
                       holes=tuple(h.append(row) for h in s.holes))


@dataclass
class Reader:
    length: int
    cursor: int = 0
    debt: int = 0

    def tick(self, newborn=False, position=0):
        if newborn:
            self.debt = position
        elif self.debt:
            self.debt += 1
        else:
            self.cursor += 1
            return
        take = min(4, self.debt)
        self.cursor += take
        self.debt -= take


def query_prefixes(saved, count, stats):
    original = repr(saved)
    s = saved
    envelopes = [Reader(s.maximum)] if s.maximum else []
    readers = [Reader(h.position) for h in s.holes]
    for h in s.holes:
        assert h.residual() <= 96*h.position
    for n in range(1, count+1):
        before_h = len(s.holes)
        # Expose each promotion separately to capture every envelope lane.
        remaining = 128
        while remaining:
            old = s.maximum
            s = s.global_work(1)
            if s.maximum > old:
                envelopes.append(Reader(s.maximum, debt=n))
                stats['query_promotions'] += 1
            remaining -= 1
        for reader in envelopes:
            # Newly created readers carry exactly n debt before first repair.
            fresh = reader.cursor == 0 and reader.debt == n
            reader.tick(newborn=fresh, position=n)
        for idx, h in enumerate(s.holes):
            fresh = idx >= before_h
            if fresh:
                readers.append(Reader(h.position))
                stats['query_hole_births'] += 1
            readers[idx].tick(newborn=fresh, position=n)
            if readers[idx].debt:
                assert n < h.position
        s = replace(s, holes=tuple(h.work(384) for h in s.holes))
        envelope = any(not x.debt and n <= x.length for x in envelopes)
        missing = False
        for h, reader in zip(s.holes, readers):
            if n == h.position:
                assert not reader.debt
                assert h.deleted or not h.queue
                missing |= not h.deleted
        got = envelope and not missing
        want = any(row.contains(n) for row in saved.selected)
        assert got == want, (saved, n, got, want)
        stats['query_prefixes'] += 1
    assert repr(saved) == original  # Every query preserves its starting version.
    stats['retained_query_forks'] += 1


def main():
    stats = dict(query_prefixes=0, retained_query_forks=0,
                 query_promotions=0, query_hole_births=0, selected_prefixes=0,
                 seed_births=0, nonempty_seed_births=0, deadline_checks=0)
    rows = []
    for h in range(1, 7):
        for r in range(min(2, h-1)+1):
            rows += [Row(h, e) for e in combinations(range(1, h), r)]
    # Exhaustive pairs with every order, including equal and different maxima.
    for first, second in product(rows, repeat=2):
        s = Receipt().select('0', first).select('1', second)
        query_prefixes(s, 8, stats)
        stats['selected_prefixes'] += 2
    rng = random.Random(186)
    for trial in range(180):
        maxima = rng.sample(range(2, 65), rng.randint(1, 4))
        dictionary = {}
        s = Receipt()
        retained = []
        for i in range(35):
            name = bin(rng.randrange(1, 24))[2:]
            if name not in dictionary:
                h = rng.choice(maxima)
                e = tuple(sorted(rng.sample(range(1, h), rng.randrange(min(2, h-1)+1))))
                dictionary[name] = Row(h, e)
            retained.append(s)
            s = s.select(name, dictionary[name])
            assert len(s.holes) <= 2*len(maxima)
            stats['selected_prefixes'] += 1
            if i % 7 == 0:
                query_prefixes(s, 66, stats)
        for old in rng.sample(retained, 3):
            query_prefixes(old, 66, stats)
        for _, _, size in s.births:
            stats['seed_births'] += 1
            stats['nonempty_seed_births'] += bool(size)
    # Long numerals force unfinished global jobs and copied queues at birth.
    for exponent in (200, 800, 2000, 4000):
        m = 1 << exponent
        dictionary = [Row(m, (1, 3)), Row(2*m, (m+1, m+3)),
                      Row(m, (2, 3)), Row(2*m, (m+1, m+4))]
        for count in (3, 12, 40):
            s = Receipt().select('0', dictionary[0])
            for i in range(1, count+1):
                s = s.select(bin(i+1)[2:], dictionary[1+(i-1)%3])
            query_prefixes(s, 64, stats)
            final = s.global_work(10**7)
            for _, _, size in final.births:
                stats['seed_births'] += 1
                stats['nonempty_seed_births'] += bool(size)
    for m in range(2, 2000):
        for old in {1, m//2, m-1}:
            tau = (old+3)//4
            assert tau+(m+5)//6 <= m
            assert tau + (max(tau-4, 0)+2)//3 <= old
            stats['deadline_checks'] += 1
    assert stats['query_hole_births'] and stats['nonempty_seed_births']
    print(json.dumps(stats, indent=2))


if __name__ == '__main__':
    main()