gen_tlc_model.py

"""
gen_tlc_model.py — generate a TLC model config from pact's live _CORO_CONSUMERS.

Reads _CORO_CONSUMERS and _CORO_CONSUMERS_QUALIFIED out of failure_mode.py via
AST (no import needed), then writes a MissingAwait.cfg that instantiates
MissingAwait.tla with real data from the codebase.

Usage:
    python3 gen_tlc_model.py [--out docs/tla/MissingAwait.cfg]

Then verify with:
    java -XX:+UseParallelGC -jar ~/.local/share/tla2tools.jar \
        -config docs/tla/MissingAwait.cfg -deadlock docs/tla/MissingAwait

The generated config proves that every name in _CORO_CONSUMERS satisfies
the NoFalsePositive and ConsumedSitesPermanentlyClean invariants — i.e.,
adding a name to that frozenset is sufficient to suppress the violation.
"""

from __future__ import annotations

import argparse
import ast
import sys
from pathlib import Path

# ---------------------------------------------------------------------------
# Extract frozenset literals from failure_mode.py without importing it
# ---------------------------------------------------------------------------


def _extract_frozenset(source: str, varname: str) -> list[str]:
    """Return the string elements of the first frozenset assigned to varname."""
    tree = ast.parse(source)
    for node in ast.walk(tree):
        if not isinstance(node, ast.Assign):
            continue
        for target in node.targets:
            if isinstance(target, ast.Name) and target.id == varname:
                val = node.value
                # frozenset({...}) or frozenset({"a", "b"})
                if isinstance(val, ast.Call):
                    args = val.args
                    if args and isinstance(args[0], ast.Set):
                        return sorted(
                            elt.value
                            for elt in args[0].elts
                            if isinstance(elt, ast.Constant)
                            and isinstance(elt.value, str)
                        )
    return []


def _extract_qualified(source: str, varname: str) -> list[tuple[str, str]]:
    """Return (receiver, name) pairs from a frozenset of 2-tuples."""
    tree = ast.parse(source)
    for node in ast.walk(tree):
        if not isinstance(node, ast.Assign):
            continue
        for target in node.targets:
            if isinstance(target, ast.Name) and target.id == varname:
                val = node.value
                if isinstance(val, ast.Call):
                    args = val.args
                    if args and isinstance(args[0], ast.Set):
                        pairs = []
                        for elt in args[0].elts:
                            if (
                                isinstance(elt, ast.Tuple)
                                and len(elt.elts) == 2
                                and all(
                                    isinstance(e, ast.Constant)
                                    and isinstance(e.value, str)
                                    for e in elt.elts
                                )
                            ):
                                pairs.append((elt.elts[0].value, elt.elts[1].value))
                        return sorted(pairs)
    return []


# ---------------------------------------------------------------------------
# Generate TLC config
# ---------------------------------------------------------------------------

_HEADER = """\
\\* MissingAwait.cfg — AUTO-GENERATED by gen_tlc_model.py
\\* Do not edit by hand; regenerate with:
\\*   python3 gen_tlc_model.py
\\*
\\* Instantiates MissingAwait.tla with pact's live _CORO_CONSUMERS frozenset.
\\* Run TLC to verify NoFalsePositive and ConsumedSitesPermanentlyClean hold
\\* for every name currently in the frozenset.
\\*
\\* Usage:
\\*   java -XX:+UseParallelGC -jar ~/.local/share/tla2tools.jar \\
\\*        -config MissingAwait.cfg -deadlock MissingAwait

SPECIFICATION Spec

"""

_INVARIANTS = """\
INVARIANTS
    TypeInvariant
    NoFalsePositive

PROPERTIES
    ConsumedSitesPermanentlyClean
    MonotonicViolations
    EventuallyFlagged
    CompletionCorrectness
"""


def _tla_set(items: list[str]) -> str:
    if not items:
        return "{}"
    quoted = ", ".join(f'"{s}"' for s in items)
    return "{" + quoted + "}"


def generate_cfg(
    consumers: list[str],
    qualified: list[tuple[str, str]],
    max_consumed: int = 5,
) -> str:
    # TLC state space is 2^|Sites|, so we use a representative sample.
    # The abstract proof in MissingAwait.tla already holds for any instantiation;
    # this config gives a concrete fast-running CI check.
    all_consumed = consumers + [f"{r}__{n}" for r, n in qualified]
    consumed_sites = all_consumed[:max_consumed]
    sites = consumed_sites + ["__real_bug__"]
    awaitable_sites = consumed_sites + ["__real_bug__"]

    total_consumers = len(consumers) + len(qualified)
    sample_note = (
        f"\\* Sampling {len(consumed_sites)} of {total_consumers} consumers "
        f"(TLC state space = 2^|Sites|; abstract proof covers all).\n"
    )
    lines = [_HEADER, sample_note, "\nCONSTANTS\n"]
    lines.append(f"    Sites          = {_tla_set(sites)}\n")
    lines.append(f"    ConsumedSites  = {_tla_set(consumed_sites)}\n")
    lines.append(f"    AwaitableSites = {_tla_set(awaitable_sites)}\n")
    lines.append("\n")
    lines.append(_INVARIANTS)
    return "".join(lines)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def main(argv=None) -> int:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument(
        "--failure-mode",
        default="failure_mode.py",
        metavar="FILE",
        help="Path to failure_mode.py (default: ./failure_mode.py)",
    )
    p.add_argument(
        "--out",
        default="docs/tla/MissingAwait.cfg",
        metavar="FILE",
        help="Output path for generated TLC config (default: docs/tla/MissingAwait.cfg)",
    )
    p.add_argument(
        "--max-consumed",
        type=int,
        default=5,
        metavar="N",
        help="Max consumers to include as sites (default: 5, keeps TLC state space tractable)",
    )
    args = p.parse_args(argv)

    src_path = Path(args.failure_mode)
    if not src_path.exists():
        print(f"error: {src_path} not found", file=sys.stderr)
        return 1

    source = src_path.read_text()
    consumers = _extract_frozenset(source, "_CORO_CONSUMERS")
    qualified = _extract_qualified(source, "_CORO_CONSUMERS_QUALIFIED")

    if not consumers and not qualified:
        print("error: could not find _CORO_CONSUMERS in source", file=sys.stderr)
        return 1

    print(
        f"Extracted {len(consumers)} unqualified + {len(qualified)} qualified consumers"
    )
    for c in consumers:
        print(f"  {c}")
    for r, n in qualified:
        print(f"  {r}.{n}  (qualified)")

    cfg = generate_cfg(consumers, qualified, max_consumed=args.max_consumed)
    out_path = Path(args.out)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(cfg)
    sample = min(args.max_consumed, len(consumers) + len(qualified))
    print(f"\nWrote {out_path}")
    print(f"Config: {sample} sampled consumers + 1 real bug = {sample + 1} sites")
    print(
        f"(Full set: {len(consumers) + len(qualified)} consumers; abstract proof covers all)"
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())