check_sparse_age_minimum.py

#!/usr/bin/env python3
"""Immutable sparse-age minimum worker with no priority or endpoint-order oracle.

Unary age/gap links, task stacks and endpoint digit lists are explicit.
Right-input production and scaffold packing are inherited; no arithmetic
counter, priority selector or endpoint-order primitive occurs in the worker.
Direct integer minima occur only in the independent audit.
"""
from dataclasses import dataclass, replace
from random import Random
import json


@dataclass(frozen=True)
class Unit:
    tail: object = None


ONE = Unit()


@dataclass(frozen=True)
class Digit:
    bit: int
    tail: object = None


def numeral(n):
    # Test preparation, outside worker; one actual digit node per bit.
    head = None
    for c in bin(n)[2:]:
        head = Digit(int(c), head)
    return head


def decode(p):
    n, power = 0, 1
    while p is not None:
        n += power*p.bit
        power *= 2
        p = p.tail
    return n


def depth(p):
    n = 0
    while p is not None:
        n += 1
        p = p.tail
    return n


@dataclass(frozen=True)
class Job:
    endpoint: object
    cursor: object


@dataclass(frozen=True)
class Stack:
    head: object
    tail: object = None


@dataclass(frozen=True)
class Compare:
    candidate: object
    left: object
    right: object
    order: int = 0


@dataclass(frozen=True)
class Bucket:
    age: object = None
    gap: object = None
    jobs: object = None
    comparison: object = None
    best: object = None
    following: object = None


def release(root, endpoint):
    task = Stack(Job(endpoint, endpoint))
    if root is None:
        return Bucket(jobs=task)
    if root.age is None:
        return replace(root, jobs=Stack(task.head, root.jobs))
    return Bucket(gap=root.age, jobs=task, following=root)


def idle(root):
    return root is None or (root.jobs is None and root.comparison is None)


def step(root):
    if idle(root):
        return root
    if root.comparison is not None:
        c = root.comparison
        a = 0 if c.left is None else c.left.bit
        b = 0 if c.right is None else c.right.bit
        order = (1 if a > b else -1) if a != b else c.order
        left = None if c.left is None else c.left.tail
        right = None if c.right is None else c.right.tail
        if left is None and right is None:
            return replace(root, comparison=None,
                           best=c.candidate if order < 0 else root.best)
        return replace(root, comparison=Compare(c.candidate, left, right, order))
    job, rest = root.jobs.head, root.jobs.tail
    cursor = job.cursor.tail
    if cursor is None:
        if root.best is None:
            # A shorter length class dominates the entire following chain.
            return Bucket(root.age, jobs=rest, best=job.endpoint)
        return replace(root, jobs=rest,
                       comparison=Compare(job.endpoint, job.endpoint, root.best))
    advanced = Job(job.endpoint, cursor)
    if root.best is not None:
        # The unfinished endpoint has crossed beyond the best's length class.
        return replace(root, jobs=rest)
    nxt = root.following
    if nxt is not None and root.gap.tail is None:
        promoted = replace(nxt, jobs=Stack(advanced, nxt.jobs))
    else:
        promoted = Bucket(Unit(root.age),
                          None if nxt is None else root.gap.tail,
                          Stack(advanced), following=nxt)
    if rest is None:
        return promoted
    return replace(root, jobs=rest, gap=ONE, following=promoted)


def serve(root, n):
    for _ in range(n):
        if idle(root):
            break
        root = step(root)
    return root


def inspect(root):
    """Independent whole-state invariant checker, not a worker operation."""
    records, completed, ages = [], [], []
    p = root
    while p is not None:
        j = depth(p.age)
        ages.append(j)
        assert p.jobs is not None or p.comparison is not None or p.best is not None
        if p.following is not None:
            assert depth(p.gap) == depth(p.following.age)-j > 0
        else:
            assert p.gap is None
        if p.best is not None:
            b = decode(p.best)
            assert b.bit_length() == j+1
            assert p.following is None
            completed.append(b)
        if p.comparison is not None:
            assert p.best is not None
            records.append(decode(p.comparison.candidate))
        jobs = p.jobs
        while jobs is not None:
            t = jobs.head
            assert decode(t.endpoint).bit_length()-decode(t.cursor).bit_length() == j
            assert t.cursor is not None
            records.append(decode(t.endpoint))
            jobs = jobs.tail
        p = p.following
    assert ages == sorted(set(ages))
    return records+completed


def finite_main():
    rng = Random(1931)
    states = answers = forks = interrupts = 0
    for trial in range(100):
        root, values, saved = None, [], []
        for i in range(70):
            # Small service keeps several ages and interruptions live.
            root = serve(root, rng.randrange(0, 7))
            v = rng.choice((1, 2, 3, 7, 17, rng.randrange(1, 1 << 25), 1 << rng.randrange(30, 250)))
            root = release(root, numeral(v))
            values.append(v)
            assert min(inspect(root)) == min(values)
            saved.append((root, tuple(values)))
            states += 1
        for old, vals in rng.sample(saved, 8):
            before = tuple(inspect(old))
            t = old
            # Complete all remaining work for exact-minimum oracle comparison.
            while not idle(t):
                t = step(t)
                assert min(inspect(t)) == min(vals)
                states += 1
            assert decode(t.best) == min(vals)
            assert tuple(inspect(old)) == before
            forks += 1
            for n in (0, 1, 2, 3, min(vals)-1, min(vals), min(vals)+1):
                assert (decode(t.best) <= n) == any(v <= n for v in vals)
                answers += 1
    # A live age-30 comparison is interrupted at every digit boundary. New
    # lower ages may reach or delete its bucket; same-age promotions preserve it.
    base, candidate = (1 << 30)+27, (1 << 30)+19
    root = serve(release(None, numeral(base)), 10000)
    root = release(root, numeral(candidate))
    while root.comparison is None:
        root = step(root)
    for delay in range(31):
        old = serve(root, delay)
        for v in (2, (1 << 30)+3, (1 << 31)+1):
            t = release(old, numeral(v))
            t = serve(t, 10000)
            assert idle(t) and decode(t.best) == min(base, candidate, v)
            assert decode(old.best) == base
            interrupts += 1
    # Build a long sparse frontier before any numeral completes, then test
    # both a gap split and promotion into an occupied successor bucket.
    deep = None
    for j in range(10, -1, -1):
        deep = release(deep, numeral(1 << 4096))
        deep = serve(deep, (1 << j)-1)
        inspect(deep)
    p, bucket_count = deep, 0
    while p is not None:
        bucket_count += 1
        p = p.following
    assert bucket_count >= 9
    deep_forks = 0
    for delay in range(200):
        frozen = serve(deep, delay)
        old_values = tuple(inspect(frozen))
        for v in (2, 17, 1 << 40):
            t = serve(release(frozen, numeral(v)), 50000)
            assert idle(t) and decode(t.best) == v
            assert tuple(inspect(frozen)) == old_values
            deep_forks += 1

    # Continuous query service and explicit lag repair; no cutoff/age oracle.
    # Reader lengths model the inherited unary rulers, not an order operation.
    phase_answers = 0
    for trial in range(50):
        root, vals = None, []
        for i in range(60):
            name = bin(i+2)[2:]
            root = serve(root, 16*len(name))
            v = rng.choice((2, 3, 127, 4096, 1 << 300))
            root = release(root, numeral(v))
            vals.append(v)
        lower = None
        lag = 0
        aligned = False
        for n in range(1, 257):
            if lower is None:
                root = serve(root, 256)
                if idle(root):
                    lower, lag = decode(root.best), n
            elif aligned:
                lower = max(0, lower-1)
            else:
                lag += 1
            if lower is not None and not aligned:
                for _ in range(4):
                    if lag:
                        lag -= 1
                        lower = max(0, lower-1)
                aligned = lag == 0
            assert (aligned and lower == 0) == (min(vals) <= n)
            phase_answers += 1
    # A partially screened enormous job does not hide a late endpoint 2.
    old = serve(release(None, numeral(1 << 50000)), 1000)
    t = release(old, numeral(2))
    t = serve(t, 256)
    assert idle(t) and decode(t.best) == 2
    assert not idle(old)
    print(json.dumps(dict(invariant_states=states, retained_forks=forks,
        completed_minimum_answers=answers, comparison_interruptions=interrupts,
        query_prefix_answers=phase_answers, deep_frontier_buckets=bucket_count,
        deep_interruption_forks=deep_forks, priority_oracle=False,
        numeral_order_oracle=False, arithmetic_counter_oracle=False),indent=2))


if __name__ == '__main__':
    finite_main()