verify_counter_free_real_time_queue.py

#!/usr/bin/env python3
"""Finite checks for the counter-free strict persistent queue.

The transition functions use only tuple constructors, finite tags, head/tail
access and emptiness tests. Length is used only by the independent audit.
"""
from collections import deque
from pathlib import Path
import json
import random


EMPTY_STATUS = ('Idle',)
EMPTY_QUEUE = ((), (), EMPTY_STATUS, ((), ()))


def pop_budget(budget):
    first, second = budget
    if first:
        return (first[1:], second), True
    if second:
        return ((), second[1:]), True
    return budget, False


def invalidate(status):
    tag = status[0]
    if tag == 'Rev':
        _, tokens, front, front_acc, rear, rear_acc = status
        return ('Rev', tokens[1:] if tokens else (), front, front_acc,
                rear, rear_acc)
    if tag == 'App':
        _, tokens, front_acc, result = status
        if not tokens:
            assert result
            return ('Done', result[1:])
        return ('App', tokens[1:], front_acc, result)
    return status


def exec_one(status):
    """Return one structural microstep and conceptual fresh-data count."""
    tag = status[0]
    if tag == 'Rev':
        _, tokens, front, front_acc, rear, rear_acc = status
        if front:
            assert rear
            # One Token and two List nodes. The final status record is counted
            # once after the two fused microsteps.
            return (('Rev', (None,) + tokens, front[1:],
                     (front[0],) + front_acc, rear[1:],
                     (rear[0],) + rear_acc), 3)
        assert len(rear) == 1
        return ('App', tokens, front_acc, (rear[0],) + rear_acc), 1
    if tag == 'App':
        _, tokens, front_acc, result = status
        if not tokens:
            return ('Done', result), 0
        assert front_acc
        return ('App', tokens[1:], front_acc[1:],
                (front_acc[0],) + result), 1
    return status, 0


def exec_two(queue):
    front, rear, status, budget = queue
    status, first_cost = exec_one(status)
    status, second_cost = exec_one(status)
    if status[0] == 'Done':
        front, status = status[1], EMPTY_STATUS
    # One final status representation and one queue root, unless Idle can be
    # virtual. Counting both is deliberately conservative.
    return (front, rear, status, budget), first_cost + second_cost + 2


def check(queue):
    front, rear, status, budget = queue
    budget, had_token = pop_budget(budget)
    if had_token:
        return exec_two((front, rear, status, budget))
    # The previous rotation must be finished when the structural balance
    # certificate is exhausted.
    assert status[0] == 'Idle'
    snapshot_front, snapshot_rear = front, rear
    return exec_two((front, (),
                     ('Rev', (), snapshot_front, (), snapshot_rear, ()),
                     (snapshot_front, snapshot_rear)))


def enqueue(queue, symbol):
    front, rear, status, budget = queue
    target, cost = check((front, (symbol,) + rear, status, budget))
    return target, cost + 1


def dequeue(queue):
    front, rear, status, budget = queue
    assert front
    target, cost = check((front[1:], rear, invalidate(status), budget))
    return target, cost


def audit(queue, word):
    front, rear, status, budget = queue
    assert bool(front) == bool(word)
    if word:
        assert front[0] == word[0]
    # Audit only: transition code never reads these lengths.
    assert len(budget[0]) + len(budget[1]) == len(word) - 2 * len(rear)
    assert status[0] in {'Idle', 'Rev', 'App'}


def exhaustive(depth):
    seen = {(EMPTY_QUEUE, '')}
    frontier = {(EMPTY_QUEUE, '')}
    transitions = maximum_cost = 0
    for _ in range(depth):
        following = set()
        for queue, word in frontier:
            for symbol in 'ab':
                target, cost = enqueue(queue, symbol)
                target_word = word + symbol
                audit(target, target_word)
                transitions += 1
                maximum_cost = max(maximum_cost, cost)
                if (target, target_word) not in seen:
                    seen.add((target, target_word))
                    following.add((target, target_word))
            if word:
                target, cost = dequeue(queue)
                target_word = word[1:]
                audit(target, target_word)
                transitions += 1
                maximum_cost = max(maximum_cost, cost)
                if (target, target_word) not in seen:
                    seen.add((target, target_word))
                    following.add((target, target_word))
        frontier = following
    return {'trace_depth': depth, 'distinct_version_states': len(seen),
            'transitions_checked': transitions,
            'maximum_conceptual_fresh_records': maximum_cost}


def seeded_branching(roots, rounds, seed):
    rng = random.Random(seed)
    versions = [(EMPTY_QUEUE, '')]
    maximum_cost = 0
    for _ in range(rounds):
        queue, word = rng.choice(versions)
        choices = ['a', 'b'] + (['deq'] if word else [])
        operation = rng.choice(choices)
        if operation == 'deq':
            target, cost = dequeue(queue)
            target_word = word[1:]
        else:
            target, cost = enqueue(queue, operation)
            target_word = word + operation
        audit(target, target_word)
        versions.append((target, target_word))
        if len(versions) > roots:
            del versions[rng.randrange(len(versions) - 1)]
        maximum_cost = max(maximum_cost, cost)
    return {'seed': seed, 'retained_branch_roots': roots, 'rounds': rounds,
            'maximum_conceptual_fresh_records': maximum_cost}


def main():
    result = {
        'scope': ('Exact finite exploration of the structural counter-free '
                  'queue. Branches old immutable versions. Not a proof of the '
                  'unbounded theorem, record compiler, filtering realization, '
                  'or proof-assistant formalization.'),
        'exhaustive': exhaustive(15),
        'seeded_branching': seeded_branching(5000, 250000, 20260930),
        'claimed_record_bounds': {
            'maximum_arity': 6,
            'old_access_radius': 6,
            'fresh_records_per_operation': 12,
            'status': 'Conservative written accounting, updated 1 October for copied old targets; not inferred from finite tests or optimized.'
        },
        'proof_status': ('Correctness and scalar elimination are written '
                         'inductive arguments. Finite checks corroborate them.'),
    }
    target = Path(__file__).with_name(
        'counter-free-real-time-queue-output.json')
    target.write_text(json.dumps(result, indent=2) + '\n')
    print(json.dumps(result, indent=2))


if __name__ == '__main__':
    main()