app_lite.py

from __future__ import annotations

import json

import gradio as gr
from protocol_bench import BASELINES, load_tasks, score, validate_trace

from minicheck import RenderTooLarge, SpecError, check_liveness, check_safety, protocol_from_spec, to_mermaid

TASKS = {t.id: t for t in load_tasks()}

EXAMPLE_BROKEN = {
    "name": "mutex",
    "fields": ["a", "b", "lock"],
    "initial": {"a": 0, "b": 0, "lock": 0},
    "transitions": [
        {"label": "a_enter", "when": {"a": 0}, "set": {"a": 1, "lock": 1}},
        {"label": "b_enter", "when": {"b": 0}, "set": {"b": 1, "lock": 1}},
        {"label": "a_exit", "when": {"a": 1}, "set": {"a": 0, "lock": 0}},
        {"label": "b_exit", "when": {"b": 1}, "set": {"b": 0, "lock": 0}},
    ],
    "invariants": {"mutual_exclusion": {"forbid": {"a": 1, "b": 1}}},
}

EXAMPLE_FIXED = {
    **EXAMPLE_BROKEN,
    "transitions": [
        {"label": "a_enter", "when": {"a": 0, "lock": 0}, "set": {"a": 1, "lock": 1}},
        {"label": "b_enter", "when": {"b": 0, "lock": 0}, "set": {"b": 1, "lock": 1}},
        {"label": "a_exit", "when": {"a": 1}, "set": {"a": 0, "lock": 0}},
        {"label": "b_exit", "when": {"b": 1}, "set": {"b": 0, "lock": 0}},
    ],
}

EXAMPLE_RETRY = {
    "name": "retry_loop",
    "fields": ["tries", "done"],
    "initial": {"tries": 0, "done": 0},
    "transitions": [
        {"label": "attempt", "when": {"done": 0}, "set": {"tries": {"incr": 1}}},
        {"label": "succeed", "when": {"done": 0}, "set": {"done": 1}},
    ],
    "invariants": {"at_most_3_retries": {"forbid": {"tries": 4}}},
    "goal": {"require": {"done": 1}},
}

EXAMPLES = {
    "Mutual exclusion — BROKEN (no lock guard)": EXAMPLE_BROKEN,
    "Mutual exclusion — FIXED": EXAMPLE_FIXED,
    "Retry loop — unbounded retries": EXAMPLE_RETRY,
}


def _fmt_state(state: dict) -> str:
    return ", ".join(f"{k}={v}" for k, v in state.items())


def check_spec(spec_text: str) -> str:
    """Check a user-supplied declarative spec. Returns markdown."""
    try:
        spec = json.loads(spec_text)
    except json.JSONDecodeError as e:
        return f"### ⚠️ Not valid JSON\n\n```\n{e}\n```"

    try:
        model = protocol_from_spec(spec)
    except SpecError as e:
        return f"### ⚠️ Spec rejected\n\n```\n{e}\n```\n\nSee **Spec format** below."

    if not model.invariants:
        return "### ⚠️ No invariants declared\n\nAdd an `invariants` block to check something."

    res = check_safety(model)
    out = [f"**Explored {res['reachable_states']} reachable states.**\n"]
    if not res["exhaustive"]:
        # Say this before any verdict, because it changes how every one of them reads.
        out.append(
            "> ⚠️ **The search did not finish**, so it did not cover the whole state space. "
            "Anything not refuted below is **undetermined**, not proved.\n"
        )
        out.append(f"> _{res.get('incomplete_reason', 'the sweep stopped early')}_\n")

    for name, r in res["properties"].items():
        if r["holds"] is True:
            out.append(f"### ✅ `{name}` HOLDS\n")
            out.append("No reachable state violates it, and every reachable state was checked.\n")
        elif r["holds"] is None:
            # Three-valued: this is NOT a pass, and must not be rendered as one.
            out.append(f"### ❓ `{name}` UNDETERMINED\n")
            out.append(
                "No violation was found, but the search did not complete — so this is not a proof. "
                "Bound the growing field with a `when` guard, then try again.\n"
            )
        else:
            cex = r["counterexample"]
            out.append(f"### ❌ `{name}` VIOLATED in {len(cex) - 1} steps\n")
            out.append("| # | action | state |")
            out.append("|---|---|---|")
            for i, step in enumerate(cex):
                label = step["label"] or "_initial_"
                out.append(f"| {i} | `{label}` | `{_fmt_state(step['state'])}` |")
            out.append("")

    # A picture of the counterexample, when there is one to draw.
    cex = next((r["counterexample"] for r in res["properties"].values() if r["holds"] is False), None)
    if cex is not None:
        try:
            out.append("\n**Counterexample diagram** (Mermaid — paste into GitHub to render):\n")
            out.append("```mermaid\n" + to_mermaid(model, cex, verdict="REFUTED") + "\n```\n")
        except RenderTooLarge:
            out.append("\n_The graph is too large to draw legibly, so no diagram is shown._\n")

    if model.goal is not None:
        live = check_liveness(model)
        if live["holds"] is True:
            out.append("### ✅ Liveness: every reachable state can still reach the goal\n")
        elif live["holds"] is None:
            out.append("### ❓ Liveness: undetermined\n")
        else:
            out.append("### ❌ Liveness: a state exists from which the goal is unreachable\n")
            if live.get("counterexample"):
                trap = live["counterexample"][-1]["state"]
                out.append(f"Trap state: `{_fmt_state(trap)}`\n")

    return "\n".join(out)


def load_example(name: str) -> str:
    return json.dumps(EXAMPLES.get(name, EXAMPLE_BROKEN), indent=2)


def list_benchmark_tasks() -> list[list]:
    """Rows for the benchmark table."""
    return [[t.id, t.standards_body, t.property, t.label, "yes" if t.fixed_builder else "—"] for t in TASKS.values()]


def inspect_task(task_id: str) -> tuple[str, str]:
    """Render a benchmark task: its state machine, and the ground-truth verdict."""
    task = TASKS.get(task_id)
    if task is None:
        return "Unknown task.", ""

    model = task.build()
    lines = [
        f"**{task.id}** — {task.standards_body}",
        f"*{task.spec_clause}*",
        "",
        f"Property: `{task.property}`",
        f"State fields: `{', '.join(model.fields)}`",
        f"Initial: `{_fmt_state(dict(zip(model.fields, model.initial)))}`",
        "",
        "**Transitions**",
        "",
        "| from | action | to |",
        "|---|---|---|",
    ]
    from minicheck._core import _reachable

    _, order = _reachable(model)
    for s in order:
        for label, ns in model.transitions(s):
            lines.append(
                f"| `{_fmt_state(dict(zip(model.fields, s)))}` | `{label}` "
                f"| `{_fmt_state(dict(zip(model.fields, ns)))}` |"
            )

    res = check_safety(model)["properties"][task.property]
    verdict = [f"### Ground truth: `{task.label}`\n"]
    if res["holds"] is True:
        verdict.append("The property **holds** over every reachable state.\n")
    elif res["holds"] is None:
        verdict.append("The property is **undetermined** — the search did not cover the whole space.\n")
    else:
        cex = res["counterexample"]
        verdict.append(f"The property is **violated** in {len(cex) - 1} steps:\n")
        verdict.append("| # | action | state |")
        verdict.append("|---|---|---|")
        for i, step in enumerate(cex):
            verdict.append(f"| {i} | `{step['label'] or '_initial_'}` | `{_fmt_state(step['state'])}` |")
        verdict.append("")
    if task.citation:
        verdict.append(f"**Citation:** {task.citation}")
    if task.fixed_builder:
        fixed = check_safety(task.build_fixed())["properties"][task.property]
        verdict.append(
            f"\n**Repaired twin:** property "
            f"{'holds' if fixed['holds'] is True else ('is undetermined' if fixed['holds'] is None else 'still fails')} — "
            "the counterexample is removable, so it is not a modelling artefact."
        )
    return "\n".join(lines), "\n".join(verdict)


def score_submission(text: str) -> str:
    """Score a pasted submission JSON, replaying every claimed counterexample."""
    if not (text or "").strip():
        return "Paste a submission, or press one of the baseline buttons."
    try:
        submission = json.loads(text)
    except json.JSONDecodeError as e:
        return f"### ⚠️ Not valid JSON\n\n```\n{e}\n```"
    if not isinstance(submission, dict):
        return "### ⚠️ Expected an object mapping task id to a prediction."

    res = score(submission)
    rows = [
        "| metric | value |",
        "|---|---|",
        f"| **balanced accuracy** | **{res['balanced_accuracy']:.3f}** |",
        f"| accuracy | {res['accuracy']:.3f} |",
        f"| _trivial always-safe accuracy_ | _{res['trivial_always_safe_accuracy']:.3f}_ |",
        f"| detections claimed | {res['detections_claimed']} |",
        f"| **counterexamples that replayed** | **{res['valid_counterexamples']}** |",
        f"| TP / FP / FN / TN | {res['true_positives']} / {res['false_positives']} "
        f"/ {res['false_negatives']} / {res['true_negatives']} |",
        "",
    ]
    bad = [r for r in res["per_task"] if r["trace"].get("supplied") and not r["trace"]["valid"]]
    if bad:
        rows.append("**Traces that did not replay**\n")
        rows.append("| task | reason |")
        rows.append("|---|---|")
        for r in bad:
            rows.append(f"| `{r['id']}` | {r['trace']['reason']} |")
    return "\n".join(rows)


def run_baseline(name: str) -> tuple[str, str]:
    """Run a built-in baseline; return (submission json, scorecard)."""
    submission = BASELINES[name]()
    text = json.dumps(submission, indent=2, default=str)
    return text, score_submission(text)


def validate_single_trace(task_id: str, trace_text: str) -> str:
    """Replay one claimed counterexample against its model."""
    task = TASKS.get(task_id)
    if task is None:
        return "Unknown task."
    if not (trace_text or "").strip():
        return 'Paste a trace: a JSON list of `{"state": {...}}` entries.'
    try:
        trace = json.loads(trace_text)
    except json.JSONDecodeError as e:
        return f"### ⚠️ Not valid JSON\n\n```\n{e}\n```"

    v = validate_trace(task, trace)
    if v["valid"]:
        return (
            f"### ✅ Replayed against the model\n\n{v['length']} steps. It starts at the initial "
            "state, every step is a real transition, and the final state violates the property."
        )
    return f"### ❌ Did not replay\n\n**Reason:** {v['reason']}"


INTRO = """
# Protocol-Bench & minicheck

**Check a state machine for a safety violation and get the shortest sequence of steps that breaks
it.** Or score a submission against 15 published IEEE 802.11 / 3GPP procedures — where a claimed
detection has to *replay* before it counts.

This runs a real model checker **entirely in your browser** (Pyodide). Nothing is pre-computed and
nothing is sent to a server.
"""

SPEC_FORMAT = """
### Spec format

- `when` — a conjunction of `field == value` tests. Omit for an always-enabled transition.
- `set` — assign a literal, or `{"incr": n}` / `{"decr": n}` for integers.
- `invariants` — `{"forbid": {...}}` fails when **every** listed field matches;
  `{"require": {...}}` fails unless they do.
- `goal` — optional, same shape, used for the liveness check.

Specs are **data, never code**: nothing you paste is executed.
"""

with gr.Blocks(title="Protocol-Bench & minicheck") as demo:
    gr.Markdown(INTRO)

    with gr.Tab("Check a state machine"):
        example_pick = gr.Dropdown(choices=list(EXAMPLES), value=list(EXAMPLES)[0], label="Load an example")
        spec_box = gr.Code(value=load_example(list(EXAMPLES)[0]), language="json", label="Spec", lines=18)
        check_btn = gr.Button("Check", variant="primary")
        spec_out = gr.Markdown()
        gr.Markdown(SPEC_FORMAT)
        example_pick.change(load_example, example_pick, spec_box)
        check_btn.click(check_spec, spec_box, spec_out)

    with gr.Tab("Browse the benchmark"):
        gr.Markdown("15 published procedures. 13 hold; 2 do not.")
        gr.Dataframe(
            value=list_benchmark_tasks(),
            headers=["task", "body", "property", "label", "repaired twin"],
            interactive=False,
            wrap=True,
        )
        task_pick = gr.Dropdown(choices=sorted(TASKS), value="ieee_4way_handshake_krack", label="Inspect a task")
        task_model = gr.Markdown()
        task_verdict = gr.Markdown()
        task_pick.change(inspect_task, task_pick, [task_model, task_verdict])

    with gr.Tab("Score a submission"):
        gr.Markdown(
            "Every trace you supply is **replayed against the model**. Try the baselines: "
            "`always-safe` scores 0.867 plain accuracy and 0.500 balanced; `always-violated` "
            "claims 15 detections and validates **zero**."
        )
        with gr.Row():
            b_bfs = gr.Button("baseline: bfs")
            b_safe = gr.Button("baseline: always-safe")
            b_viol = gr.Button("baseline: always-violated")
        sub_box = gr.Code(language="json", label="Submission", lines=14)
        score_btn = gr.Button("Score", variant="primary")
        sub_out = gr.Markdown()
        b_bfs.click(lambda: run_baseline("bfs"), None, [sub_box, sub_out])
        b_safe.click(lambda: run_baseline("always-safe"), None, [sub_box, sub_out])
        b_viol.click(lambda: run_baseline("always-violated"), None, [sub_box, sub_out])
        score_btn.click(score_submission, sub_box, sub_out)

    with gr.Tab("Replay one trace"):
        trace_task = gr.Dropdown(choices=sorted(TASKS), value="ieee_4way_handshake_krack", label="Task")
        trace_box = gr.Code(
            language="json",
            label="Trace",
            lines=10,
            value='[{"state": {"ptk_installed": false, "tx_nonce": 0, "nonce_reused": false}}]',
        )
        trace_btn = gr.Button("Replay", variant="primary")
        trace_out = gr.Markdown()
        trace_btn.click(validate_single_trace, [trace_task, trace_box], trace_out)

    gr.Markdown(
        "---\n"
        "**A verdict you cannot check is not a verdict** — and *undetermined is not a pass*. "
        "This demo is one piece of a larger set.\n\n"
        "[Documentation](https://nickharris808.github.io/verification-docs/) · "
        "[minicheck](https://github.com/nickharris808/minicheck) · "
        "[protocol-bench](https://github.com/nickharris808/protocol-bench) · "
        "[specforge](https://github.com/nickharris808/specforge) · "
        "[the leaderboard](https://huggingface.co/spaces/nickh007/specforge-leaderboard)"
    )

demo.launch()