check_one_counter_certificates.py

#!/usr/bin/env python3
"""Exact finite certificate checker; see one-counter-certificates.md.

Run without arguments for examples and bounded implementation audits.
Run --check instance.json to check one source/clock/rows JSON instance.
No saturation, synthesis, determinization or scaffold generation is implemented.
"""
import argparse
from dataclasses import dataclass
from itertools import product
import json
from math import lcm


@dataclass(frozen=True)
class UP:
    T: int
    p: int
    below: frozenset
    residues: frozenset

    def __post_init__(self):
        if type(self.T) is not int or self.T < 0:
            raise ValueError('threshold must be a nonnegative integer')
        if type(self.p) is not int or self.p < 1:
            raise ValueError('period must be a positive integer')
        if any(type(x) is not int or not 0 <= x < self.T for x in self.below):
            raise ValueError('invalid finite-prefix member')
        if any(type(x) is not int or not 0 <= x < self.p for x in self.residues):
            raise ValueError('invalid residue')

    def has(self, z):
        return z >= 0 and (z in self.below if z < self.T
                           else z % self.p in self.residues)

    def nonempty(self):
        return bool(self.below or self.residues)


EMPTY = UP(0, 1, frozenset(), frozenset())
ZERO = UP(1, 1, frozenset({0}), frozenset())
NAT = UP(0, 1, frozenset(), frozenset({0}))


def shifted_failure(A, t, target):
    """Least positive z violating (A intersect positive)+t subset target.

    None means the infinite inclusion holds, by the finite-period proof.
    """
    if t not in (-1, 0, 1):
        raise ValueError('shift outside unit range')
    M = max(1, A.T, target.T + 1)
    L = lcm(A.p, target.p)
    return next((z for z in range(1, M + L)
                 if A.has(z) and not target.has(z + t)), None)


def enabled(guard, h):
    return guard == 'any' or (guard == 'zero' and h == 0) or (
        guard == 'positive' and h > 0)


def boundary(e, d, c):
    h = max(0, max(e, 0) + d)
    H = max(0, max(-e, 0) + c)
    return h - H, min(h, H)


def interior(e, d, c):
    return e + d - c, min(max(e, 0) + d, max(-e, 0) + c)


def check(instance):
    source, clock = instance['source'], instance['clock']
    Q, R, alphabet = set(source['states']), set(clock['states']), set(source['alphabet'])
    initial = set(source['initial'])
    if not initial or not initial <= Q or clock['initial'] not in R:
        raise ValueError('invalid initialization')
    table = {}
    for row in clock['table']:
        r, a, nxt, c = row
        if r not in R or nxt not in R or a not in alphabet or type(c) is not int or c not in (-1, 0, 1):
            raise ValueError('invalid clock transition')
        if (r, a) in table:
            raise ValueError('duplicate clock transition')
        table[r, a] = nxt, c
    if set(table) != set(product(R, alphabet)):
        raise ValueError('clock must be complete')
    rules = source['rules']
    for q, a, g, d, nxt in rules:
        if q not in Q or nxt not in Q or a not in alphabet or g not in ('zero', 'positive', 'any') or type(d) is not int or d not in (-1, 0, 1):
            raise ValueError('invalid source rule')
    rows = {}
    for row in instance['rows']:
        q, r, e = row['q'], row['r'], row['e']
        if q not in Q or r not in R or type(e) is not int or (q, r, e) in rows:
            raise ValueError('invalid or duplicate certificate row')
        rows[q, r, e] = UP(row['T'], row['p'], frozenset(row['below']), frozenset(row['residues']))
    for q in sorted(initial):
        if not rows.get((q, clock['initial'], 0), EMPTY).has(0):
            return {'valid': False, 'reason': 'initialization', 'q': q}
    for (q, r, e), A in sorted(rows.items()):
        for index, (old, a, guard, d, nxt) in enumerate(rules):
            if old != q:
                continue
            nr, c = table[r, a]
            if A.has(0) and enabled(guard, max(e, 0)):
                ne, nz = boundary(e, d, c)
                if not rows.get((nxt, nr, ne), EMPTY).has(nz):
                    return {'valid': False, 'reason': 'boundary',
                            'mode': [q, r, e], 'z': 0, 'rule_index': index,
                            'target_mode': [nxt, nr, ne], 'target_z': nz}
            if guard != 'zero':
                ne, t = interior(e, d, c)
                z = shifted_failure(A, t, rows.get((nxt, nr, ne), EMPTY))
                if z is not None:
                    return {'valid': False, 'reason': 'interior',
                            'mode': [q, r, e], 'z': z, 'rule_index': index,
                            'target_mode': [nxt, nr, ne], 'target_z': z + t}
    return {'valid': True, 'bound': max(abs(e) for (_, _, e), A in rows.items() if A.nonempty())}


def row(q, r, e, A):
    return {'q': q, 'r': r, 'e': e, 'T': A.T, 'p': A.p,
            'below': sorted(A.below), 'residues': sorted(A.residues)}


def one_state(rules, outputs, rows):
    return {'source': {'states': [0], 'initial': [0], 'alphabet': list(outputs),
                       'rules': [[0, a, guard, d, 0] for a, guard, d in rules]},
            'clock': {'states': [0], 'initial': 0,
                      'table': [[0, a, 0, c] for a, c in outputs.items()]},
            'rows': [row(0, 0, e, A) for e, A in rows]}


def examples():
    bottom = one_state([('x', 'any', -1), ('x', 'any', 0)], {'x': 0}, [(0, ZERO)])
    once = one_state([('x', 'zero', 1), ('x', 'positive', 0)], {'x': 0}, [(0, ZERO), (1, ZERO)])
    dyck = one_state([('a', 'any', 1), ('b', 'positive', -1),
                      ('x', 'any', 0), ('x', 'zero', -1)],
                     {'a': 1, 'b': -1, 'x': 0}, [(0, NAT)])
    parity = {'source': {'states': [0, 1], 'initial': [0], 'alphabet': ['a'],
                         'rules': [[0, 'a', 'any', 1, 1], [1, 'a', 'any', 1, 0]]},
              'clock': {'states': [0], 'initial': 0, 'table': [[0, 'a', 0, 1]]},
              'rows': [row(0, 0, 0, UP(0, 2, frozenset(), frozenset({0}))),
                       row(1, 0, 0, UP(0, 2, frozenset(), frozenset({1})))]}
    clock_once = {'source': bottom['source'],
                  'clock': {'states': [0, 1], 'initial': 0,
                            'table': [[0, 'x', 1, 1], [1, 'x', 1, 0]]},
                  'rows': [row(0, 0, 0, ZERO), row(0, 1, -1, ZERO)]}
    unbounded = one_state([('x', 'any', 1), ('x', 'any', 0)], {'x': 0}, [(0, ZERO), (1, ZERO)])
    false_ray = one_state([('x', 'any', -1), ('x', 'any', 0)], {'x': 0}, [(0, NAT)])
    bad_parity = json.loads(json.dumps(parity))
    bad_parity['rows'][1]['residues'] = [0]
    return [('bottom_pop_keep', bottom, True), ('guarded_push_once', once, True),
            ('dyck_with_bottom_noise', dyck, True), ('parity_rows', parity, True),
            ('clock_ahead_at_boundary', clock_once, True),
            ('unbounded_source_truncated_certificate', unbounded, False),
            ('unrestricted_ray_for_bottom_only_source', false_ray, False),
            ('corrupted_parity_row', bad_parity, False)]


def finite_audit():
    # These are implementation corroboration, not unrestricted theorem proofs.
    updates = 0
    for e, z, d, c in product(range(-5, 6), range(9), (-1, 0, 1), (-1, 0, 1)):
        h, H = z + max(e, 0), z + max(-e, 0)
        u, v = max(0, h + d), max(0, H + c)
        ne, nz = boundary(e, d, c) if z == 0 else interior(e, d, c)
        if z:
            nz += z
        assert (ne, nz) == (u - v, min(u, v))
        updates += 1
    sets = []
    for T, p in product(range(3), range(1, 4)):
        for maskF, maskD in product(range(1 << T), range(1 << p)):
            sets.append(UP(T, p, frozenset(i for i in range(T) if maskF >> i & 1),
                           frozenset(i for i in range(p) if maskD >> i & 1)))
    inclusions = 0
    for A, target, t in product(sets, sets, (-1, 0, 1)):
        witness = shifted_failure(A, t, target)
        oracle = next((z for z in range(1, 81) if A.has(z) and not target.has(z + t)), None)
        assert witness == oracle
        inclusions += 1
    return {'coordinate_formula_cases': updates,
            'coordinate_ranges': {'e': [-5, 5], 'z': [0, 8], 'd_and_c': [-1, 0, 1]},
            'periodic_descriptions': len(sets), 'inclusion_cases': inclusions,
            'periodic_ranges': {'threshold': [0, 2], 'period': [1, 3],
                                'prefix_and_residue_subsets': 'all', 'oracle_z': [1, 80]},
            'status': 'bounded implementation audit passed'}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--check', metavar='JSON')
    args = parser.parse_args()
    if args.check:
        with open(args.check, encoding='utf-8') as f:
            result = check(json.load(f))
        print(json.dumps(result, indent=2, sort_keys=True))
        raise SystemExit(0 if result['valid'] else 1)
    results = []
    for name, instance, expected in examples():
        result = check(instance)
        assert result['valid'] == expected, (name, result)
        results.append({'name': name, 'instance': instance, 'exact_check': result})
    print(json.dumps({'note': 'one-counter-certificates.md',
                      'exact_scope': 'local inductive certificate obligations, including infinite periodic inclusions',
                      'not_implemented': ['saturation', 'certificate synthesis', 'VPA determinization',
                                          'scaffold generation', 'proof-assistant verification'],
                      'examples': results, 'finite_audit': finite_audit()}, indent=2, sort_keys=True))


if __name__ == '__main__':
    main()