check_name_funded_ball_reuse.py

#!/usr/bin/env python3
"""Finite audit of name-funded pending maximum receipts (assessment 182).

Tuples/maps/integers are immutable semantic stand-ins, not native operations.
The proof supplies serial dictionaries, strict queues, run counters and rulers.
This model deliberately services old jobs BEFORE releasing the current record.
"""
from dataclasses import dataclass, replace
from itertools import product
from random import Random
import json

A = 4
Q = 4 * A
B = 32 * A
R = 4 * B


@dataclass(frozen=True)
class Active:
    value: int
    left: tuple
    right: tuple
    last: int = 0


@dataclass(frozen=True)
class Receipt:
    seen: frozenset = frozenset()
    candidate: int = 0
    queue: tuple = ()
    active: Active | None = None


def bits(n):
    return tuple(int(x) for x in bin(n)[:1:-1])


def step(s):
    """One cursor/start/finish step; return a strict promotion if one occurs."""
    if s.active is None:
        if not s.queue:
            return s, None
        value, *tail = s.queue
        return replace(s, queue=tuple(tail), active=Active(
            value, bits(s.candidate), bits(value))), None
    j = s.active
    if j.left and j.right:
        x, y = j.left[0], j.right[0]
        last = (y > x) - (y < x) if x != y else j.last
        return replace(s, active=replace(j, left=j.left[1:],
                       right=j.right[1:], last=last)), None
    sign = 1 if j.right else -1 if j.left else j.last
    assert sign == (j.value > s.candidate) - (j.value < s.candidate)
    if sign > 0:
        return replace(s, candidate=j.value, active=None), j.value
    return replace(s, active=None), None


def service(s, budget):
    promotions = []
    for _ in range(budget):
        if s.active is None and not s.queue:
            break
        s, raised = step(s)
        if raised is not None:
            promotions.append(raised)
    return s, promotions


def record(s, name, radius):
    # v bits pay service even if this name is a duplicate or absent on the right.
    s, _ = service(s, Q * len(name))
    if name in s.seen:
        return s
    s = replace(s, seen=s.seen | {name})
    if not radius:
        return s
    if not s.candidate:
        return replace(s, candidate=radius)
    return replace(s, queue=s.queue + (radius,))


def deficit(m):
    c = A * (1 + (m + 1 - 1).bit_length())  # ceil(log2(m+1))
    f = c
    for d in range(1, (c - 1) // Q + 1):
        f += (1 << d) * (c - Q * d)
    assert f <= B * m
    return f


def threshold_steps(s, m):
    """Stop as soon as candidate exceeds m; no cost for later jobs."""
    assert 0 < s.candidate <= m
    work = 0
    while s.candidate <= m and (s.queue or s.active):
        s, _ = step(s)
        work += 1
    return work


def query_prefixes(saved, count, actual_max, k):
    """Run one query stream; check the answer at EVERY possible final prefix."""
    if not saved.candidate:
        assert actual_max == 0
        return count, 0, 0
    s = saved
    # Each lane is [radius, remaining ruler length, debt, predecessor radius].
    lanes = [[s.candidate, s.candidate, 0, 0]]
    raised_count = repairs = 0
    for t in range(1, count + 1):
        for lane in lanes:
            if lane[2]:
                lane[2] += 1
            else:
                lane[1] -= 1
        for _ in range(R):
            if not s.queue and s.active is None:
                break
            before = s.candidate
            s, raised = step(s)
            if raised is not None:
                assert t <= (before + 3) // 4
                lanes.append([raised, raised, t, before])
                raised_count += 1
        for i, lane in enumerate(lanes):
            if lane[2]:
                work = min(4, lane[2])
                lane[2] -= work
                lane[1] -= work
                repairs += work
                if lane[2] == 0:
                    assert t <= lane[3]
            if lane[2] == 0:
                assert lane[1] == lane[0] - t
        assert len(lanes) <= k
        answer = any(debt == 0 and remaining >= 0
                     for _, remaining, debt, _ in lanes)
        assert answer == (t <= actual_max), (saved, t, lanes, actual_max)
        # Any unfinished lane remains shielded by its predecessor's radius.
        assert all(not debt or t <= pred for _, _, debt, pred in lanes)
    return count, raised_count, repairs


def main():
    histories = prefixes = queries = promotions = repairs = thresholds = 0
    # Exhaustive small histories, all prefixes retained as immutable receipts.
    names = ('0', '1', '00', '01')
    for values in ((1, 3, 1, 0), (2, 4, 2, 4), (1, 2, 3, 4)):
        table = dict(zip(names, values))
        k = len(set(values) - {0})
        for n in range(5):
            for seq in product(names, repeat=n):
                histories += 1
                s, selected = Receipt(), set()
                saved = [(s, 0)]
                for name in seq:
                    s = record(s, name, table[name])
                    selected.add(table[name])
                    saved.append((s, max(selected)))
                for old, maximum in saved:
                    prefixes += 1
                    a, b, c = query_prefixes(old, maximum + 2, maximum, k)
                    queries += a; promotions += b; repairs += c
                    if old.candidate:
                        for threshold in {old.candidate, maximum}:
                            assert threshold_steps(old, threshold) <= deficit(threshold)
                            thresholds += 1
    rng = Random(182)
    # Interleaved ordinary sources, duplicates, same-residue radii, 1..5 classes.
    for trial in range(100):
        k = 1 + trial % 5
        radii = sorted(rng.sample(range(1, 70), k))
        table = {bin(i + 2)[2:]: rng.choice(radii) for i in range(25)}
        states = [Receipt() for _ in range(4)]
        selected = [set() for _ in range(4)]
        snapshots = []
        for _ in range(100):
            u = rng.randrange(4); v = rng.choice(list(table))
            states[u] = record(states[u], v, table[v]); selected[u].add(table[v])
            snapshots.append((states[u], max(selected[u])))
        histories += 1
        for s, maximum in snapshots[::5]:
            prefixes += 1
            a, b, c = query_prefixes(s, maximum + 2, maximum, k)
            queries += a; promotions += b; repairs += c
            assert threshold_steps(s, s.candidate) <= deficit(s.candidate)
            thresholds += 1
    # Giant encoded radii create genuinely unfinished comparison queues.
    max_pending = 0
    for exponent in (64, 256, 1024, 4096):
        low = 1 << exponent
        for k in (2, 3, 5):
            s = Receipt(); names = [bin(i)[2:] for i in range(1, 160)]
            values = [low + 2 * (i % k) for i in range(len(names))]
            for name, value in zip(names, values):
                s = record(s, name, value)
            before = s
            max_pending = max(max_pending, len(s.queue) + (s.active is not None))
            # A run of short duplicate selections adds service, never new jobs.
            for _ in range(5):
                s = record(s, names[0], values[0])
            for snapshot in (before, s):
                assert threshold_steps(snapshot, snapshot.candidate) <= deficit(snapshot.candidate)
                thresholds += 1
                a, b, c = query_prefixes(snapshot, 1200, max(values), k)
                queries += a; promotions += b; repairs += c
            histories += 1; prefixes += 2
    # Force late, overlapping promotions behind a long equal-radius queue.
    late_promotions = late_repairs = 0
    for exponent in (64, 256, 1024):
        low = 1 << exponent
        names = [bin(i)[2:] for i in range(1, 33)]
        values = [low] * 28 + [low + 2 * i for i in range(1, 5)]
        s = Receipt()
        for name, value in zip(names, values):
            s = record(s, name, value)
        a, b, c = query_prefixes(s, 700, max(values), 5)
        queries += a; promotions += b; repairs += c
        late_promotions += b; late_repairs += c
        histories += 1; prefixes += 1
    assert late_promotions >= 4 and late_repairs > 2 * late_promotions
    # Without deduplication, one-bit duplicates can add unbounded debt.
    m = 1 << 64
    job_cost = m.bit_length() + 2
    repetitions = (R * m) // (job_cost - Q) + 1
    assert repetitions * (job_cost - Q) > R * m
    # Exact numeric inequalities, independent of worker histories.
    for m in list(range(1, 10001)) + [(1 << j) + d for j in range(1, 1000, 17) for d in (-1, 0, 1)]:
        deficit(m)
        t = (m + 3) // 4
        assert t + (max(0, t - 4) + 2) // 3 <= m
    print(json.dumps(dict(histories=histories, retained_prefixes=prefixes,
          query_prefix_answers=queries, query_promotions=promotions,
          repair_units=repairs, threshold_checks=thresholds,
          maximum_pending_jobs=max_pending, late_promotions=late_promotions,
          late_repair_units=late_repairs, status='passed'), indent=2))


if __name__ == '__main__':
    main()