check_release_deadlines.py

#!/usr/bin/env python3
"""Finite controls for review 230; no native scaffold execution is simulated.

The scheduling oracle services one unit per tick, only on actual symbols.
The separate algebraic checks use the recurrence, max expansion, and tails.
Semantic controls enumerate every binary two-state DFA, using whole words as
the independent oracle for all-state transformation summaries. Counts below
are accumulated during this invocation, not retained historical evidence.
"""

from itertools import product
import json


def tick_service(horizon, capacity, jobs):
    """Jobs are (release symbol, positive cost), in fixed serial order."""
    completed = []
    current = 0
    remaining = jobs[0][1] if jobs else 0
    for symbol in range(1, horizon + 1):
        for slot in range(1, capacity + 1):
            if current == len(jobs) or jobs[current][0] > symbol:
                continue
            remaining -= 1
            if remaining == 0:
                completed.append((symbol - 1) * capacity + slot)
                current += 1
                if current < len(jobs):
                    remaining = jobs[current][1]
    return completed


def recurrence(capacity, jobs):
    completed = []
    previous = 0
    for release, cost in jobs:
        previous = max(previous, capacity * (release - 1)) + cost
        completed.append(previous)
    return completed


def check_scheduling():
    counts = dict(cases=0, feasible=0, infeasible=0, nonmonotone_releases=0,
                  empty_jobs=0, exact_deadline=0, late_single_job_controls=0)
    for horizon in range(1, 5):
        for capacity in range(1, 4):
            choices = tuple(product(range(1, horizon + 1), range(1, 4)))
            for size in range(5):
                for jobs in product(choices, repeat=size):
                    actual = tick_service(horizon, capacity, jobs)
                    predicted = recurrence(capacity, jobs)
                    budget = horizon * capacity
                    assert actual == [f for f in predicted if f <= budget], (
                        horizon, capacity, jobs, actual, predicted)
                    expanded = max((capacity * (release - 1)
                                    + sum(p for _, p in jobs[j:])
                                    for j, (release, _) in enumerate(jobs)),
                                   default=0)
                    assert expanded == (predicted[-1] if predicted else 0)
                    tails_fit = all(sum(p for _, p in jobs[j:])
                                    <= capacity * (horizon - release + 1)
                                    for j, (release, _) in enumerate(jobs))
                    feasible = len(actual) == size
                    assert feasible == tails_fit == (expanded <= budget)
                    counts['cases'] += 1
                    counts['feasible' if feasible else 'infeasible'] += 1
                    counts['empty_jobs'] += not jobs
                    counts['exact_deadline'] += expanded == budget
                    counts['nonmonotone_releases'] += any(
                        jobs[j][0] > jobs[j + 1][0] for j in range(size - 1))
    # Arbitrarily late releases cannot borrow earlier unused service.
    for capacity in range(1, 5):
        for horizon in range(capacity + 1, capacity + 9):
            jobs = ((horizon, horizon),)
            assert not tick_service(horizon, capacity, jobs)
            assert recurrence(capacity, jobs)[-1] > horizon * capacity
            counts['late_single_job_controls'] += 1
    return counts


def words(maximum):
    return [w for size in range(maximum + 1)
            for w in product(range(2), repeat=size)]


def run(delta, word, state):
    for letter in word:
        state = delta[state][letter]
    return state


def summary(delta, word):
    return tuple(run(delta, word, q) for q in range(2))


def concat(left, right):
    """T_(uv) = T_v composed with T_u, including empty words."""
    return tuple(right[left[q]] for q in range(2))


def check_semantics():
    counts = dict(dfas=0, concatenation_pairs=0, append_prepend_prefixes=0,
                  simultaneous_old_value_steps=0, accepting_set_decisions=0)
    short_words = words(4)
    for transitions in product(range(2), repeat=4):
        delta = (transitions[:2], transitions[2:])
        counts['dfas'] += 1
        assert summary(delta, ()) == (0, 1)
        for u in short_words:
            for v in short_words:
                assert concat(summary(delta, u), summary(delta, v)) == summary(delta, u + v)
                counts['concatenation_pairs'] += 1
        for word in words(7):
            sx = sy = (0, 1)
            for letter in word:
                a = summary(delta, (letter,))
                sx, sy = concat(sx, a), concat(a, sy)
            whole = word + word[::-1]
            combined = concat(sx, sy)
            assert sx == summary(delta, word)
            assert sy == summary(delta, word[::-1])
            assert combined == summary(delta, whole)
            counts['append_prepend_prefixes'] += 1
            for initial in range(2):
                for mask in range(4):
                    accepting = {q for q in range(2) if mask & (1 << q)}
                    assert (combined[initial] in accepting) == (run(delta, whole, initial) in accepting)
                    counts['accepting_set_decisions'] += 1
        # Cross-register uses detect accidental sequential updates. Operation 1
        # copies old x, deliberately showing copylessness is unnecessary here.
        for trace in words(5):
            x, y = (), ()
            sx = sy = (0, 1)
            for op in trace:
                if op == 0:
                    x, y = y + (0,), (1,) + x
                    sx, sy = concat(sy, summary(delta, (0,))), concat(summary(delta, (1,)), sx)
                else:
                    x, y = x + x, x + y
                    sx, sy = concat(sx, sx), concat(sx, sy)
                assert sx == summary(delta, x) and sy == summary(delta, y)
                assert concat(concat(sx, sy), sx) == summary(delta, x + y + x)
                counts['simultaneous_old_value_steps'] += 1
    return counts


if __name__ == '__main__':
    print(json.dumps({
        'status': 'PASS',
        'scope': 'finite mathematical controls; not native execution evidence',
        'domains': {
            'scheduling': 'T=1..4, C=1..3, k=0..4, each r=1..T, each p=1..3; arbitrary release order',
            'semantics': 'all 16 binary two-state DFAs; concatenation words <=4; prepend/append words <=7; simultaneous traces <=5',
        },
        'scheduling': check_scheduling(),
        'semantics': check_semantics(),
    }, indent=2))