Apalache RPC Client

Minimalistic Python client for interaction with the Apalache model checker over JSON RPC.

Why? Because now you can write your own test harnesses and symbolic search tools that interact with TLA⁺ and Quint specifications. Use it with an AI agent, and you have got superpowers!

Minimalistic. Connections usually require tinkering with the parameters. This library only provides a thin wrapper around the JSON-RPC calls to the Apalache server. Start with this library, to get the initial setup working. Once, you have hit the limits, just fork this project and extend it to your needs.

Compression. The client compresses request bodies with gzip when they exceed 512 bytes and requests gzip-compressed responses from the server. This is especially beneficial for loadSpec (base64-encoded source files) and query (full state/trace responses). Compression is enabled by default and can be disabled with JsonRpcClient(compression=False).

Server API. Refer to the Apalache JSON-RPC API for the interaction details.

Installation

You can install the package via pip:

pip install apalache-rpc-client

Make sure you have Apalache installed. Follow the Apalache Installation Instructions.

Usage

Install Apalache

# Install Java if you don't have it yet, e.g., on macOS:
brew install temurin@21
# Download Apalache if you don't have it yet
curl -L -o apalache.tgz https://github.com/apalache-mc/apalache/releases/latest/download/apalache.tgz
tar -xzf apalache.tgz
export APALACHE_HOME="$PWD/apalache"

Start and Stop the Apalache Server

import logging
import sys
import time
from tempfile import TemporaryDirectory

from apalache_rpc.server import ApalacheServer

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[ logging.StreamHandler(sys.stdout) ]
)

with TemporaryDirectory() as log_dir:
    server = ApalacheServer(log_dir=log_dir, hostname="localhost", port=18080)
    assert server.start_server(), "Failed to start the Apalache server"
    time.sleep(5)  # Let the server run for a bit
    assert server.stop_server(), "Failed to stop the Apalache server"

Call JSON-RPC Methods on the Circular Buffer Example

This example loads examples/circular-buffer/MC10u8_BuggyCircularBuffer.tla and the imported base module, chooses an initial state, queries an exported operator, checks invariants, explores a Put transition, compacts the current symbolic trace, and rolls back to the initial snapshot.

from pathlib import Path
from tempfile import TemporaryDirectory

from apalache_rpc.client import (
    InvariantSatisfied,
    JsonRpcClient,
    NextModelFalse,
    NextModelTrue,
    TransitionEnabled,
)
from apalache_rpc.server import ApalacheServer

repo_root = Path.cwd()
if not (repo_root / "examples").exists():
    repo_root = repo_root.parent

examples_dir = repo_root / "examples" / "circular-buffer"
sources = [
    str(examples_dir / "MC10u8_BuggyCircularBuffer.tla"),
    str(examples_dir / "BuggyCircularBuffer.tla"),
]

with TemporaryDirectory() as log_dir:
    server = ApalacheServer(log_dir=log_dir, hostname="localhost", port=18081)
    assert server.start_server(), "Failed to start the Apalache server"

    try:
        with JsonRpcClient(port=18081, solver_timeout=30) as client:
            # Call loadSpec to register the TLA+ sources and named operators.
            spec = client.load_spec(
                sources=sources,
                init="Init",
                next="Next",
                invariants=["SafeInv"],
                view="CountView",
            )
            assert spec is not None
            assert spec["next"] == [
                {"index": 0, "labels": ["Put"]},
                {"index": 1, "labels": ["Get"]},
            ]

            # Call assumeTransition on the single initial transition.
            init_status = client.assume_transition(spec["init"][0]["index"])
            assert isinstance(init_status, TransitionEnabled)

            # Call nextStep to advance to the next frame (symbolic state).
            # This does not fix any initial state!
            init_snapshot = client.next_step()

            # Call query with OPERATOR to evaluate the exported CountView operator.
            count_view = client.query(["OPERATOR"], operator="CountView")
            assert count_view == {"operatorValue": {"#tup": [{"#bigint": "0"}]}}

            # Call checkInvariant for each registered invariant through check_invariants.
            inv_status = client.check_invariants(
                len(spec["state"]),
                len(spec["action"]),
            )
            assert isinstance(inv_status, InvariantSatisfied)

            # Call assumeTransition on the Put action from the current state.
            put_status = client.assume_transition(spec["next"][0]["index"])
            assert isinstance(put_status, TransitionEnabled)

            # Call compact to collapse the current symbolic trace into one concrete state.
            compacted_snapshot = client.compact(init_snapshot)
            assert isinstance(compacted_snapshot, int)

            # Call nextModel to ask whether CountView can change in the next state.
            next_model = client.next_model("CountView")
            assert next_model["oldValue"] == {"#tup": [{"#bigint": "1"}]}
            assert isinstance(next_model["hasOld"], NextModelTrue)
            assert isinstance(next_model["hasNext"], NextModelFalse)

            # Call rollback to return to the saved initial snapshot.
            client.rollback(init_snapshot)

            # Call query again to confirm the rollback restored the old view.
            assert client.query(["OPERATOR"], operator="CountView") == count_view

            # Call disposeSpec to release the server-side session explicitly.
            client.dispose_spec()
    finally:
        assert server.stop_server(), "Failed to stop the Apalache server"