verify_bounded_threshold_catenation.py

#!/usr/bin/env python3
"""Finite corroboration for bounded-threshold structural probes.

This does not prove the general lemma, the OCaml implementation, or the
source-to-record compiler. With --source it also records static facts about a
checkout of VerifiedCatenableDeque.
"""

from __future__ import annotations

import argparse
import hashlib
import json
from pathlib import Path
import random
import subprocess


def pop(xs):
    return None if not xs else (xs[0], xs[1:])


def eject(xs):
    return None if not xs else (xs[:-1], xs[-1])


def has3_probe(xs):
    r1 = pop(xs)
    if r1 is None:
        return ("less", ())
    a, t1 = r1
    r2 = pop(t1)
    if r2 is None:
        return ("less", (a,))
    b, t2 = r2
    r3 = pop(t2)
    if r3 is None:
        return ("less", (a, b))
    return ("at_least", xs)


def has7p_probe(xs):
    taken = []
    cur = xs
    after_two = None
    for i in range(7):
        result = pop(cur)
        if result is None:
            return ("less", tuple(taken))
        value, cur = result
        taken.append(value)
        if i == 1:
            after_two = cur
    return ("at_least", (tuple(taken[:2]), after_two))


def has7s_probe(xs):
    taken_reversed = []
    cur = xs
    after_two = None
    for i in range(7):
        result = eject(cur)
        if result is None:
            return ("less", tuple(reversed(taken_reversed)))
        cur, value = result
        taken_reversed.append(value)
        if i == 1:
            after_two = cur
    return ("at_least", (after_two, tuple(reversed(taken_reversed[:2]))))


def has8_probe(xs):
    assert len(xs) >= 5
    first_five = xs[:5]
    cur = xs[5:]
    extra = []
    for _ in range(3):
        result = pop(cur)
        if result is None:
            return ("less", (first_five, tuple(extra)))
        value, cur = result
        extra.append(value)
    return ("at_least", xs)


def has3p8_probe(xs):
    assert len(xs) >= 8
    first_three = xs[:3]
    result = has8_probe(xs[3:])
    if result[0] == "less":
        five, vector = result[1]
        return ("less", (first_three + five, vector))
    return ("at_least", (first_three, xs[3:]))


def expected_has3(xs):
    return ("less", xs) if len(xs) < 3 else ("at_least", xs)


def expected_has7p(xs):
    if len(xs) < 7:
        return ("less", xs)
    return ("at_least", (xs[:2], xs[2:]))


def expected_has7s(xs):
    if len(xs) < 7:
        return ("less", xs)
    return ("at_least", (xs[:-2], xs[-2:]))


def expected_has8(xs):
    if len(xs) < 8:
        return ("less", (xs[:5], xs[5:]))
    return ("at_least", xs)


def expected_has3p8(xs):
    if len(xs) < 11:
        return ("less", (xs[:8], xs[8:]))
    return ("at_least", (xs[:3], xs[3:]))


def exhaustive(max_length=64):
    checks = 0
    for n in range(max_length + 1):
        xs = tuple(range(n))
        assert has3_probe(xs) == expected_has3(xs)
        checks += 1
        if n >= 1:
            assert has7p_probe(xs) == expected_has7p(xs)
            assert has7s_probe(xs) == expected_has7s(xs)
            checks += 2
        if n >= 5:
            assert has8_probe(xs) == expected_has8(xs)
            checks += 1
        if n >= 8:
            assert has3p8_probe(xs) == expected_has3p8(xs)
            checks += 1
    return {"maximum_length": max_length, "classification_checks": checks}


def branching(operations=250_000, retained=5_000, seed=20260930):
    rng = random.Random(seed)
    versions = [tuple()]
    max_length = 0
    checks = 0
    for step in range(operations):
        base = rng.choice(versions)
        old = base
        choice = rng.randrange(4)
        if choice == 0:
            new = (step,) + base
        elif choice == 1:
            new = base + (step,)
        elif choice == 2:
            result = pop(base)
            new = base if result is None else result[1]
        else:
            result = eject(base)
            new = base if result is None else result[0]
        assert base == old
        assert has3_probe(base) == expected_has3(base)
        checks += 1
        if base:
            assert has7p_probe(base) == expected_has7p(base)
            assert has7s_probe(base) == expected_has7s(base)
            checks += 2
        if len(base) >= 5:
            assert has8_probe(base) == expected_has8(base)
            checks += 1
        if len(base) >= 8:
            assert has3p8_probe(base) == expected_has3p8(base)
            checks += 1
        versions.append(new)
        if len(versions) > retained:
            del versions[rng.randrange(len(versions) - 1)]
        max_length = max(max_length, len(new))
    return {
        "operations": operations,
        "retained_versions": retained,
        "seed": seed,
        "classification_checks": checks,
        "maximum_length": max_length,
    }


def sha256(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()


def count(text, needle):
    return text.count(needle)


def source_audit(root):
    root = root.resolve()
    dq = root / "lib" / "deque_core.ml"
    cq = root / "lib" / "cadeque_core.ml"
    dq_text = dq.read_text()
    cq_text = cq.read_text()
    head = subprocess.check_output(
        ["git", "rev-parse", "HEAD"], cwd=root, text=True
    ).strip()
    return {
        "checkout_head": head,
        "deque_core_sha256": sha256(dq),
        "cadeque_core_sha256": sha256(cq),
        "core_source_characters_with_colors": sum(
            (root / "lib" / name).stat().st_size
            for name in ("color_GYR.ml", "color_GYOR.ml", "deque_core.ml", "cadeque_core.ml")
        ),
        "cadeque_core_Deque_length_occurrences": count(cq_text, "Deque.length"),
        "deque_core_let_rec_occurrences": count(dq_text, "let rec"),
        "cadeque_core_let_rec_occurrences": count(cq_text, "let rec"),
        "deque_core_mutation_tokens": sum(count(dq_text, x) for x in (":=", " ref ", "Array.")),
        "cadeque_core_mutation_tokens": sum(count(cq_text, x) for x in (":=", " ref ", "Array.")),
        "maximum_structural_probe_operations": 11,
    }


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--source", type=Path)
    parser.add_argument("--output", type=Path)
    args = parser.parse_args()
    result = {
        "scope": {
            "claim": "finite corroboration, not an unrestricted proof",
            "classifiers": ["has3", "has7p", "has7s", "has8", "has3p8"],
        },
        "exhaustive": exhaustive(),
        "branching_persistence": branching(),
    }
    if args.source:
        result["source_audit"] = source_audit(args.source)
    rendered = json.dumps(result, indent=2, sort_keys=True) + "\n"
    if args.output:
        args.output.write_text(rendered)
    print(rendered, end="")


if __name__ == "__main__":
    main()