check_speculative_service.py

#!/usr/bin/env python3
"""Finite traversal-game audit; no native-machine or language lower bound.

Independent comparisons: backward ideal viability vs whole-word service traces
joined only when their observed-prefix histories agree; exhaustive two-stage
minimum work vs mandatory-prefix union. Selectors are literal binary strings; task edge labels use a fixed alphabet.
"""
from functools import lru_cache
from itertools import product, combinations
import json


def path(word):
    return frozenset(word[:i] for i in range(1, len(word) + 1))


def tree(words):
    return frozenset().union(*(path(w) for w in words))


def ideals(edges):
    edges = sorted(edges)
    return tuple(frozenset(e for e, b in zip(edges, bits) if b)
                 for bits in product((0, 1), repeat=len(edges))
                 if all(not b or len(e) == 1 or bits[edges.index(e[:-1])]
                        for e, b in zip(edges, bits)))


def mandatory(candidates, remaining, rate):
    return frozenset().union(*(path(v[:max(0, len(v)-rate*remaining)])
                               for v in candidates))


def viability(mapping, rate):
    all_ideals = ideals(tree(mapping.values()))
    depth = len(next(iter(mapping)))

    @lru_cache(None)
    def win(prefix, done):
        if len(prefix) == depth:
            return path(mapping[prefix]) <= done
        for bit in '01':
            child = prefix + bit
            candidates = [v for q, v in mapping.items() if q.startswith(child)]
            if not candidates:
                continue
            allowed = tree(candidates)
            if not any(win(child, after) for after in all_ideals
                       if done <= after and len(after-done) <= rate
                       and after-done <= allowed):
                return False
        return True
    return win, all_ideals


def trace_join(mapping, rate):
    """Enumerate serial jobs, then join successful full-word histories.

    Unlike viability(), this does not enumerate ideals or propagate winning
    states backwards. A policy exists exactly when successful traces for all
    words agree after every common observed prefix.
    """
    depth = len(next(iter(mapping)))

    def batch(done, allowed, left):
        yield done
        if left:
            for edge in allowed-done:
                if len(edge) == 1 or edge[:-1] in done:
                    yield from batch(done | {edge}, allowed, left-1)

    def traces(word, prefix='', done=frozenset(), history=()):
        if len(prefix) == depth:
            if path(mapping[word]) <= done:
                yield history
            return
        child = word[:len(prefix)+1]
        allowed = tree([v for q, v in mapping.items() if q.startswith(child)])
        for after in set(batch(done, allowed, rate)):
            yield from traces(word, child, after, history+((child, after),))

    trace_lists = [list(traces(word)) for word in sorted(mapping)]

    def join(i, chosen):
        if i == len(trace_lists):
            return True
        for trace in trace_lists[i]:
            if all(p not in chosen or chosen[p] == s for p, s in trace):
                if join(i+1, {**chosen, **dict(trace)}):
                    return True
        return False
    return join(0, {})


def main():
    counts = dict(games=0, winning_states=0, two_stage=0,
                  offline_false_positives=0, long_tail_families=0,
                  shallow_controls=0)
    queries = ('00', '01', '10', '11')
    targets = ('00', '01', '10', '11')
    for output in product(targets, repeat=4):
        mapping = dict(zip(queries, output))
        for rate in (1, 2):
            win, states = viability(mapping, rate)
            actual = win('', frozenset())
            assert actual == trace_join(mapping, rate)
            counts['games'] += 1
            offline = all(len(v) <= rate*2 for v in output)
            counts['offline_false_positives'] += offline and not actual
            for prefix in ('', '0', '1', *queries):
                candidates = [v for q, v in mapping.items() if q.startswith(prefix)]
                need = mandatory(candidates, 2-len(prefix), rate)
                for state in states:
                    if win(prefix, state):
                        assert need <= state
                        counts['winning_states'] += 1
    # Exact minimum in the fully-revealed-after-cut relaxation.
    full_states = ideals(tree(targets))
    for size in range(1, 5):
        for subset in combinations(targets, size):
            for remaining in range(3):
                for rate in (1, 2):
                    required = mandatory(subset, remaining, rate)
                    for initial in full_states:
                        optimum = min(len(s-initial) for s in full_states
                                      if initial <= s and all(
                                          len(path(v)-s) <= rate*remaining
                                          for v in subset))
                        assert optimum == len(required-initial)
                        counts['two_stage'] += 1
    # Explicit minimal online/offline mismatch, selector differs at last bit.
    m = {'00': '00', '01': '11'}
    assert all(len(v) <= 2 for v in m.values())
    assert not viability(m, 1)[0]('', frozenset())
    assert not trace_join(m, 1)
    # Scalar root-budget cuts pass everywhere, but policies conflict.
    conflict = dict(zip(('000', '001', '010', '011'),
                        ('a0', 'b0', 'c0', 'd0')))
    for prefix in ('', '0', '00', '01', *conflict):
        possible = [v for q, v in conflict.items() if q.startswith(prefix)]
        assert len(mandatory(possible, 3-len(prefix), 1)) <= len(prefix)
    assert not viability(conflict, 1)[0]('', frozenset())
    assert not trace_join(conflict, 1)
    counts['incompatible_cut_policy_control'] = 1
    # All-length formulas corroborated at finite parameter values.
    for rate in range(1, 7):
        for n in range(2, 10):
            length, stem, width = 2*rate*n, 3*n, 2**n
            assert n+length <= rate*(stem+n)
            assert width*(length-rate*n) > rate*stem
            counts['long_tail_families'] += 1
    for n in range(1, 7):
        words = [''.join(b) for b in product('01', repeat=n)]
        assert not mandatory(words, n, 1)
        for word in words:
            done = frozenset()
            for i in range(1, n+1):
                before = done
                done |= {word[:i]}
                assert len(done-before) == 1
            assert done == path(word)
        counts['shallow_controls'] += 1
    print(json.dumps({'status': 'pass', 'counts': counts,
                      'scope': 'finite optimistic traversal service game'}, indent=2))


if __name__ == '__main__':
    main()