check_threshold_refinement.py

#!/usr/bin/env python3
"""Finite screening audit, NOT a native priority-queue implementation.

Tuple scans select least-attained jobs for the abstract oracle. Their cost,
integer comparisons, completed-set storage and integer minima are NOT charged
native operations. Integer bit_length models cursor exhaustion; the program
does not execute the native numeral codec. The note isolates the open gates.
"""
from dataclasses import dataclass, replace
from itertools import permutations, product
from random import Random
import json


@dataclass(frozen=True)
class Job:
    name: str
    value: int
    attained: int = 0


@dataclass(frozen=True)
class Screen:
    pending: tuple = ()
    completed: tuple = ()
    steps: int = 0

    def step(self):
        if not self.pending:
            return self
        # Deliberately uncharged priority oracle. No native cost conclusion.
        i = min(range(len(self.pending)), key=lambda j: self.pending[j].attained)
        job = self.pending[i]
        a = job.attained + 1
        rest = self.pending[:i] + self.pending[i+1:]
        if a == job.value.bit_length():
            return Screen(rest, self.completed + (job,), self.steps + 1)
        return Screen(rest + (replace(job, attained=a),), self.completed, self.steps + 1)

    def serve(self, count):
        s = self
        for _ in range(count):
            if not s.pending:
                break
            s = s.step()
        return s

    def release(self, name, value):
        assert value > 0
        assert name not in {j.name for j in self.pending + self.completed}
        return replace(self, pending=self.pending + (Job(name, value),))

    def debt(self, k):
        return sum(max(0, min(k, j.value.bit_length())-j.attained) for j in self.pending)

    def covered(self, k):
        return all(j.attained >= k for j in self.pending)


def bound(k, q):
    return k + sum((1 << d)*(k-q*d) for d in range(1, (k-1)//q+1))


def main():
    rng = Random(192)
    prefixes = inequalities = forks = threshold_checks = 0
    names = ('0', '1', '00', '01')
    for order in permutations(names):
        for values in product((1, 2, 16), repeat=4):
            s = Screen()
            for name, value in zip(order, values):
                s = s.serve(4*len(name)).release(name, value)
                prefixes += 1
                for k in range(1, 14):
                    assert s.debt(k) <= bound(k, 4)
                    inequalities += 1
    for trial in range(90):
        s = Screen()
        names = [bin(i+2)[2:] for i in range(rng.randrange(10, 80))]
        rng.shuffle(names)
        snapshots = []
        for name in names:
            v = rng.choice((2, 3, 17, 1 << rng.randrange(5, 180)))
            s = s.serve(4*len(name)).release(name, v)
            snapshots.append(s)
            prefixes += 1
            for k in range(1, 41):
                assert s.debt(k) <= bound(k, 4)
                inequalities += 1
        for old in rng.sample(snapshots, min(6, len(snapshots))):
            for k in (2, 4, 8, 16, 32):
                before = old
                debt = old.debt(k)
                t = old
                for _ in range(debt):
                    assert not t.covered(k)
                    t = t.step()
                assert t.covered(k)
                assert all(j.value >= 1 << k for j in t.pending)
                # Semantic oracle only: retaining/scanning all completed rows is
                # precisely NOT the claimed native minimum accumulator.
                all_rows = old.pending + old.completed
                for n in (1, 2, (1 << k)-1):
                    assert any(j.value <= n for j in all_rows) == any(j.value <= n for j in t.completed)
                    threshold_checks += 1
                assert old == before
                forks += 1
    # First adversary: huge unresolved values followed by a late 2, then many
    # distinct equal-small names. The screening oracle completes each small row.
    s = Screen().release('0', 1 << 10000)
    s = s.serve(4).release('1', 1 << 10000)
    s = s.serve(8).release('00', 2)
    s = s.serve(s.debt(2))
    assert any(j.name == '00' for j in s.completed)
    for i in range(1024):
        name = '11' + format(i, '010b')
        s = s.serve(4*len(name)).release(name, 2)
    s = s.serve(s.debt(2))
    assert sum(j.value == 2 for j in s.completed) == 1025
    # Second route: newest-first preemption can bury an older small endpoint.
    # q bounds all observations between two short-name releases, including
    # delimiters and routing in the normalized witness. Its q+2-bit numeral is not
    # complete. A subsequent enormous task consumes the whole short query.
    lifo = []
    for q in range(1, 9):
        small = 1 << (q+1)
        small_length = q+2
        query_service = 4*small
        huge_length = query_service + q + 20
        assert q < small_length and query_service < huge_length
        lifo.append({'preemption_gap_budget':q, 'query_length':small,
                     'small_unread_bits':small_length-q,
                     'huge_unread_bits':huge_length-query_service})
    # A deferred comparison against a stale candidate may overwrite a newer
    # minimum: both comparisons are individually correct. This rejects only
    # the unconditional commit rule, not version-aware pending reductions.
    old_min = 10
    delayed_result = min(9, old_min)
    newer_result = min(2, old_min)
    assert delayed_result == 9 and newer_result == 2
    assert delayed_result != min(10, 9, 2)
    # Phase bounds are checked exactly with integer powers, avoiding floats.
    phase_checks = 0
    for k in range(4, 513):
        assert bound(k, 4) <= 36*(1 << (k-4))
        phase_checks += 1
    print(json.dumps({'release_prefixes':prefixes,'debt_inequalities':inequalities,
                      'retained_screening_forks':forks,'threshold_memberships':threshold_checks,
                      'equal_small_completed':1025,'lifo_witnesses':lifo,
                      'phase_inequalities':phase_checks,'stale_commit_witness':[10,9,2,9],
                      'native_scheduler_implemented':False},indent=2))


if __name__ == '__main__':
    main()