check_completion_debt.py

#!/usr/bin/env python3
"""Semantic checks for final-diversity completion debt, not native primitives."""
from collections import defaultdict
from heapq import heappop, heappush
from itertools import product, combinations
from math import comb
import json


def debt(counts, prices, cap):
    groups = defaultdict(int)
    for c, p in zip(counts, prices):
        groups[c] += p
    vals = sorted(groups)
    d = len(vals)
    table = [[float('inf')] * (d + 1) for _ in range(cap + 1)]
    for row in table:
        row[0] = 0
    for k in range(1, cap + 1):
        for j in range(1, d + 1):
            for i in range(j):
                cost = sum(groups[vals[t]] * (vals[j - 1] - vals[t])
                           for t in range(i, j))
                table[k][j] = min(table[k][j], table[k - 1][i] + cost)
    return table[cap][d]


def search(counts, prices, cap):
    """Dijkstra on named row counts; no ordered-block premise used."""
    initial = tuple(counts)
    ceiling = max(initial, default=0)
    queue = [(0, initial)]
    dist = {initial: 0}
    while queue:
        cost, state = heappop(queue)
        if cost != dist[state]:
            continue
        if len(set(state)) <= cap:
            return cost
        for i, val in enumerate(state):
            if val == ceiling:
                continue
            nxt = state[:i] + (val + 1,) + state[i + 1:]
            alt = cost + prices[i]
            if alt < dist.get(nxt, float('inf')):
                dist[nxt] = alt
                heappush(queue, (alt, nxt))
    raise AssertionError('unreachable completion')


def pair_bound(d, cap):
    q, r = divmod(d, cap)
    return cap * q * (q - 1) // 2 + r * q


def shortest_price(m):
    return 10 * m + sum(len(v) for v in shortest_names(m))


def shortest_names(m):
    out = []
    width = 1
    while len(out) < m:
        for bits in product('01', repeat=width):
            out.append(''.join(bits))
            if len(out) == m:
                break
        width += 1
    return out


def encoded_bound(d, cap):
    return max((t * shortest_price(d - cap * t)
                for t in range(1, (d - 1) // cap + 1)), default=0)


def clause(v, w):
    return f'(-01{v},+1{w});'


def release_checks():
    """Independent workload simulation and literal radial terminal witnesses."""
    counts = dict(workload_checks=0, streaming_checks=0, final_record_witnesses=0)
    for horizon in range(1, 6):
        for arrivals in product(range(4), repeat=horizon):
            for initial in range(5):
                for rate in (1, 2, 3):
                    # Explicit unit-work queue, rather than the closed-form maximum.
                    queue = [None] * initial
                    for new in arrivals:
                        queue.extend([None] * new)
                        for _ in range(rate):
                            if queue:
                                queue.pop()
                    terms = [0, initial + sum(arrivals) - rate * horizon]
                    terms += [sum(arrivals[r:]) - rate * (horizon-r)
                              for r in range(horizon)]
                    assert len(queue) == max(terms)
                    counts['workload_checks'] += 1
                    if max(arrivals) <= rate and initial <= (rate-max(arrivals))*horizon:
                        assert not queue
                        counts['streaming_checks'] += 1
    for width in range(2, 11):
        full = {''.join(bits) for bits in product('01', repeat=width)}
        last = '1' * width
        rows = {'10': full.copy(), '11': {'0', '1'}, '0': full - {last}}
        assert len({len(s) for s in rows.values()}) == 3
        assert debt([len(s) for s in rows.values()], [12, 12, 11], 2) == 11
        literal = clause('0', last)
        assert len(literal) == width + 10 and last not in rows['0']
        rows['0'].add(last)
        assert len({frozenset(s) for s in rows.values()}) == 2
        assert all(s == {''.join(b) for b in product('01', repeat=len(next(iter(s))))}
                   for s in rows.values())
        # Generous model: all delimiter work becomes ready before its service.
        rate = width-1
        burst = [0]*(len(literal)-1) + [width]
        backlog = 0
        for work in burst:
            backlog = max(0, backlog + work-rate)
        assert sum(burst) <= rate*len(literal) and backlog == 1
        counts['final_record_witnesses'] += 1
    return counts


def main():
    checks = dict(weighted_searches=0, ordered_states=0, name_assignments=0,
                  offset_witnesses=0, late_merge_witnesses=0, debt_drain_checks=0)
    for counts in product(range(1, 5), repeat=3):
        for prices in product((1, 3), repeat=3):
            for cap in (1, 2, 3):
                assert debt(counts, prices, cap) == search(counts, prices, cap)
                checks['weighted_searches'] += 1
                old = debt(counts, prices, cap)
                for i, price in enumerate(prices):
                    nxt = list(counts)
                    nxt[i] += 1
                    assert debt(nxt, prices, cap) >= old - price
                    checks['debt_drain_checks'] += 1
    for d in range(1, 8):
        for vals in combinations(range(1, 10), d):
            for cap in range(1, 5):
                assert debt(vals, [1] * d, cap) >= pair_bound(d, cap)
                assert debt([c + 1000 for c in vals], [1] * d, cap) == debt(vals, [1] * d, cap)
                checks['ordered_states'] += 1
        for cap in range(1, 5):
            assert debt(list(range(d)), [1] * d, cap) == pair_bound(d, cap)
    for d in range(1, 8):
        names = shortest_names(d)
        prices = [10 + len(v) for v in names]
        for rotation in range(d):
            pp = prices[rotation:] + prices[:rotation]
            for cap in range(1, 5):
                assert debt(list(range(d)), pp, cap) >= encoded_bound(d, cap)
                checks['name_assignments'] += 1
    assert encoded_bound(8, 2) == 92
    assert encoded_bound(7, 2) == 68
    assert 92 - 68 > len(clause('0', '0')) == 11
    for m in range(1, 100):
        level = 0
        while 2 ** (level + 2) - 2 <= m:
            level += 1
        s = 2 ** (level + 1) - 2
        formula = 10 * m + (level - 1) * 2 ** (level + 1) + 2 + (m - s) * (level + 1)
        assert shortest_price(m) == formula
    for d in range(2, 9):
        for width in range(max(1, (d - 1).bit_length()), 9):
            names = [format(i, f'0{(d - 1).bit_length()}b') for i in range(d)]
            targets = [''.join(b) for b in product('01', repeat=width)]
            rows = {v: set(targets[:2 ** width - d + i + 1]) for i, v in enumerate(names)}
            suffix = []
            for v in names:
                for w in targets:
                    if w not in rows[v]:
                        suffix.append(clause(v, w))
                        rows[v].add(w)
            assert len(suffix) == comb(d, 2)
            assert sum(map(len, suffix)) == comb(d, 2) * (9 + (d - 1).bit_length() + width)
            assert len({frozenset(s) for s in rows.values()}) == 1
            checks['offset_witnesses'] += 1
    for k in range(9):
        rows = {'11' + ''.join(u): {'0', '1'} for u in product('01', repeat=k)}
        rows['10'] = {'00', '01', '10', '11'}
        rows['0'] = {'00', '01', '10'}
        assert len({len(ws) for ws in rows.values()}) == 3
        assert debt([len(ws) for ws in rows.values()], [1] * len(rows), 2) == 1
        assert len(clause('0', '11')) == 12
        rows['0'].add('11')
        assert len({frozenset(ws) for ws in rows.values()}) == 2
        checks['late_merge_witnesses'] += 1
    checks.update(release_checks())
    print(json.dumps(checks, sort_keys=True))


if __name__ == '__main__':
    main()