check_two_maximum_suffix.py

#!/usr/bin/env python3
"""Two-maximum suffix pipeline: finite semantics, debt and persistence audit.

The minimum worker is the explicit immutable sparse-age implementation.
Classification observes linked digits only. Tuple FIFO/dictionaries and reader
lag arithmetic model inherited primitives; they are not native timing evidence.
"""
from dataclasses import dataclass, replace
from random import Random
import json
from check_sparse_age_minimum import numeral, decode, release, step, idle, inspect


@dataclass(frozen=True)
class Row:
    lower: object
    upper: object


@dataclass(frozen=True)
class Compare:
    row: Row
    left: object
    right: object


@dataclass(frozen=True)
class Receipt:
    all_min: object = None
    high_min: object = None
    high_upper: object = None
    queue: tuple = ()
    active: object = None


def enqueue(s, row):
    return replace(s, all_min=release(s.all_min, row.lower),
                   queue=s.queue+(row,))


def macro(s, low):
    # One fixed macro services all three lanes, even after one becomes idle.
    s = replace(s, all_min=step(s.all_min), high_min=step(s.high_min))
    c = s.active
    if c is None:
        if not s.queue:
            return s
        row = s.queue[0]
        return replace(s, queue=s.queue[1:], active=Compare(row, low, row.upper))
    a, b = c.left, c.right
    unequal = (a is None) != (b is None)
    unequal = unequal or (a is not None and b is not None and a.bit != b.bit)
    done = unequal or (a is None and b is None)
    if done:
        return replace(s, active=None,
                       high_min=release(s.high_min, c.row.lower) if unequal else s.high_min,
                       high_upper=(c.row.upper if s.high_upper is None else s.high_upper)
                       if unequal else s.high_upper)
    return replace(s, active=Compare(c.row, a.tail, b.tail))


def quiet(s):
    return not s.queue and s.active is None and idle(s.high_min)


def serve(s, low, count):
    for _ in range(count):
        if quiet(s) and idle(s.all_min):
            break
        s = macro(s, low)
    return s


def answer_prefixes(s, L, rows, stop, qquery=1024):
    # Arithmetic lag is an audit model of unary debt replay, not a worker op.
    low = numeral(L)
    all_birth = high_birth = None
    all_pos = high_pos = 0
    all_value = high_value = upper = None
    checks = 0
    for n in range(1, stop+1):
        s = serve(s, low, qquery)
        if all_birth is None and idle(s.all_min) and s.all_min is not None:
            all_birth, all_value = n, decode(s.all_min.best)
        if high_birth is None and quiet(s) and s.high_min is not None:
            high_birth, high_value, upper = n, decode(s.high_min.best), decode(s.high_upper)
        if all_birth is not None:
            all_pos = min(n, all_pos+4)
        if high_birth is not None:
            high_pos = min(n, high_pos+4)
        if n <= L:
            got = all_birth is not None and all_pos == n and all_value <= n
        else:
            got = high_birth is not None and high_pos == n and high_value <= n <= upper
        expected = any(b <= n <= h for b, h in rows)
        assert got == expected, (L, rows, n, got, expected, all_birth, high_birth)
        checks += 1
    return checks


def minimum_export(old_hist, old_max, new_max):
    """Independent semantic model of the target-driven histogram walk."""
    assert new_max > old_max
    first = None
    for depth in range(1, new_max+1):
        count = old_hist.get(depth, 0)
        if depth == old_max:
            count -= 1
        if first is None and count > 0:
            first = depth
    return new_max if first is None else first


def main():
    rng = Random(194)
    exports = prefixes = forks = completed = debt_checks = 0
    maxima, histogram = {}, {}
    for _ in range(3000):
        name, n = rng.randrange(80), rng.randrange(1, 160)
        old = maxima.get(name, 0)
        if n > old:
            exported = minimum_export(histogram, old, n)
            if old:
                histogram[old] -= 1
            histogram[n] = histogram.get(n, 0)+1
            maxima[name] = n
            assert exported == min(maxima.values())
            exports += 1
    # Fresh small classes, removal of the sole minimum, and distant successors.
    for hist, old, new in [({2:1,100:1},2,7),({2:2,100:1},2,101),({99:1},0,1)]:
        got = minimum_export(hist,old,new)
        updated=dict(hist)
        if old: updated[old]-=1
        updated[new]=updated.get(new,0)+1
        assert got==min(x for x,c in updated.items() if c)
        exports += 1
    for trial in range(90):
        L = rng.choice((1,2,3,7,15,31,63,127))
        U = L+rng.randrange(1,100)
        low, s, selected, saved = numeral(L), Receipt(), [], []
        for i in range(45):
            name = bin(i+1)[2:]
            s = serve(s, low, 64*len(name))
            h = rng.choice((L,U))
            b = rng.randrange(1,h+1)
            s = enqueue(s,Row(numeral(b),numeral(h)))
            selected.append((b,h))
            if i%9==0:
                saved.append((s,tuple(selected)))
        for old, rows in saved:
            before = repr(old)
            prefixes += answer_prefixes(old,L,rows,U+2)
            assert repr(old)==before
            done = serve(old,low,20000)
            assert quiet(done) and idle(done.all_min)
            assert decode(done.all_min.best)==min(b for b,h in rows)
            high=[b for b,h in rows if h>L]
            assert (done.high_min is None)==(not high)
            if high: assert decode(done.high_min.best)==min(high)
            forks+=1; completed+=1
    # Large classifier debt cannot obstruct the early all-row minimum.
    L=(1<<1500)+1; U=L+2
    rows=[(1,U),(2,L),(3,U)]
    s=Receipt()
    for b,h in rows: s=enqueue(s,Row(numeral(b),numeral(h)))
    prefixes+=answer_prefixes(s,L,rows,32)
    # Debt stress uses much smaller name service and counts total pipeline work.
    for L in (1,3,31,255):
        low=numeral(L); s=Receipt(); rows=[]
        for i in range(1,100):
            d=len(bin(i)[2:]); s=serve(s,low,2*d)
            h=rng.choice((L,L+17)); b=rng.randrange(1,h+1)
            rows.append((b,h)); s=enqueue(s,Row(numeral(b),numeral(h)))
            high=[b for b,h in rows if h>L]
            k=max(4,L.bit_length(),min(high).bit_length() if high else 0)
            C=L.bit_length()+2; D=C+2*k
            bound=D+sum((1<<d)*max(0,D-2*d) for d in range(1,D//2+1))
            z=s; steps=0
            while not quiet(z): z=macro(z,low); steps+=1
            assert steps<=bound,(L,i,steps,bound)
            debt_checks+=1
    # Exact rational certificate after division by 2^(k-4):
    # 4k/2^(k-4)<=16 and 256*2^(4-15k/16)<=512 for k>=4.
    for k in range(4,513):
        assert 4*k <= 16*(1<<(k-4))
        assert 64-15*k <= 16  # exponent <=1, hence second term <=512.
    print(json.dumps(dict(minimum_exports=exports,query_prefix_answers=prefixes,
                         retained_forks=forks,completed_receipts=completed,
                         stressed_pipeline_debt_checks=debt_checks,deadline_inequalities=509,
                         oracle_scope='tuple FIFO, producer histogram and reader lag are semantic models'),indent=2))


if __name__=='__main__': main()