From e58c2849a784f5a2f9f364e3260982babfcac772 Mon Sep 17 00:00:00 2001 From: Noam Cohen Date: Thu, 17 Sep 2026 10:40:19 +0200 Subject: [PATCH 1/2] Use the published Kepler Formal Python library in MCP tools --- .github/workflows/tests.yml | 40 ++++ .gitignore | 7 + .gitmodules | 3 - README.md | 76 ++++++-- build_kepler_formal.sh | 97 ---------- docs/instructions-claude.md | 75 +++----- docs/test-generation.md | 86 ++++----- kepler_formal_mcp/__init__.py | 1 + kepler_formal_mcp/__main__.py | 3 + kepler_formal_mcp/config.py | 117 +++++++++++ kepler_formal_mcp/runner.py | 65 +++++++ kepler_formal_mcp/server.py | 106 ++++++++++ kepler_formal_mcp/worker.py | 94 +++++++++ pyproject.toml | 24 +++ server.py | 352 +--------------------------------- tests/test_server.py | 297 ++++++++++++++++++++++++++++ thirdparty/kepler-formal | 1 - 17 files changed, 877 insertions(+), 567 deletions(-) create mode 100644 .github/workflows/tests.yml create mode 100644 .gitignore delete mode 100644 .gitmodules delete mode 100755 build_kepler_formal.sh create mode 100644 kepler_formal_mcp/__init__.py create mode 100644 kepler_formal_mcp/__main__.py create mode 100644 kepler_formal_mcp/config.py create mode 100644 kepler_formal_mcp/runner.py create mode 100644 kepler_formal_mcp/server.py create mode 100644 kepler_formal_mcp/worker.py create mode 100644 pyproject.toml create mode 100644 tests/test_server.py delete mode 160000 thirdparty/kepler-formal diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..583e7cd --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,40 @@ +name: Python integration + +on: + push: + branches: [main] + pull_request: + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: python-integration-${{ github.ref }} + cancel-in-progress: ${{ github.event_name == 'pull_request' }} + +jobs: + test: + name: ${{ matrix.os }} / Python ${{ matrix.python }} + runs-on: ${{ matrix.os }} + timeout-minutes: 15 + strategy: + fail-fast: false + matrix: + os: [ubuntu-24.04, macos-latest, windows-latest] + python: ['3.10', '3.14'] + steps: + - uses: actions/checkout@v4 + with: + persist-credentials: false + - uses: actions/setup-python@v6 + with: + python-version: ${{ matrix.python }} + cache: pip + cache-dependency-path: pyproject.toml + - name: Install with published Kepler Formal and NajaEDA wheels + run: python -m pip install --only-binary=kepler-formal,najaeda . + - name: Validate dependency versions + run: python -m pip check + - name: Test Python verification and MCP transport + run: python -m unittest discover -s tests -v diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..cfdb9f9 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +.venv/ +__pycache__/ +*.py[cod] +*.egg-info/ +build/ +dist/ +*.log diff --git a/.gitmodules b/.gitmodules deleted file mode 100644 index 86252bd..0000000 --- a/.gitmodules +++ /dev/null @@ -1,3 +0,0 @@ -[submodule "thirdparty/kepler-formal"] - path = thirdparty/kepler-formal - url = https://github.com/keplertech/kepler-formal.git diff --git a/README.md b/README.md index c6d9736..4577aa0 100644 --- a/README.md +++ b/README.md @@ -1,46 +1,80 @@ # kepler-formal-mcp -## Usage +MCP tools for checking whether two Verilog designs are equivalent. NajaEDA loads the designs, then the published **Kepler Formal Python library** verifies them using the same native Naja runtime. -To use this MCP, simply: +## Install -1. Clone the repository with submodules, or initialize the submodule afterward. -2. Run the build script: +Use CPython 3.10–3.14 on Linux x86_64/aarch64, macOS ARM64, or Windows AMD64, where the pinned dependencies provide wheels. ```bash -./build_kepler_formal.sh +git clone https://github.com/keplertech/kepler-formal-mcp.git +cd kepler-formal-mcp +python3 -m venv .venv +source .venv/bin/activate +python -m pip install --only-binary=kepler-formal,najaeda . ``` -If you also want the system dependencies to be installed automatically, run: +On Windows, create and activate the environment with PowerShell: + +```powershell +py -3 -m venv .venv +.venv\Scripts\Activate.ps1 +python -m pip install --only-binary=kepler-formal,najaeda . +``` + +Installing this project installs `kepler-formal==0.5.0`, which requires `najaeda==0.7.24`, plus the MCP SDK and PyYAML. No Nix, submodule checkout, C++ build, or system compiler setup is needed. + +Start the stdio MCP server with: ```bash -./build_kepler_formal.sh --install-deps +kepler-formal-mcp ``` -This script will fetch `kepler-formal`, initialize its submodules, and then build the binary in `thirdparty/kepler-formal/build/`. +The existing `python server.py` launcher also works using the installed environment. For desktop clients, use the absolute path to `.venv/bin/kepler-formal-mcp` (Windows: `.venv\Scripts\kepler-formal-mcp.exe`). See [Claude Desktop configuration](docs/instructions-claude.md). -The repository now includes `kepler-formal` as a proper Git submodule in `thirdparty/kepler-formal`. +## Tools -After that, `server.py` automatically looks for the generated binary in the correct directory, so there is no extra manual step to locate it. +- `run_kepler_formal_yaml`: verify the pair specified by an existing YAML file. +- `create_yaml_and_run_kepler_formal`: create a YAML file from two Verilog paths and optional common Liberty libraries, then verify the pair. -## Dependencies +Both tools use the Python library in a separate Python worker for each call. This keeps native solver output out of the MCP stdio channel and allows `timeout_seconds` to stop a verification. The worker loads both designs with NajaEDA and passes their live design handles to `kepler_formal.verify_designs`. -On Linux Ubuntu/Debian: +A minimal YAML file is: -```bash -sudo apt-get install g++ libboost-dev python3.9-dev capnproto libcapnp-dev libtbb-dev pkg-config bison flex doxygen libspdlog-dev libfmt-dev libboost-iostreams-dev zlib1g-dev +```yaml +format: verilog +input_paths: + - reference.v + - candidate.v +liberty_files: [] +verification: lec +solver: kissat +cnf_export: false ``` -On Fedora: +Exactly two structural Verilog netlist files are required. Synthesize behavioral RTL before loading it with NajaEDA. Paths inside an existing YAML file are resolved relative to that file; paths passed to the create tool are resolved relative to the server's launch directory. The create tool writes absolute design/library paths into its generated YAML. -```bash -sudo dnf install gcc-c++ boost-devel python3-devel capnproto capnproto-devel tbb-devel pkgconf-pkg-config bison flex doxygen spdlog-devel fmt-devel boost-iostreams-devel zlib-devel cmake git -``` +Set `KEPLER_FORMAL_AI_OUTPUT_DIR` to a writable output directory, or pass `allowed_output_dir` to a tool. The default is the server's launch directory. Generated YAML and logs must stay inside the selected output directory. An existing YAML file is read without rewriting it. + +Supported verification settings are `verification` (`lec` or `sec`; `mode` is an alias), `solver`, `max_k`, `sec_engine`, `sec_encoding`, `allow_boundary_mismatch`, `report_skipped_outputs`, `log_file`, and `log_level`. CNF export is unavailable through this library API: `cnf_export` defaults to `false`, and `true` returns a clear error. Unsupported CLI-only YAML options also return an error. + +## Results + +The returned JSON separates tool execution from the verification verdict: + +- `status`: `success` when verification ran, or `error` for invalid input, timeout, or a worker failure. +- `verdict`: the equivalence outcome, such as `equivalent`, `different`, or `inconclusive`. +- `verification_result`: the structured result returned by Kepler Formal. +- `stdout_tail` / `stderr_tail`: captured worker output for diagnosis. + +Always inspect `verdict`: a successful execution can report different designs. A bounded SEC check can be inconclusive; it is not an equivalence proof. + +See [the small design example](docs/test-generation.md) for equivalent and different inputs. -On macOS with Homebrew: +## Test ```bash -brew install cmake doxygen capnp tbb bison flex boost spdlog zlib +python -m unittest discover -s tests -v ``` -On Windows, the easiest approach is to use WSL2 or MSYS2 with a compatible Bash environment. Automatic installation of system dependencies is not supported on native Windows. \ No newline at end of file +CI runs the tests against the published dependencies on Linux x86_64, macOS ARM64, and Windows AMD64 with Python 3.10 and 3.14. Tests cover the real Python verifier, configuration validation, worker isolation, and MCP stdio calls. diff --git a/build_kepler_formal.sh b/build_kepler_formal.sh deleted file mode 100755 index 8e4dd63..0000000 --- a/build_kepler_formal.sh +++ /dev/null @@ -1,97 +0,0 @@ -#!/usr/bin/env bash - -set -euo pipefail - -SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" -ROOT_DIR="$SCRIPT_DIR" -REPO_DIR="$ROOT_DIR/thirdparty/kepler-formal" -BUILD_DIR="$REPO_DIR/build" -INSTALL_DEPS=0 - -usage() { - cat <<'EOF' -Usage: build_kepler_formal.sh [--install-deps] - - --install-deps Install system dependencies for the current OS before building. -EOF -} - -install_dependencies() { - case "$(uname -s)" in - Linux) - if command -v apt-get >/dev/null 2>&1; then - sudo apt-get update - sudo apt-get install -y \ - g++ libboost-dev python3.9-dev capnproto libcapnp-dev libtbb-dev \ - pkg-config bison flex doxygen libspdlog-dev libfmt-dev \ - libboost-iostreams-dev zlib1g-dev cmake git - elif command -v dnf >/dev/null 2>&1; then - sudo dnf install -y \ - gcc-c++ boost-devel python3-devel capnproto capnproto-devel \ - tbb-devel pkgconf-pkg-config bison flex doxygen spdlog-devel \ - fmt-devel boost-iostreams-devel zlib-devel cmake git - else - echo "Unsupported Linux package manager. Install the dependencies manually." >&2 - exit 1 - fi - ;; - Darwin) - if ! command -v brew >/dev/null 2>&1; then - echo "Homebrew is required on macOS to install dependencies." >&2 - exit 1 - fi - brew install cmake doxygen capnp tbb bison flex boost spdlog zlib - ;; - *) - echo "Automatic dependency installation is not configured for $(uname -s)." >&2 - exit 1 - ;; - esac -} - -while [[ $# -gt 0 ]]; do - case "$1" in - --install-deps) - INSTALL_DEPS=1 - shift - ;; - -h|--help) - usage - exit 0 - ;; - *) - echo "Unknown argument: $1" >&2 - usage >&2 - exit 1 - ;; - esac -done - -if [[ "$INSTALL_DEPS" -eq 1 ]]; then - install_dependencies -fi - -cpu_count() { - if command -v nproc >/dev/null 2>&1; then - nproc - elif command -v sysctl >/dev/null 2>&1; then - sysctl -n hw.ncpu - else - echo 1 - fi -} - -git -C "$ROOT_DIR" submodule update --init --recursive -- thirdparty/kepler-formal - -mkdir -p "$BUILD_DIR" -cd "$BUILD_DIR" - -cmake .. \ - -DCMAKE_BUILD_TYPE=Release \ - -DCMAKE_CXX_STANDARD=20 \ - -DCMAKE_CXX_FLAGS="-O3 -march=native -ffast-math -flto -DNDEBUG" \ - -DCMAKE_CXX_FLAGS_RELEASE="-Ofast -march=native -ffast-math -flto -DNDEBUG" \ - -DCMAKE_EXE_LINKER_FLAGS="-flto" -cmake --build . --parallel "$(cpu_count)" - -echo "Build complete: $BUILD_DIR" \ No newline at end of file diff --git a/docs/instructions-claude.md b/docs/instructions-claude.md index 7f2bbcd..0ee93aa 100644 --- a/docs/instructions-claude.md +++ b/docs/instructions-claude.md @@ -1,76 +1,45 @@ # Configuring Kepler Formal MCP in Claude Desktop -This guide explains how to add the Kepler Formal MCP server to Claude Desktop. +First install the project in a virtual environment using the [README](../README.md). The environment includes the published Kepler Formal and NajaEDA packages. -**Note:** This guide assumes you have already installed Kepler Formal MCP. See the main README and build script for installation instructions. - -## Configuration - -### Step 1: Locate Your Configuration File - -Find your Claude Desktop configuration file: -- **Linux/Mac**: `~/.config/Claude/claude_desktop_config.json` -- **Windows**: `%APPDATA%\Claude\claude_desktop_config.json` - -### Step 2: Add Kepler Formal to Your MCP Servers - -Edit your configuration file and add the Kepler Formal server. Replace the placeholders with your actual paths: -- ``: A name for this server (e.g., `kepler`, `kepler-formal`, `formal-verification`) -- ``: Full path to your Kepler Formal MCP repository -- ``: Path to the Kepler Formal source code (usually `/thirdparty/kepler-formal/src`) +Add this entry to your Claude Desktop MCP configuration, replacing both paths with absolute paths on your machine: ```json { "mcpServers": { - "": { - "command": "python3", - "args": [ - "/server.py" - ], + "kepler-formal": { + "command": "/absolute/path/to/kepler-formal-mcp/.venv/bin/kepler-formal-mcp", "env": { - "PYTHONPATH": "" + "KEPLER_FORMAL_AI_OUTPUT_DIR": "/absolute/path/to/verification-output" } } } } ``` -### Step 3: Verify - -1. Save the configuration file -2. Restart Claude Desktop -3. Check that your MCP server appears as "Connected" in Claude's MCP server list - -## Strongly Recommended: Add a Shared Folder - -Adding a shared folder is strongly recommended. Without it, you'll need to copy-paste potentially large files directly into Claude's prompt, which is inefficient and can hit token limits. - -With the filesystem MCP server, Claude can access your files directly: +On Windows, use the virtual environment's executable and JSON-escaped backslashes: ```json -"filesystem": { - "command": "npx", - "args": [ - "-y", - "@modelcontextprotocol/server-filesystem", - "" - ] +{ + "mcpServers": { + "kepler-formal": { + "command": "C:\\absolute\\path\\kepler-formal-mcp\\.venv\\Scripts\\kepler-formal-mcp.exe", + "env": { + "KEPLER_FORMAL_AI_OUTPUT_DIR": "C:\\absolute\\path\\verification-output" + } + } + } } ``` -Replace `` with an absolute path to a folder where you want to store files accessible to Claude. This allows Claude to read large design files, test vectors, and documentation without copy-pasting. +No `PYTHONPATH` or Kepler Formal source directory is required. The executable uses its virtual environment even when Claude starts from another directory. Save the configuration and restart Claude Desktop. + +Use absolute design and library paths in tool calls because desktop clients may launch servers from an unexpected directory. The output environment variable controls where generated YAML and logs can be written; each call can override it with `allowed_output_dir`. -## Troubleshooting +For example: -**Server won't connect:** -- Verify paths are absolute (not relative) -- Restart Claude Desktop after saving the config -- Ensure the config file is valid JSON -- Check that `/server.py` exists +> Use `create_yaml_and_run_kepler_formal` to compare `/my/designs/reference.v` and `/my/designs/candidate.v`, with no Liberty libraries. Report the verification verdict. -**"Command not found" for python3:** -- Use the full path to Python: `/usr/bin/python3` instead of `python3` +For an existing YAML file, paths inside it are relative to its directory. Ask Claude to use `run_kepler_formal_yaml` with the YAML file's absolute path. The MCP reads the designs directly; a filesystem MCP is only needed if you also want Claude to create or edit the design files. -**Invalid JSON errors:** -- Use a JSON validator to check your config file -- Ensure all commas and quotes are correct +If connection fails, check that the configured executable exists and that the environment was installed successfully with `python -m pip check`. Run the executable in a terminal to inspect errors on stderr. It normally waits silently for MCP messages on stdin. diff --git a/docs/test-generation.md b/docs/test-generation.md index 9048e2e..6b7c47f 100644 --- a/docs/test-generation.md +++ b/docs/test-generation.md @@ -1,60 +1,60 @@ -# Using Kepler Formal MCP for Design Comparison +# A small equivalence example -This guide explains how to use the Kepler Formal MCP to analyze and compare Verilog designs with their libraries. +Create these three files in a design directory. They use only Verilog assignments, so no Liberty files are needed. -## Overview +`reference.v`: -The [kepler-formal-regress](https://github.com/keplertech/kepler-formal-regress.git) repository contains ready-to-use design examples with their libraries: -- **black_parrot**: A complete processor design with all necessary files -- **tinyrocket**: A simple RISC-V processor design with all necessary files - -Each directory contains `.v` (Verilog) files and `.lib` (library) files that you can use immediately. - -## Step 1: Clone the Kepler Formal Regress Repository - -```bash -git clone https://github.com/keplertech/kepler-formal-regress.git +```verilog +module reference(input a, output y); + assign y = a; +endmodule ``` -## Step 2: Copy Design Folder to Your Shared Folder +`equivalent.v`: -Copy an entire design directory (black_parrot or tinyrocket) to your shared folder: - -```bash -# Choose one design -cp -r /black_parrot / -# or -cp -r /tinyrocket / +```verilog +module candidate(input a, output y); + wire connection; + assign connection = a; + assign y = connection; +endmodule ``` -Your shared folder now contains all Verilog files and libraries needed for analysis. - -## Step 3: Generate Design Variants +`different.v`: -Each design folder may contain a Python script that generates modified versions of the `.v` files (with intentional changes for testing): - -```bash -cd /black_parrot -python3 black_parrot_edit.py +```verilog +module candidate(input a, output y); + assign y = 1'b0; +endmodule ``` -This creates modified versions of the design files that you can compare with the originals. - -## Step 4: Use Kepler Formal MCP in Claude +After [configuring your MCP client](instructions-claude.md), call `create_yaml_and_run_kepler_formal` with: -After setting up Claude and configuring the shared folder path in Claude, you can ask it to compare the two versions using the Kepler Formal MCP tools: +```json +{ + "input_paths": ["/absolute/path/reference.v", "/absolute/path/equivalent.v"], + "liberty_files": [], + "allowed_output_dir": "/absolute/path/verification-output", + "yaml_output_path": "equivalent.yaml", + "verification": "lec" +} +``` -### Example: Compare Two Versions -"Use the Kepler Formal MCP tools to compare `tinyrocket.v` with `tinyrocket_modified.v` and identify all differences" +The result should have `status: "success"` and `verdict: "equivalent"`. Change the second input to `different.v` and the YAML output name to `different.yaml`: execution should still succeed, with `verdict: "different"`. -or +You can also create a YAML file beside the designs and pass its path to `run_kepler_formal_yaml`: -"Load both versions of black_parrot with their libraries and perform a formal verification comparison to find the differences" +```yaml +format: verilog +input_paths: + - reference.v + - equivalent.v +liberty_files: [] +verification: lec +solver: kissat +cnf_export: false +``` -**Important:** Claude should use the MCP tools directly to analyze the files. Do not ask Claude to read files manually - let the MCP handle the analysis. +Pass `allowed_output_dir` to choose a writable log directory. Relative input and library paths in an existing YAML file are resolved from that file's directory. The input YAML is not rewritten. -The Kepler Formal MCP will: -- Read both `.v` files from your shared folder -- Load the corresponding `.lib` library files -- Perform formal verification analysis -- Report all logic differences between the two versions +For cell-based netlists, supply the common `.lib` files in `liberty_files`. For sequential checking, use `verification: sec` with the supported `max_k`, `sec_engine`, and `sec_encoding` settings. Inspect the returned verdict and counterexample information: `inconclusive` does not mean equivalent, and a reported counterexample does not enumerate every possible difference. diff --git a/kepler_formal_mcp/__init__.py b/kepler_formal_mcp/__init__.py new file mode 100644 index 0000000..67c0119 --- /dev/null +++ b/kepler_formal_mcp/__init__.py @@ -0,0 +1 @@ +"""MCP integration for the Kepler Formal Python library.""" diff --git a/kepler_formal_mcp/__main__.py b/kepler_formal_mcp/__main__.py new file mode 100644 index 0000000..bc40eef --- /dev/null +++ b/kepler_formal_mcp/__main__.py @@ -0,0 +1,3 @@ +from .server import main + +main() diff --git a/kepler_formal_mcp/config.py b/kepler_formal_mcp/config.py new file mode 100644 index 0000000..a61ef07 --- /dev/null +++ b/kepler_formal_mcp/config.py @@ -0,0 +1,117 @@ +"""Translate file-based MCP requests into Python verification options.""" + +from __future__ import annotations + +import os +from pathlib import Path +from typing import Any + +import yaml + + +SUPPORTED_KEYS = { + "format", "input_paths", "liberty_files", "verification", "mode", "solver", + "max_k", "sec_engine", "sec_encoding", "allow_boundary_mismatch", + "report_skipped_outputs", "log_file", "log_level", "cnf_export", "cnf_export_path", +} + + +def resolve_path(value: str, base: Path) -> Path: + if not isinstance(value, str) or not value.strip(): + raise ValueError("Paths must be nonempty strings") + path = Path(value).expanduser() + return (path if path.is_absolute() else base / path).resolve() + + +def output_root(override: str | None = None) -> Path: + return resolve_path( + override or os.environ.get("KEPLER_FORMAL_AI_OUTPUT_DIR") or str(Path.cwd()), + Path.cwd(), + ) + + +def allowed_path(path: Path, root: Path) -> Path: + path = path.resolve() + if not path.is_relative_to(root.resolve()): + raise ValueError(f"Path not allowed outside output directory {root}: {path}") + return path + + +def load_yaml(path: Path) -> dict[str, Any]: + document = yaml.safe_load(path.read_text(encoding="utf-8")) + if not isinstance(document, dict): + raise ValueError("YAML config must be a mapping") + return document + + +def _boolean(config: dict, key: str) -> bool: + value = config.get(key, False) + if not isinstance(value, bool): + raise ValueError(f"{key} must be a boolean") + return value + + +def normalize(config: dict, yaml_path: Path, root: Path, + log_file_name: str | None = None) -> dict[str, Any]: + unknown = set(config) - SUPPORTED_KEYS + if unknown: + raise ValueError("Unsupported Python-library settings: " + ", ".join(sorted(map(str, unknown)))) + if config.get("format", "verilog") != "verilog": + raise ValueError("The MCP Python-library loader supports format: verilog") + if _boolean(config, "cnf_export"): + raise ValueError("CNF export is not supported by the Kepler Python library; set cnf_export: false") + + inputs = config.get("input_paths") + if not isinstance(inputs, list) or len(inputs) != 2: + raise ValueError("input_paths must contain exactly two Verilog files") + libraries = config.get("liberty_files") + if libraries is None: + libraries = [] + if not isinstance(libraries, list): + raise ValueError("liberty_files must be a list of paths") + resolved_inputs = [resolve_path(value, yaml_path.parent) for value in inputs] + resolved_libraries = [resolve_path(value, yaml_path.parent) for value in libraries] + for path in resolved_inputs + resolved_libraries: + if not path.is_file(): + raise ValueError(f"Input file not found: {path}") + + mode = config.get("verification", config.get("mode", "lec")) + if "mode" in config and config["mode"] != mode: + raise ValueError("mode and verification must agree") + result = { + "input_paths": [str(path) for path in resolved_inputs], + "liberty_files": [str(path) for path in resolved_libraries], + "mode": mode, + "solver": config.get("solver", "kissat"), + "log_level": config.get("log_level", "info"), + "allow_boundary_mismatch": _boolean(config, "allow_boundary_mismatch"), + "report_skipped_outputs": _boolean(config, "report_skipped_outputs"), + } + for key, choices in (("mode", ("lec", "sec")), + ("solver", ("kissat", "cadical", "glucose")), + ("log_level", ("info", "debug"))): + if result[key] not in choices: + raise ValueError(f"{key} must be one of: {', '.join(choices)}") + for key, choices in (("sec_engine", ("pdr", "k_induction", "imc")), + ("sec_encoding", ("dual_rail_steady", "binary"))): + value = config.get(key) + if value is not None and value not in choices: + raise ValueError(f"{key} must be one of: {', '.join(choices)}") + result[key] = value + max_k = config.get("max_k") + if max_k is not None and (type(max_k) is not int or max_k < 0): + raise ValueError("max_k must be a non-negative integer") + result["max_k"] = max_k + if mode == "lec" and any(result[key] is not None for key in ("max_k", "sec_engine", "sec_encoding")): + raise ValueError("max_k, sec_engine and sec_encoding require verification: sec") + if mode == "sec" and result["allow_boundary_mismatch"]: + raise ValueError("allow_boundary_mismatch is only supported for LEC") + + requested_log = log_file_name if log_file_name is not None else config.get("log_file") + log_file = (resolve_path(requested_log, yaml_path.parent) if requested_log is not None + else root / f"{yaml_path.stem}.log") + log_file = allowed_path(log_file, root) + if log_file in resolved_inputs + resolved_libraries + [yaml_path.resolve()]: + raise ValueError("The log file must not overwrite the YAML config or an input file") + result["log_file"] = str(log_file) + return result diff --git a/kepler_formal_mcp/runner.py b/kepler_formal_mcp/runner.py new file mode 100644 index 0000000..136cbdb --- /dev/null +++ b/kepler_formal_mcp/runner.py @@ -0,0 +1,65 @@ +"""Run the installed Python library with isolated native output and a hard timeout.""" + +from __future__ import annotations + +import json +from pathlib import Path +import subprocess +import sys +import tempfile + + +def tail(text: str | bytes | None) -> str: + if isinstance(text, bytes): + text = text.decode("utf-8", errors="replace") + return "\n".join((text or "").splitlines()[-120:]) + + +def error_result(message: str, yaml_path: Path | None = None) -> dict: + return { + "status": "error", "exit_code": -1, "verdict": "error", + "yaml_config": str(yaml_path) if yaml_path is not None else None, + "stdout_tail": "", "stderr_tail": message, + } + + +def run(request: dict, yaml_path: Path, timeout_seconds: int) -> dict: + if type(timeout_seconds) is not int or timeout_seconds <= 0: + raise ValueError("timeout_seconds must be a positive integer") + log_path = Path(request["log_file"]) + log_path.parent.mkdir(parents=True, exist_ok=True) + # Native code can write directly to stdout and cannot be interrupted by a + # Python thread timeout. A worker keeps both behaviors away from MCP stdio. + with tempfile.TemporaryDirectory(prefix="kepler-mcp-") as directory: + work = Path(directory) + request_path, result_path = work / "request.json", work / "result.json" + request_path.write_text(json.dumps(request), encoding="utf-8") + try: + completed = subprocess.run( + [sys.executable, "-m", "kepler_formal_mcp.worker", str(request_path), str(result_path)], + cwd=work, capture_output=True, text=True, encoding="utf-8", errors="replace", + timeout=timeout_seconds, check=False, + ) + except subprocess.TimeoutExpired as error: + result = error_result(f"Kepler-Formal timed out after {timeout_seconds} seconds", yaml_path) + result.update(stdout_tail=tail(error.stdout), generated_log_file=str(log_path)) + return result + if not result_path.is_file(): + result = error_result("Kepler Python worker exited without a verification result", yaml_path) + result.update(exit_code=completed.returncode, stdout_tail=tail(completed.stdout), + stderr_tail=tail(completed.stderr) or result["stderr_tail"], + generated_log_file=str(log_path)) + return result + verification = json.loads(result_path.read_text(encoding="utf-8")) + status = "error" if completed.returncode != 0 or verification["status"] == "error" else "success" + return { + "status": status, + "exit_code": verification["exit_code"], + "verdict": verification["status"], + "verification_result": verification, + "yaml_config": str(yaml_path), + "generated_log_file": str(log_path), + "stdout_tail": tail(completed.stdout), + "stderr_tail": tail(completed.stderr), + "log_tail": tail(log_path.read_text(encoding="utf-8", errors="replace")) if log_path.is_file() else "", + } diff --git a/kepler_formal_mcp/server.py b/kepler_formal_mcp/server.py new file mode 100644 index 0000000..c735ab5 --- /dev/null +++ b/kepler_formal_mcp/server.py @@ -0,0 +1,106 @@ +"""MCP tools backed by the published Kepler Formal Python library.""" + +from __future__ import annotations + +import json +import logging +from pathlib import Path +import sys + +from mcp.server.fastmcp import FastMCP +import yaml + +from . import config, runner + + +app = FastMCP("kepler-formal") + + +@app.tool() +def run_kepler_formal_yaml( + yaml_file: str, + timeout_seconds: int = 600, + log_file_name: str | None = None, + allowed_output_dir: str | None = None, +) -> str: + """Compare two Verilog designs using an existing YAML config and the Python API. + + Relative input/library/log paths are relative to the YAML file. The config + is never rewritten. Logs must be under allowed_output_dir (or the + KEPLER_FORMAL_AI_OUTPUT_DIR environment setting; default: launch directory). + `verdict` reports equivalence independently of operation `status`/exit_code. + Unsupported CLI-only YAML options are rejected, including CNF export. + """ + yaml_path = None + try: + yaml_path = config.resolve_path(yaml_file, Path.cwd()) + root = config.output_root(allowed_output_dir) + request = config.normalize(config.load_yaml(yaml_path), yaml_path, root, log_file_name) + result = runner.run(request, yaml_path, timeout_seconds) + except (OSError, ValueError, yaml.YAMLError) as error: + result = runner.error_result(str(error), yaml_path) + return json.dumps(result, indent=2) + + +@app.tool() +def create_yaml_and_run_kepler_formal( + input_paths: list[str], + liberty_files: list[str], + yaml_output_path: str = "test_config_verilog.yaml", + log_level: str = "info", + solver: str = "kissat", + cnf_export: bool = False, + cnf_export_path: str = "./sat.cnf", + log_file_name: str | None = None, + allowed_output_dir: str | None = None, + timeout_seconds: int = 600, + verification: str = "lec", + max_k: int | None = None, + sec_engine: str | None = None, + sec_encoding: str | None = None, + allow_boundary_mismatch: bool = False, + report_skipped_outputs: bool = False, +) -> str: + """Load exactly two Verilog designs in NajaEDA and check them with Kepler Python. + + Input/library paths are relative to the launch directory. YAML output is + relative to allowed_output_dir (or KEPLER_FORMAL_AI_OUTPUT_DIR; default: + launch directory). log_file_name is relative to the generated YAML file. + Use verification='lec' or 'sec'; max_k/SEC engine/encoding apply only to SEC. + CNF export is unavailable in the Python API and cnf_export must stay false. + Read `verdict` for equivalent/different/inconclusive, not the native exit code. + """ + yaml_path = None + try: + root = config.output_root(allowed_output_dir) + yaml_path = config.allowed_path(config.resolve_path(yaml_output_path, root), root) + document = { + "format": "verilog", + "input_paths": [str(config.resolve_path(path, Path.cwd())) for path in input_paths], + "liberty_files": [str(config.resolve_path(path, Path.cwd())) for path in liberty_files], + "log_level": log_level, "solver": solver, "cnf_export": cnf_export, + "verification": verification, "max_k": max_k, "sec_engine": sec_engine, + "sec_encoding": sec_encoding, "allow_boundary_mismatch": allow_boundary_mismatch, + "report_skipped_outputs": report_skipped_outputs, + } + request = config.normalize(document, yaml_path, root, log_file_name) + if str(yaml_path) in request["input_paths"] + request["liberty_files"]: + raise ValueError("The YAML output must not overwrite an input file") + if type(timeout_seconds) is not int or timeout_seconds <= 0: + raise ValueError("timeout_seconds must be a positive integer") + document["log_file"] = request["log_file"] + document = {key: value for key, value in document.items() if value is not None} + text = yaml.safe_dump(document, sort_keys=False) + yaml_path.parent.mkdir(parents=True, exist_ok=True) + yaml_path.write_text(text, encoding="utf-8") + result = runner.run(request, yaml_path, timeout_seconds) + result.update(generated_yaml=str(yaml_path), generated_yaml_preview=text) + except (OSError, ValueError, yaml.YAMLError) as error: + result = runner.error_result(str(error), yaml_path) + return json.dumps(result, indent=2) + + +def main() -> None: + logging.basicConfig(level=logging.INFO, stream=sys.stderr, + format="[kepler-mcp] [%(levelname)s] %(message)s") + app.run() diff --git a/kepler_formal_mcp/worker.py b/kepler_formal_mcp/worker.py new file mode 100644 index 0000000..3d3a8fc --- /dev/null +++ b/kepler_formal_mcp/worker.py @@ -0,0 +1,94 @@ +"""Run one verification in a separate process using the installed Python API. + +Native parsers and solvers may write directly to stdout. The MCP server captures +this process's output and reads the structured verdict from a separate JSON file. +""" + +from __future__ import annotations + +import argparse +from dataclasses import asdict +import json +from pathlib import Path +import sys +import traceback +from typing import Any + + +def verify(request: dict[str, Any]) -> dict[str, Any]: + """Load the requested designs in NajaEDA and borrow them for verification.""" + # Import NajaEDA first: it arranges the shared native runtime used by KF. + from najaeda import naja + from kepler_formal import VerificationOptions, verify_designs + + if naja.NLUniverse.get() is not None: + raise RuntimeError("The verification worker requires a fresh Naja universe") + + universe = naja.NLUniverse.create() + try: + designs = [] + for path in request["input_paths"]: + # Independent databases allow both inputs to have the same module + # names, while retaining their pointers in one shared runtime. + database = naja.NLDB.create(universe) + if request["liberty_files"]: + database.loadLibertyPrimitives(request["liberty_files"]) + database.loadVerilog([path]) + design = database.getTopDesign() + if design is None: + raise RuntimeError(f"NajaEDA did not find a top design in {path}") + designs.append(design) + + result = verify_designs( + designs[0], + designs[1], + options=VerificationOptions( + mode=request["mode"], + solver=request["solver"], + max_k=request.get("max_k"), + sec_engine=request.get("sec_engine"), + sec_encoding=request.get("sec_encoding"), + allow_boundary_mismatch=request["allow_boundary_mismatch"], + report_skipped_outputs=request["report_skipped_outputs"], + log_file=request["log_file"], + log_level=request["log_level"], + ), + ) + serialized = asdict(result) + serialized.update( + status=result.status.value, + equivalent=result.equivalent, + conclusive=result.conclusive, + coverage_percent=result.coverage_percent, + ) + return serialized + finally: + # This universe belongs solely to this short-lived worker. The MCP + # server and any callers' in-memory netlists are never reset. + universe.destroy() + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("request", type=Path) + parser.add_argument("result", type=Path) + arguments = parser.parse_args(argv) + try: + request = json.loads(arguments.request.read_text(encoding="utf-8")) + result = verify(request) + except Exception as error: + traceback.print_exc(file=sys.stderr) + result = { + "status": "error", + "exit_code": -1, + "reason": f"{type(error).__name__}: {error}", + "equivalent": False, + "conclusive": False, + "coverage_percent": None, + } + arguments.result.write_text(json.dumps(result) + "\n", encoding="utf-8") + return 1 if result["status"] == "error" else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..3547988 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,24 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "kepler-formal-mcp" +version = "0.1.0" +description = "MCP tools for equivalence checking with the Kepler Formal Python library" +readme = "README.md" +requires-python = ">=3.10" +dependencies = [ + "kepler-formal==0.5.0", + "mcp>=1.12,<2", + "PyYAML>=6,<7", +] + +[project.scripts] +kepler-formal-mcp = "kepler_formal_mcp.server:main" + +[project.urls] +Repository = "https://github.com/keplertech/kepler-formal-mcp" + +[tool.setuptools.packages.find] +include = ["kepler_formal_mcp*"] diff --git a/server.py b/server.py index 5772a97..9f3e514 100644 --- a/server.py +++ b/server.py @@ -1,354 +1,8 @@ #!/usr/bin/env python3 -"""MCP server exposing Kepler-Formal helpers. +"""Compatibility launcher; install this repository with pip before running.""" -Tools: -1) Run Kepler-Formal from an existing YAML config file. -2) Build a YAML config from provided design/library paths, then run tool #1. -""" - -from __future__ import annotations - -import json -import logging -import os -from pathlib import Path -import subprocess -import sys - -from mcp.server.fastmcp import FastMCP - - -app = FastMCP("kepler-formal") - -def _workspace_root() -> Path: - return Path(__file__).resolve().parent - - -def _default_ai_output_dir() -> Path: - """Return the workspace-local default writable folder for AI outputs.""" - return _workspace_root() - - -# Writable folder for AI-generated outputs (yaml/log). -AI_OUTPUT_DIR = _default_ai_output_dir() - -# Keep stdout dedicated to MCP JSON-RPC messages. -logging.basicConfig( - level=logging.INFO, - stream=sys.stderr, - format="[kepler-mcp] [%(levelname)s] %(message)s", - force=True, -) - - - -def _binary_path() -> Path: - # Prefer the submodule build tree under thirdparty/. - candidates = [ - _workspace_root() / "thirdparty" / "kepler-formal" / "build" / "src" / "bin" / "kepler-formal", - _workspace_root() / "build" / "src" / "bin" / "kepler-formal", - ] - for candidate in candidates: - resolved = candidate.resolve() - if resolved.exists(): - return resolved - return candidates[-1].resolve() - - -def _resolve_path(path_value: str) -> Path: - path = Path(path_value) - if path.is_absolute(): - return path - return (_workspace_root() / path).resolve() - - -def get_allowed_dirs(allowed_output_dir: str | None = None) -> list[Path]: - """Return allowed writable directories for AI outputs. - - Priority: - 1) explicit tool parameter - 2) KEPLER_FORMAL_AI_OUTPUT_DIR environment variable - 3) workspace root - """ - if allowed_output_dir: - return [Path(allowed_output_dir).expanduser().resolve()] - - env_value = os.environ.get("KEPLER_FORMAL_AI_OUTPUT_DIR") - env_dir = Path(env_value) if env_value else _default_ai_output_dir() - return [env_dir.expanduser().resolve()] - - -def _yaml_output_path_in_ai_dir(yaml_output_path: str, output_root: Path) -> Path: - """Resolve YAML output while preserving the caller-provided path.""" - candidate = Path(yaml_output_path).expanduser() - if not candidate.name: - candidate = Path("test_config_verilog.yaml") - if candidate.is_absolute(): - return candidate.resolve() - return (output_root / candidate).resolve() - - -def _ensure_allowed(path: Path, allowed_dirs: list[Path] | None = None) -> Path: - resolved = path.expanduser().resolve() - dirs = allowed_dirs if allowed_dirs is not None else get_allowed_dirs() - for allowed_dir in dirs: - try: - resolved.relative_to(allowed_dir) - return resolved - except ValueError: - continue - raise ValueError(f"Path not allowed: {resolved}") - - -def _read_yaml_log_file(yaml_text: str, yaml_dir: Path) -> Path | None: - for raw_line in yaml_text.splitlines(): - line = raw_line.strip() - if not line or line.startswith("#"): - continue - if not line.startswith("log_file:"): - continue - - value = line.split(":", 1)[1].strip() - if (value.startswith('"') and value.endswith('"')) or ( - value.startswith("'") and value.endswith("'") - ): - value = value[1:-1] - if not value: - return None - - path = Path(value) - if not path.is_absolute(): - path = (yaml_dir / path).resolve() - return path - return None - - -def _normalize_log_file_path(candidate: Path, fallback_base: Path, allowed_dirs: list[Path]) -> Path: - """Return an allowed log file path rooted at the caller-provided base directory.""" - if candidate.is_absolute(): - allowed_candidate = candidate.resolve() - else: - allowed_candidate = (fallback_base / candidate).resolve() - - if allowed_candidate.name in {"", ".", ".."}: - allowed_candidate = (fallback_base / "kepler-formal.log").resolve() - elif allowed_candidate.suffix.lower() not in {".log", ".txt"}: - allowed_candidate = allowed_candidate.with_suffix(".log") - - return _ensure_allowed(allowed_candidate, allowed_dirs) - - -def _fix_yaml_log_file(yaml_path: Path, yaml_text: str, log_file_path: Path) -> None: - lines = yaml_text.splitlines() - log_line = f"log_file: {json.dumps(str(log_file_path))}" - for index, raw_line in enumerate(lines): - if raw_line.strip().startswith("log_file:"): - lines[index] = log_line - yaml_path.write_text("\n".join(lines) + "\n", encoding="utf-8") - return - - lines.append(log_line) - yaml_path.write_text("\n".join(lines) + "\n", encoding="utf-8") - - -def _format_result(result: subprocess.CompletedProcess[str], yaml_path: Path) -> dict[str, object]: - return { - "status": "success" if result.returncode == 0 else "error", - "exit_code": result.returncode, - "yaml_config": str(yaml_path), - "stdout_tail": "\n".join(result.stdout.splitlines()[-120:]), - "stderr_tail": "\n".join(result.stderr.splitlines()[-120:]), - } - - -def _run_from_yaml( - yaml_path: Path, - timeout_seconds: int, - allowed_dirs: list[Path], - log_file_name: str | None = None, -) -> dict[str, object]: - binary = _binary_path() - if not binary.exists(): - return { - "status": "error", - "exit_code": -1, - "yaml_config": str(yaml_path), - "stdout_tail": "", - "stderr_tail": f"Kepler-Formal binary not found: {binary}", - } - - if not yaml_path.exists(): - return { - "status": "error", - "exit_code": -1, - "yaml_config": str(yaml_path), - "stdout_tail": "", - "stderr_tail": f"YAML file not found: {yaml_path}", - } - - yaml_text = yaml_path.read_text(encoding="utf-8") - log_file_path = _read_yaml_log_file(yaml_text, yaml_path.parent) - if log_file_name: - log_file_path = Path(log_file_name) - if log_file_path is None: - normalized_log_file_path = _normalize_log_file_path(yaml_path.with_suffix(".log"), yaml_path.parent, allowed_dirs) - _fix_yaml_log_file(yaml_path, yaml_text, normalized_log_file_path) - yaml_text = yaml_path.read_text(encoding="utf-8") - log_file_path = normalized_log_file_path - else: - normalized_log_file_path = _normalize_log_file_path(log_file_path, yaml_path.parent, allowed_dirs) - if normalized_log_file_path != log_file_path: - _fix_yaml_log_file(yaml_path, yaml_text, normalized_log_file_path) - yaml_text = yaml_path.read_text(encoding="utf-8") - log_file_path = normalized_log_file_path - - log_file_path.parent.mkdir(parents=True, exist_ok=True) - - try: - result = subprocess.run( - [str(binary), "--config", str(yaml_path)], - cwd=str(_workspace_root()), - capture_output=True, - text=True, - timeout=timeout_seconds, - check=False, - ) - except subprocess.TimeoutExpired: - return { - "status": "error", - "exit_code": -1, - "yaml_config": str(yaml_path), - "stdout_tail": "", - "stderr_tail": f"Kepler-Formal timed out after {timeout_seconds} seconds", - } - - if result.returncode == 0 and not log_file_path.exists(): - return { - "status": "error", - "exit_code": result.returncode, - "yaml_config": str(yaml_path), - "stdout_tail": "\n".join(result.stdout.splitlines()[-120:]), - "stderr_tail": f"Expected log file was not created: {log_file_path}", - "generated_log_file": str(log_file_path), - } - - formatted_result = _format_result(result, yaml_path) - formatted_result["generated_log_file"] = str(log_file_path) - return formatted_result - - -@app.tool() -def run_kepler_formal_yaml( - yaml_file: str, - timeout_seconds: int = 600, - log_file_name: str | None = None, - allowed_output_dir: str | None = None, -) -> str: - """Run Kepler-Formal from an existing YAML file. - - Args: - yaml_file: Path to YAML config file. - timeout_seconds: Timeout for command execution. - log_file_name: Optional log filename/path. Final location is forced under allowed_output_dir root. - allowed_output_dir: Optional writable directory override for this run. - """ - yaml_path = _resolve_path(yaml_file) - allowed_dirs = get_allowed_dirs(allowed_output_dir) - for allowed_dir in allowed_dirs: - allowed_dir.mkdir(parents=True, exist_ok=True) - output = _run_from_yaml( - yaml_path=yaml_path, - timeout_seconds=timeout_seconds, - allowed_dirs=allowed_dirs, - log_file_name=log_file_name, - ) - return json.dumps(output, indent=2) - - -@app.tool() -def create_yaml_and_run_kepler_formal( - input_paths: list[str], - liberty_files: list[str], - yaml_output_path: str = "test_config_verilog.yaml", - log_level: str = "info", - solver: str = "kissat", - cnf_export: bool = True, - cnf_export_path: str = "./sat.cnf", - log_file_name: str | None = None, - allowed_output_dir: str | None = None, - timeout_seconds: int = 600, -) -> str: - """Create a verilog YAML config file from provided data and run Kepler-Formal. - - Args: - input_paths: Usually [golden_verilog, revised_verilog]. - liberty_files: List of .lib files. - yaml_output_path: Where to write YAML config. - log_level: YAML log_level value. - solver: YAML solver value. - cnf_export: YAML cnf_export value. - cnf_export_path: YAML cnf_export_path value. - log_file_name: Optional log filename/path. Final location is forced under allowed_output_dir root. - allowed_output_dir: Optional writable directory override for this run. - timeout_seconds: Timeout for Kepler-Formal run. - """ - if len(input_paths) < 2: - return json.dumps( - { - "status": "error", - "exit_code": -1, - "stdout_tail": "", - "stderr_tail": "input_paths must contain at least 2 files", - }, - indent=2, - ) - - resolved_inputs = [str(_resolve_path(p)) for p in input_paths] - resolved_libs = [str(_resolve_path(p)) for p in liberty_files] - allowed_dirs = get_allowed_dirs(allowed_output_dir) - for allowed_dir in allowed_dirs: - allowed_dir.mkdir(parents=True, exist_ok=True) - yaml_path = _ensure_allowed(_yaml_output_path_in_ai_dir(yaml_output_path, allowed_dirs[0]), allowed_dirs) - yaml_path.parent.mkdir(parents=True, exist_ok=True) - requested_log = Path(log_file_name) if log_file_name else yaml_path.with_suffix(".log") - log_file_path = _normalize_log_file_path(requested_log, yaml_path.parent, allowed_dirs) - - lines: list[str] = [ - "format: verilog", - "input_paths:", - ] - for path in resolved_inputs: - lines.append(f" - {json.dumps(path)}") - - lines.append("liberty_files:") - for path in resolved_libs: - lines.append(f" - {json.dumps(path)}") - - lines.extend( - [ - f"log_level: {log_level}", - f"solver: {solver}", - f"cnf_export: {'true' if cnf_export else 'false'}", - f"cnf_export_path: {cnf_export_path}", - f"log_file: {json.dumps(str(log_file_path))}", - ] - ) - - yaml_path.write_text("\n".join(lines) + "\n", encoding="utf-8") - log_file_path.parent.mkdir(parents=True, exist_ok=True) - - run_result = _run_from_yaml( - yaml_path=yaml_path, - timeout_seconds=timeout_seconds, - allowed_dirs=allowed_dirs, - log_file_name=log_file_name, - ) - run_result["generated_yaml"] = str(yaml_path) - run_result["generated_log_file"] = str(log_file_path) - run_result["generated_yaml_preview"] = "\n".join(lines) - return json.dumps(run_result, indent=2) +from kepler_formal_mcp.server import main if __name__ == "__main__": - logging.info("Starting kepler-formal MCP server") - app.run() + main() diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 0000000..d904599 --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,297 @@ +"""Exercise the MCP tools against the installed Kepler Formal Python package.""" + +from __future__ import annotations + +import asyncio +from contextlib import contextmanager +from datetime import timedelta +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest +from unittest.mock import patch + +from mcp import ClientSession +from mcp.client.stdio import StdioServerParameters, stdio_client +import yaml + +from kepler_formal_mcp import server + + +PASS_THROUGH = "module top(input a, output y); assign y = a; endmodule\n" +CONSTANT_OUTPUT = "module top(input a, output y); assign y = 1'b0; endmodule\n" +LIBERTY = """ +library(test_cells) { + cell(BUF) { + pin(A) { direction : input; } + pin(Y) { direction : output; function : "A"; } + } + cell(INV) { + pin(A) { direction : input; } + pin(Y) { direction : output; function : "!A"; } + } +} +""" + + +@contextmanager +def working_directory(path: Path): + previous = Path.cwd() + os.chdir(path) + try: + yield + finally: + os.chdir(previous) + + +class ToolTest(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory(prefix="kepler mcp test ") + self.addCleanup(self.temporary.cleanup) + self.root = Path(self.temporary.name).resolve() + self.designs = self.root / "designs" + self.designs.mkdir() + self.outputs = self.root / "outputs" + self.outputs.mkdir() + self.reference = self.write_design("reference.v", PASS_THROUGH) + self.candidate = self.write_design("candidate.v", PASS_THROUGH) + + def write_design(self, name, text): + path = self.designs / name + path.write_text(text, encoding="utf-8") + return path + + def create(self, **overrides): + arguments = { + "input_paths": [str(self.reference), str(self.candidate)], + "liberty_files": [], + "yaml_output_path": str(self.outputs / "comparison.yaml"), + "allowed_output_dir": str(self.outputs), + "timeout_seconds": 30, + } + arguments.update(overrides) + return json.loads(server.create_yaml_and_run_kepler_formal(**arguments)) + + def run_yaml(self, path, **overrides): + arguments = { + "yaml_file": str(path), + "allowed_output_dir": str(self.outputs), + "timeout_seconds": 30, + } + arguments.update(overrides) + return json.loads(server.run_kepler_formal_yaml(**arguments)) + + def assert_verdict(self, result, verdict): + self.assertEqual(result["status"], "success", result) + self.assertEqual(result["verdict"], verdict, result) + self.assertEqual(result["verification_result"]["status"], verdict, result) + self.assertTrue(Path(result["generated_log_file"]).is_file(), result) + + def assert_error(self, result, message=None): + self.assertEqual(result["status"], "error", result) + if message: + self.assertIn(message.lower(), result["stderr_tail"].lower(), result) + + def test_equivalent_designs_with_identical_top_names(self): + result = self.create() + self.assert_verdict(result, "equivalent") + self.assertEqual(result["exit_code"], 0) + generated = Path(result["generated_yaml"]) + self.assertEqual(generated, self.outputs / "comparison.yaml") + configuration = yaml.safe_load(generated.read_text(encoding="utf-8")) + self.assertEqual(configuration["input_paths"], [str(self.reference), str(self.candidate)]) + self.assertFalse(configuration.get("cnf_export", False)) + + def test_difference_is_a_completed_operation_and_has_structured_verdict(self): + self.candidate.write_text(CONSTANT_OUTPUT, encoding="utf-8") + result = self.create() + self.assert_verdict(result, "different") + # Native completion codes alone do not distinguish these two verdicts. + self.assertEqual(result["exit_code"], 0) + + def test_sec_options_and_nonzero_difference_code_keep_semantic_verdict(self): + self.candidate.write_text(CONSTANT_OUTPUT, encoding="utf-8") + result = self.create(verification="sec", sec_engine="pdr", sec_encoding="binary", max_k=2) + self.assert_verdict(result, "different") + self.assertEqual(result["verification_result"]["verification"], "sec") + self.assertNotEqual(result["exit_code"], 0) + + def test_documented_examples_run_with_the_published_parser(self): + import re + + guide = Path(__file__).resolve().parents[1] / "docs/test-generation.md" + sources = re.findall(r"```verilog\n(.*?)```", guide.read_text(encoding="utf-8"), re.S) + self.assertEqual(len(sources), 3) + self.reference.write_text(sources[0], encoding="utf-8") + for source, verdict in zip(sources[1:], ("equivalent", "different")): + self.candidate.write_text(source, encoding="utf-8") + self.assert_verdict(self.create(), verdict) + + def test_liberty_is_loaded_for_both_designs(self): + library = self.designs / "cells.lib" + library.write_text(LIBERTY, encoding="utf-8") + cell_design = "module top(input a, output y); %s gate(.A(a), .Y(y)); endmodule\n" + self.reference.write_text(cell_design % "BUF", encoding="utf-8") + for cell, verdict in [("BUF", "equivalent"), ("INV", "different")]: + with self.subTest(cell=cell): + self.candidate.write_text(cell_design % cell, encoding="utf-8") + self.assert_verdict(self.create(liberty_files=[str(library)]), verdict) + + def test_existing_yaml_uses_its_own_directory_and_is_not_rewritten(self): + configuration = self.designs / "existing.yaml" + original = ( + "# Preserve the user's comments and relative paths.\n" + "format: verilog\n" + "input_paths: [reference.v, candidate.v]\n" + "liberty_files: []\n" + "solver: kissat\n" + "log_file: ../outputs/from-yaml.log\n" + ) + configuration.write_text(original, encoding="utf-8") + with working_directory(self.outputs): + result = self.run_yaml(configuration) + self.assert_verdict(result, "equivalent") + self.assertEqual(Path(result["generated_log_file"]), self.outputs / "from-yaml.log") + self.assertEqual(configuration.read_text(encoding="utf-8"), original) + + def test_log_override_does_not_modify_input_yaml(self): + configuration = self.designs / "existing.yaml" + original = yaml.safe_dump({ + "input_paths": ["reference.v", "candidate.v"], + "liberty_files": [], + "log_file": "old.log", + }) + configuration.write_text(original, encoding="utf-8") + expected_log = self.outputs / "override.log" + result = self.run_yaml(configuration, log_file_name=str(expected_log)) + self.assert_verdict(result, "equivalent") + self.assertEqual(Path(result["generated_log_file"]), expected_log) + self.assertEqual(configuration.read_text(encoding="utf-8"), original) + self.assertFalse((self.designs / "old.log").exists()) + + def test_environment_selects_output_directory_and_inputs_resolve_from_cwd(self): + with working_directory(self.designs), patch.dict( + os.environ, {"KEPLER_FORMAL_AI_OUTPUT_DIR": str(self.outputs)} + ): + result = self.create( + input_paths=["reference.v", "candidate.v"], + allowed_output_dir=None, + ) + self.assert_verdict(result, "equivalent") + + def test_requires_exactly_two_designs(self): + for paths in [[], [str(self.reference)], [str(self.reference)] * 3]: + with self.subTest(paths=paths): + self.assert_error(self.create(input_paths=paths), "two") + + def test_missing_input_and_invalid_verilog_are_reported(self): + self.assert_error(self.create(input_paths=[str(self.reference), str(self.designs / "missing.v")])) + self.candidate.write_text("this is not valid Verilog\n", encoding="utf-8") + self.assert_error(self.create()) + + def test_cnf_export_is_rejected_instead_of_silently_ignored(self): + self.assert_error(self.create(cnf_export=True), "cnf_export") + + def test_yaml_rejects_unknown_keys_and_unsupported_options(self): + configuration = self.designs / "invalid.yaml" + base = { + "input_paths": ["reference.v", "candidate.v"], + "liberty_files": [], + "log_file": str(self.outputs / "validation.log"), + } + for extra in [ + {"misspelled_option": True}, + {"format": "blif"}, + {"cnf_export": True}, + {"solver": "not-a-solver"}, + {"verification": "not-a-mode"}, + ]: + with self.subTest(extra=extra): + configuration.write_text(yaml.safe_dump({**base, **extra}), encoding="utf-8") + self.assert_error(self.run_yaml(configuration)) + + def test_invalid_yaml_is_a_structured_error(self): + configuration = self.designs / "invalid.yaml" + for text in ["input_paths: [\n", "- not\n- a\n- mapping\n"]: + with self.subTest(text=text): + configuration.write_text(text, encoding="utf-8") + self.assert_error(self.run_yaml(configuration)) + + def test_generated_files_cannot_escape_allowed_directory(self): + outside = self.root / "outside.yaml" + self.assert_error(self.create(yaml_output_path=str(outside))) + self.assertFalse(outside.exists()) + outside_log = self.root / "outside.log" + self.assert_error(self.create(log_file_name=str(outside_log))) + self.assertFalse(outside_log.exists()) + + def test_generated_files_cannot_overwrite_designs(self): + for options in ({"yaml_output_path": str(self.reference)}, + {"log_file_name": str(self.reference)}): + with self.subTest(options=options): + self.assert_error(self.create(allowed_output_dir=str(self.root), **options), "overwrite") + self.assertEqual(self.reference.read_text(encoding="utf-8"), PASS_THROUGH) + + def test_yaml_log_cannot_escape_allowed_directory(self): + configuration = self.designs / "escape.yaml" + configuration.write_text(yaml.safe_dump({ + "input_paths": ["reference.v", "candidate.v"], + "log_file": "outside.log", + }), encoding="utf-8") + self.assert_error(self.run_yaml(configuration)) + self.assertFalse((self.designs / "outside.log").exists()) + + def test_worker_timeout_is_reported_without_crashing_the_server(self): + with patch("kepler_formal_mcp.runner.subprocess.run", side_effect=subprocess.TimeoutExpired( + [sys.executable, "-m", "kepler_formal_mcp.worker"], timeout=1 + )) as run: + self.assert_error(self.create(timeout_seconds=1), "timed out") + self.assertEqual(run.call_args.kwargs["timeout"], 1) + + def test_native_worker_crash_is_reported_without_a_result_file(self): + with patch("kepler_formal_mcp.runner.subprocess.run", return_value=subprocess.CompletedProcess( + args=[sys.executable], returncode=-11, stdout="", stderr="native worker crash" + )): + result = self.create() + self.assert_error(result, "native worker crash") + self.assertEqual(result["exit_code"], -11) + + def test_native_output_does_not_corrupt_stdio_mcp(self): + async def exercise(): + parameters = StdioServerParameters( + command=sys.executable, + args=["-m", "kepler_formal_mcp"], + cwd=str(self.root), + env=dict(os.environ), + ) + with tempfile.TemporaryFile(mode="w+", encoding="utf-8") as errors: + async with stdio_client(parameters, errlog=errors) as (reader, writer): + async with ClientSession(reader, writer, read_timeout_seconds=timedelta(seconds=45)) as session: + await session.initialize() + tools = await session.list_tools() + names = {tool.name for tool in tools.tools} + self.assertTrue({"run_kepler_formal_yaml", "create_yaml_and_run_kepler_formal"} <= names) + for content, verdict in [(PASS_THROUGH, "equivalent"), (CONSTANT_OUTPUT, "different")]: + self.candidate.write_text(content, encoding="utf-8") + response = await session.call_tool("create_yaml_and_run_kepler_formal", { + "input_paths": [str(self.reference), str(self.candidate)], + "liberty_files": [], + "solver": "glucose", + "yaml_output_path": str(self.outputs / "stdio.yaml"), + "allowed_output_dir": str(self.outputs), + "timeout_seconds": 30, + }) + self.assertFalse(response.isError, response) + text = next(item.text for item in response.content if item.type == "text") + self.assert_verdict(json.loads(text), verdict) + # A second protocol operation also works after native solver logging. + self.assertEqual(names, {tool.name for tool in (await session.list_tools()).tools}) + + asyncio.run(exercise()) + + +if __name__ == "__main__": + unittest.main() diff --git a/thirdparty/kepler-formal b/thirdparty/kepler-formal deleted file mode 160000 index 70f7810..0000000 --- a/thirdparty/kepler-formal +++ /dev/null @@ -1 +0,0 @@ -Subproject commit 70f7810ecb7d64b8d0a53cc502eb1dca0f73cbe6 From 52e2b358a1dc7c72d1101e0772291f777af154d2 Mon Sep 17 00:00:00 2001 From: Noam Cohen Date: Thu, 17 Sep 2026 10:53:22 +0200 Subject: [PATCH 2/2] Expose Kepler API capabilities and preserve verification reports --- README.md | 8 +- docs/python-api.md | 78 ++++++++++++++++ kepler_formal_mcp/capabilities.py | 55 +++++++++++ kepler_formal_mcp/config.py | 16 ++-- kepler_formal_mcp/options.py | 14 +++ kepler_formal_mcp/runner.py | 74 ++++++++++++--- kepler_formal_mcp/server.py | 29 ++++-- kepler_formal_mcp/worker.py | 10 +- tests/test_api_coverage.py | 148 ++++++++++++++++++++++++++++++ tests/test_capabilities.py | 76 +++++++++++++++ tests/test_reports.py | 116 +++++++++++++++++++++++ tests/test_server.py | 5 + 12 files changed, 599 insertions(+), 30 deletions(-) create mode 100644 docs/python-api.md create mode 100644 kepler_formal_mcp/capabilities.py create mode 100644 kepler_formal_mcp/options.py create mode 100644 tests/test_api_coverage.py create mode 100644 tests/test_capabilities.py create mode 100644 tests/test_reports.py diff --git a/README.md b/README.md index 4577aa0..3bca92a 100644 --- a/README.md +++ b/README.md @@ -34,10 +34,11 @@ The existing `python server.py` launcher also works using the installed environm ## Tools +- `get_kepler_formal_info`: report installed versions, build revision, and the available Python API options/results. - `run_kepler_formal_yaml`: verify the pair specified by an existing YAML file. - `create_yaml_and_run_kepler_formal`: create a YAML file from two Verilog paths and optional common Liberty libraries, then verify the pair. -Both tools use the Python library in a separate Python worker for each call. This keeps native solver output out of the MCP stdio channel and allows `timeout_seconds` to stop a verification. The worker loads both designs with NajaEDA and passes their live design handles to `kepler_formal.verify_designs`. +The tools use the Python library in a separate Python worker for each call. This keeps native solver output out of the MCP stdio channel and allows `timeout_seconds` to stop a verification. For verification, the worker loads both designs with NajaEDA and passes their live design handles to `kepler_formal.verify_designs`. A minimal YAML file is: @@ -58,6 +59,8 @@ Set `KEPLER_FORMAL_AI_OUTPUT_DIR` to a writable output directory, or pass `allow Supported verification settings are `verification` (`lec` or `sec`; `mode` is an alias), `solver`, `max_k`, `sec_engine`, `sec_encoding`, `allow_boundary_mismatch`, `report_skipped_outputs`, `log_file`, and `log_level`. CNF export is unavailable through this library API: `cnf_export` defaults to `false`, and `true` returns a clear error. Unsupported CLI-only YAML options also return an error. +All current `VerificationOptions` fields are exposed. See [Python API coverage and SEC usage](docs/python-api.md) for the mapping and supported choices. MCP tool schemas list the accepted modes, solvers, engines and encodings; `get_kepler_formal_info` reads them from the installed library. + ## Results The returned JSON separates tool execution from the verification verdict: @@ -66,6 +69,7 @@ The returned JSON separates tool execution from the verification verdict: - `verdict`: the equivalence outcome, such as `equivalent`, `different`, or `inconclusive`. - `verification_result`: the structured result returned by Kepler Formal. - `stdout_tail` / `stderr_tail`: captured worker output for diagnosis. +- `reports`: contents of native skipped-output reports, retained after worker cleanup when `report_skipped_outputs` is enabled. Always inspect `verdict`: a successful execution can report different designs. A bounded SEC check can be inconclusive; it is not an equivalence proof. @@ -77,4 +81,4 @@ See [the small design example](docs/test-generation.md) for equivalent and diffe python -m unittest discover -s tests -v ``` -CI runs the tests against the published dependencies on Linux x86_64, macOS ARM64, and Windows AMD64 with Python 3.10 and 3.14. Tests cover the real Python verifier, configuration validation, worker isolation, and MCP stdio calls. +CI runs the tests against the published dependencies on Linux x86_64, macOS ARM64, and Windows AMD64 with Python 3.10 and 3.14. Tests cover every solver and SEC engine/encoding, API option/result coverage, diagnostic reports, configuration validation, worker isolation, and MCP stdio calls. diff --git a/docs/python-api.md b/docs/python-api.md new file mode 100644 index 0000000..f549eeb --- /dev/null +++ b/docs/python-api.md @@ -0,0 +1,78 @@ +# Python API coverage + +The MCP uses published Kepler Formal `0.5.0`. Its verification API is +`verify_designs(reference, candidate, options=VerificationOptions(...))`. +NajaEDA loads both designs inside the worker, and Kepler borrows their existing +native handles for the check. + +All nine `VerificationOptions` fields are available through +`create_yaml_and_run_kepler_formal` and existing YAML configurations: + +| Python option | MCP argument | YAML key | Accepted values | +| --- | --- | --- | --- | +| `mode` | `verification` | `verification` or `mode` | `lec`, `sec` | +| `solver` | `solver` | `solver` | `kissat`, `cadical`, `glucose` | +| `max_k` | `max_k` | `max_k` | Nonnegative integer or null; SEC only | +| `sec_engine` | `sec_engine` | `sec_engine` | `pdr`, `k_induction`, `imc`, or null | +| `sec_encoding` | `sec_encoding` | `sec_encoding` | `dual_rail_steady`, `binary`, or null | +| `allow_boundary_mismatch` | Same name | Same name | Boolean; LEC only | +| `report_skipped_outputs` | Same name | Same name | Boolean | +| `log_file` | `log_file_name` | `log_file` | Path within the allowed output directory | +| `log_level` | `log_level` | `log_level` | `info`, `debug`, or null | + +The MCP creates a log path when one is omitted. Its log level defaults to +`info`; an explicit null delegates that setting to the library. Unspecified +SEC settings use the library defaults: PDR, dual-rail steady encoding, and +`max_k=32`. Non-null SEC options cannot be used with LEC. + +## SEC verification + +For example, call `create_yaml_and_run_kepler_formal` with: + +```json +{ + "input_paths": ["/designs/reference.v", "/designs/candidate.v"], + "liberty_files": ["/designs/cells.lib"], + "verification": "sec", + "sec_engine": "pdr", + "sec_encoding": "dual_rail_steady", + "max_k": 32, + "report_skipped_outputs": true, + "allowed_output_dir": "/verification-output" +} +``` + +Use the returned `verdict` to distinguish `equivalent`, `different`, +`inconclusive`, and other outcomes. A completed verification is not necessarily +an equivalence proof, and the native exit code is not a substitute for the +verdict. + +## Results and diagnostics + +`verification_result` contains every field from `VerificationResult`, including +the bound, reason, coverage counts, unproven outputs, and skipped observed +outputs. It also includes the computed `equivalent`, `conclusive`, and +`coverage_percent` properties. + +With `report_skipped_outputs=true`, `reports` returns the contents of any native +`skipped_multi_driver_pos.txt`, `skipped_no_driver_pos.txt`, and +`skipped_logical_loop_pos.txt` files before the temporary worker directory is +deleted. Reports can also accompany a failed or timed-out verification. Empty +reports are preserved as empty strings; these files are not copied elsewhere. + +## Version and capability discovery + +Call `get_kepler_formal_info` without arguments for the installed Kepler version +and git revision, NajaEDA version, supported enums/statuses, Python option +defaults, and result field names. This exposes `version()` and `git_hash()` +through MCP and reads the API metadata from the installed package. + +`NativeDesign` and `from_najaeda()` operate on live Python/C++ objects within +one process. The worker uses this interface internally; raw pointers and handles +from another process cannot be sent through MCP JSON. The `najaeda` export is a +compatibility alias for the separate NajaEDA package, not a remote netlist API. + +The tests compare MCP option names, enum choices, and returned fields against +the installed Python API, so missing coverage is detected when the dependency +is upgraded. CLI-only features absent from the Python API, including CNF and +BTOR2 export, remain unsupported and are explicitly rejected. diff --git a/kepler_formal_mcp/capabilities.py b/kepler_formal_mcp/capabilities.py new file mode 100644 index 0000000..13341da --- /dev/null +++ b/kepler_formal_mcp/capabilities.py @@ -0,0 +1,55 @@ +"""Describe the installed Kepler Formal API from an isolated worker.""" + +from __future__ import annotations + +from dataclasses import asdict, fields +from enum import Enum +from importlib import metadata +import platform +from typing import Any + + +def _json_value(value: Any) -> Any: + if isinstance(value, Enum): + return value.value + if isinstance(value, dict): + return {key: _json_value(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [_json_value(item) for item in value] + return value + + +def get_capabilities() -> dict[str, Any]: + """Return installed API metadata without constructing or loading designs. + + Import this function freely, but call it only in the worker: importing the + native package there keeps any native output away from MCP stdout. + """ + import kepler_formal + + enum_names = ( + "VerificationMode", "Solver", "SecEngine", "SecEncoding", "VerificationStatus", + ) + result_fields = [field.name for field in fields(kepler_formal.VerificationResult)] + result_fields.extend( + name for name, attribute in vars(kepler_formal.VerificationResult).items() + if isinstance(attribute, property) and not name.startswith("_") + ) + return { + "status": "success", + "kepler_formal_version": kepler_formal.version(), + "kepler_formal_git_hash": kepler_formal.git_hash(), + "najaeda_version": metadata.version("najaeda"), + "python_version": platform.python_version(), + "enums": { + name: [item.value for item in getattr(kepler_formal, name)] + for name in enum_names + }, + "option_defaults": _json_value(asdict(kepler_formal.VerificationOptions())), + "result_fields": result_fields, + "public_exports": list(kepler_formal.__all__), + "in_process_only": ( + "NativeDesign and from_najaeda use local native pointer handles. " + "They stay inside the Python worker and are not transported by MCP." + ), + } diff --git a/kepler_formal_mcp/config.py b/kepler_formal_mcp/config.py index a61ef07..6c1e66e 100644 --- a/kepler_formal_mcp/config.py +++ b/kepler_formal_mcp/config.py @@ -4,10 +4,12 @@ import os from pathlib import Path -from typing import Any +from typing import Any, get_args import yaml +from .options import LogLevel, Mode, SecEncoding, SecEngine, Solver + SUPPORTED_KEYS = { "format", "input_paths", "liberty_files", "verification", "mode", "solver", @@ -87,13 +89,13 @@ def normalize(config: dict, yaml_path: Path, root: Path, "allow_boundary_mismatch": _boolean(config, "allow_boundary_mismatch"), "report_skipped_outputs": _boolean(config, "report_skipped_outputs"), } - for key, choices in (("mode", ("lec", "sec")), - ("solver", ("kissat", "cadical", "glucose")), - ("log_level", ("info", "debug"))): + for key, choices in (("mode", get_args(Mode)), + ("solver", get_args(Solver)), + ("log_level", (*get_args(LogLevel), None))): if result[key] not in choices: - raise ValueError(f"{key} must be one of: {', '.join(choices)}") - for key, choices in (("sec_engine", ("pdr", "k_induction", "imc")), - ("sec_encoding", ("dual_rail_steady", "binary"))): + raise ValueError(f"{key} must be one of: {', '.join(map(str, choices))}") + for key, choices in (("sec_engine", get_args(SecEngine)), + ("sec_encoding", get_args(SecEncoding))): value = config.get(key) if value is not None and value not in choices: raise ValueError(f"{key} must be one of: {', '.join(choices)}") diff --git a/kepler_formal_mcp/options.py b/kepler_formal_mcp/options.py new file mode 100644 index 0000000..b0fdd9a --- /dev/null +++ b/kepler_formal_mcp/options.py @@ -0,0 +1,14 @@ +"""JSON-compatible option choices shared by MCP schemas and YAML validation. + +The API coverage tests compare these choices with the installed Kepler enums. +Keeping native imports out of the server protects the MCP stdio channel. +""" + +from typing import Literal + + +Mode = Literal["lec", "sec"] +Solver = Literal["kissat", "cadical", "glucose"] +SecEngine = Literal["pdr", "k_induction", "imc"] +SecEncoding = Literal["dual_rail_steady", "binary"] +LogLevel = Literal["info", "debug"] diff --git a/kepler_formal_mcp/runner.py b/kepler_formal_mcp/runner.py index 136cbdb..4580e66 100644 --- a/kepler_formal_mcp/runner.py +++ b/kepler_formal_mcp/runner.py @@ -4,11 +4,35 @@ import json from pathlib import Path +import stat import subprocess import sys import tempfile +REPORT_FILENAMES = ( + "skipped_multi_driver_pos.txt", + "skipped_no_driver_pos.txt", + "skipped_logical_loop_pos.txt", +) + + +def read_reports(directory: Path) -> dict[str, str]: + """Collect known native diagnostics before the isolated workspace is removed.""" + reports = {} + for name in REPORT_FILENAMES: + report = directory / name + try: + mode = report.lstat().st_mode + except FileNotFoundError: + continue + # Do not follow symlinks or read directories/devices created in the + # worker directory. Report content is returned, never copied elsewhere. + if stat.S_ISREG(mode): + reports[name] = report.read_text(encoding="utf-8", errors="replace") + return reports + + def tail(text: str | bytes | None) -> str: if isinstance(text, bytes): text = text.decode("utf-8", errors="replace") @@ -23,11 +47,9 @@ def error_result(message: str, yaml_path: Path | None = None) -> dict: } -def run(request: dict, yaml_path: Path, timeout_seconds: int) -> dict: +def _invoke_worker(request: dict, timeout_seconds: int, yaml_path: Path | None = None) -> dict: if type(timeout_seconds) is not int or timeout_seconds <= 0: raise ValueError("timeout_seconds must be a positive integer") - log_path = Path(request["log_file"]) - log_path.parent.mkdir(parents=True, exist_ok=True) # Native code can write directly to stdout and cannot be interrupted by a # Python thread timeout. A worker keeps both behaviors away from MCP stdio. with tempfile.TemporaryDirectory(prefix="kepler-mcp-") as directory: @@ -42,24 +64,48 @@ def run(request: dict, yaml_path: Path, timeout_seconds: int) -> dict: ) except subprocess.TimeoutExpired as error: result = error_result(f"Kepler-Formal timed out after {timeout_seconds} seconds", yaml_path) - result.update(stdout_tail=tail(error.stdout), generated_log_file=str(log_path)) + result.update(stdout_tail=tail(error.stdout), reports=read_reports(work)) return result if not result_path.is_file(): - result = error_result("Kepler Python worker exited without a verification result", yaml_path) + result = error_result("Kepler Python worker exited without a result", yaml_path) result.update(exit_code=completed.returncode, stdout_tail=tail(completed.stdout), stderr_tail=tail(completed.stderr) or result["stderr_tail"], - generated_log_file=str(log_path)) + reports=read_reports(work)) return result - verification = json.loads(result_path.read_text(encoding="utf-8")) - status = "error" if completed.returncode != 0 or verification["status"] == "error" else "success" + payload = json.loads(result_path.read_text(encoding="utf-8")) + status = "error" if completed.returncode != 0 or payload["status"] == "error" else "success" return { "status": status, - "exit_code": verification["exit_code"], - "verdict": verification["status"], - "verification_result": verification, - "yaml_config": str(yaml_path), - "generated_log_file": str(log_path), + "exit_code": completed.returncode, + "payload": payload, "stdout_tail": tail(completed.stdout), "stderr_tail": tail(completed.stderr), - "log_tail": tail(log_path.read_text(encoding="utf-8", errors="replace")) if log_path.is_file() else "", + "reports": read_reports(work), } + + +def get_info() -> dict: + """Query installed versions and API choices without importing native code here.""" + result = _invoke_worker({"operation": "info"}, timeout_seconds=30) + payload = result.pop("payload", {}) + # Preserve process failures even if native code wrote a successful payload. + status = result["status"] + result.update(payload) + result["status"] = status + return result + + +def run(request: dict, yaml_path: Path, timeout_seconds: int) -> dict: + if type(timeout_seconds) is not int or timeout_seconds <= 0: + raise ValueError("timeout_seconds must be a positive integer") + log_path = Path(request["log_file"]) + log_path.parent.mkdir(parents=True, exist_ok=True) + result = _invoke_worker(request, timeout_seconds, yaml_path) + verification = result.pop("payload", None) + result.update(yaml_config=str(yaml_path), generated_log_file=str(log_path)) + if verification is not None: + result.update(exit_code=verification["exit_code"], verdict=verification["status"], + verification_result=verification) + result["log_tail"] = (tail(log_path.read_text(encoding="utf-8", errors="replace")) + if log_path.is_file() else "") + return result diff --git a/kepler_formal_mcp/server.py b/kepler_formal_mcp/server.py index c735ab5..d598091 100644 --- a/kepler_formal_mcp/server.py +++ b/kepler_formal_mcp/server.py @@ -11,11 +11,27 @@ import yaml from . import config, runner +from .options import LogLevel, Mode, SecEncoding, SecEngine, Solver app = FastMCP("kepler-formal") +@app.tool() +def get_kepler_formal_info() -> str: + """Report the installed versions, build revision, and supported Python API. + + Lists verification modes, solvers, SEC engines/encodings, option defaults, + result fields and statuses from the installed Kepler library. Native design + handles are process-local Python objects and cannot be passed through MCP. + """ + try: + result = runner.get_info() + except (OSError, ValueError) as error: + result = runner.error_result(str(error)) + return json.dumps(result, indent=2) + + @app.tool() def run_kepler_formal_yaml( yaml_file: str, @@ -47,17 +63,17 @@ def create_yaml_and_run_kepler_formal( input_paths: list[str], liberty_files: list[str], yaml_output_path: str = "test_config_verilog.yaml", - log_level: str = "info", - solver: str = "kissat", + log_level: LogLevel | None = "info", + solver: Solver = "kissat", cnf_export: bool = False, cnf_export_path: str = "./sat.cnf", log_file_name: str | None = None, allowed_output_dir: str | None = None, timeout_seconds: int = 600, - verification: str = "lec", + verification: Mode = "lec", max_k: int | None = None, - sec_engine: str | None = None, - sec_encoding: str | None = None, + sec_engine: SecEngine | None = None, + sec_encoding: SecEncoding | None = None, allow_boundary_mismatch: bool = False, report_skipped_outputs: bool = False, ) -> str: @@ -89,7 +105,8 @@ def create_yaml_and_run_kepler_formal( if type(timeout_seconds) is not int or timeout_seconds <= 0: raise ValueError("timeout_seconds must be a positive integer") document["log_file"] = request["log_file"] - document = {key: value for key, value in document.items() if value is not None} + document = {key: value for key, value in document.items() + if value is not None or key == "log_level"} text = yaml.safe_dump(document, sort_keys=False) yaml_path.parent.mkdir(parents=True, exist_ok=True) yaml_path.write_text(text, encoding="utf-8") diff --git a/kepler_formal_mcp/worker.py b/kepler_formal_mcp/worker.py index 3d3a8fc..6125c81 100644 --- a/kepler_formal_mcp/worker.py +++ b/kepler_formal_mcp/worker.py @@ -75,7 +75,15 @@ def main(argv: list[str] | None = None) -> int: arguments = parser.parse_args(argv) try: request = json.loads(arguments.request.read_text(encoding="utf-8")) - result = verify(request) + operation = request.get("operation", "verify") + if operation == "verify": + result = verify(request) + elif operation == "info": + from .capabilities import get_capabilities + + result = get_capabilities() + else: + raise ValueError(f"Unknown worker operation: {operation!r}") except Exception as error: traceback.print_exc(file=sys.stderr) result = { diff --git a/tests/test_api_coverage.py b/tests/test_api_coverage.py new file mode 100644 index 0000000..c1efe09 --- /dev/null +++ b/tests/test_api_coverage.py @@ -0,0 +1,148 @@ +"""Keep the MCP verification surface aligned with the installed public KF API.""" + +from __future__ import annotations + +import asyncio +from dataclasses import fields +import inspect +from itertools import product +import json +from pathlib import Path +import tempfile +from typing import get_args +import unittest + +import kepler_formal as kf + +from kepler_formal_mcp import server + + +class PublishedApiCoverageTest(unittest.TestCase): + def setUp(self): + directory = tempfile.TemporaryDirectory(prefix="kepler api coverage ") + self.addCleanup(directory.cleanup) + self.root = Path(directory.name).resolve() + self.reference = self.root / "reference.v" + self.candidate = self.root / "candidate.v" + self.reference.write_text( + "module top(input a, output y); assign y = a; endmodule\n", + encoding="utf-8", + ) + + def compare(self, expected: str, **options) -> dict: + expression = "a" if expected == "equivalent" else "1'b0" + self.candidate.write_text( + f"module top(input a, output y); assign y = {expression}; endmodule\n", + encoding="utf-8", + ) + response = json.loads(server.create_yaml_and_run_kepler_formal( + input_paths=[str(self.reference), str(self.candidate)], + liberty_files=[], + yaml_output_path=str(self.root / "comparison.yaml"), + allowed_output_dir=str(self.root), + timeout_seconds=30, + **options, + )) + self.assertEqual(response["status"], "success", response) + self.assertEqual(response["verdict"], expected, response) + result = response["verification_result"] + self.assertEqual(result["status"], expected, response) + self.assertEqual(result["equivalent"], expected == "equivalent", response) + self.assertTrue(result["conclusive"], response) + # Every public result field is preserved, including coverage details and + # diagnostics which callers cannot reconstruct from an exit code. + self.assertEqual( + set(result), + {field.name for field in fields(kf.VerificationResult)} + | {"equivalent", "conclusive", "coverage_percent"}, + ) + self.assertIn(result["status"], {status.value for status in kf.VerificationStatus}) + return response + + def test_every_published_solver_handles_both_lec_verdicts(self): + for solver, expected in product(kf.Solver, ("equivalent", "different")): + with self.subTest(solver=solver.value, expected=expected): + result = self.compare(expected, solver=solver.value) + self.assertEqual(result["verification_result"]["verification"], "lec") + + def test_every_sec_engine_and_encoding_handles_both_verdicts(self): + for engine, encoding, expected in product( + kf.SecEngine, kf.SecEncoding, ("equivalent", "different") + ): + with self.subTest(engine=engine.value, encoding=encoding.value, expected=expected): + result = self.compare( + expected, + verification="sec", + max_k=2, + sec_engine=engine.value, + sec_encoding=encoding.value, + ) + verification = result["verification_result"] + self.assertEqual(verification["verification"], "sec") + self.assertEqual(verification["total_outputs"], 1) + self.assertEqual(verification["covered_outputs"], 1) + self.assertEqual(verification["coverage_percent"], 100.0) + + def test_none_log_level_is_supported_like_the_python_api(self): + self.compare("equivalent", log_level=None) + + def test_boundary_mismatch_requires_explicit_permission(self): + # The LEC boundary check concerns input names. Renaming this input + # exercises the flag; simply adding an output does not. + self.candidate.write_text( + "module top(input b, output y); assign y = b; endmodule\n", + encoding="utf-8", + ) + arguments = { + "input_paths": [str(self.reference), str(self.candidate)], + "liberty_files": [], + "yaml_output_path": str(self.root / "boundary.yaml"), + "allowed_output_dir": str(self.root), + "timeout_seconds": 30, + } + rejected = json.loads(server.create_yaml_and_run_kepler_formal(**arguments)) + self.assertEqual(rejected["status"], "error", rejected) + self.assertIn("boundary mismatch", rejected["verification_result"]["reason"].lower()) + permitted = json.loads(server.create_yaml_and_run_kepler_formal( + **arguments, allow_boundary_mismatch=True, + )) + self.assertEqual(permitted["status"], "success", permitted) + self.assertEqual(permitted["verdict"], "equivalent", permitted) + + def test_all_public_verification_options_have_tool_arguments(self): + renamed = {"mode": "verification", "log_file": "log_file_name"} + signature = inspect.signature(server.create_yaml_and_run_kepler_formal) + for field in fields(kf.VerificationOptions): + argument = renamed.get(field.name, field.name) + with self.subTest(option=field.name): + self.assertIn(argument, signature.parameters) + + def test_advertised_enums_match_the_installed_python_api(self): + from kepler_formal_mcp import options + + advertised = { + "verification": (options.Mode, kf.VerificationMode), + "solver": (options.Solver, kf.Solver), + "sec_engine": (options.SecEngine, kf.SecEngine), + "sec_encoding": (options.SecEncoding, kf.SecEncoding), + } + tools = asyncio.run(server.app.list_tools()) + tool = next(tool for tool in tools if tool.name == "create_yaml_and_run_kepler_formal") + properties = tool.inputSchema["properties"] + + def enum_values(schema): + values = set(schema.get("enum", ())) + for alternative in schema.get("anyOf", ()): + values.update(enum_values(alternative)) + return values + + for argument, (alias, native_enum) in advertised.items(): + expected = {member.value for member in native_enum} + with self.subTest(argument=argument): + self.assertEqual(set(get_args(alias)), expected) + self.assertEqual(enum_values(properties[argument]), expected) + self.assertEqual(enum_values(properties["log_level"]), set(get_args(options.LogLevel))) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_capabilities.py b/tests/test_capabilities.py new file mode 100644 index 0000000..0f23cbd --- /dev/null +++ b/tests/test_capabilities.py @@ -0,0 +1,76 @@ +"""Query installed API metadata through the same isolated worker as MCP.""" + +from __future__ import annotations + +from importlib import metadata +import json +from pathlib import Path +import platform +import subprocess +import sys +import tempfile +import unittest + + +class CapabilitiesTest(unittest.TestCase): + def run_worker(self, request): + with tempfile.TemporaryDirectory(prefix="kepler capabilities ") as directory: + root = Path(directory) + request_path, result_path = root / "request.json", root / "result.json" + request_path.write_text(json.dumps(request), encoding="utf-8") + completed = subprocess.run( + [sys.executable, "-m", "kepler_formal_mcp.worker", str(request_path), str(result_path)], + cwd=root, capture_output=True, text=True, encoding="utf-8", errors="replace", + timeout=30, check=False, + ) + self.assertTrue(result_path.is_file(), completed.stderr) + return completed, json.loads(result_path.read_text(encoding="utf-8")) + + def test_info_reports_installed_api_without_design_inputs(self): + completed, result = self.run_worker({"operation": "info"}) + self.assertEqual(completed.returncode, 0, completed.stderr) + self.assertEqual(result["status"], "success") + self.assertEqual(result["kepler_formal_version"], metadata.version("kepler-formal")) + self.assertEqual(result["najaeda_version"], metadata.version("najaeda")) + self.assertEqual(result["python_version"], platform.python_version()) + self.assertRegex(result["kepler_formal_git_hash"], r"^[0-9a-f]{7,40}$") + self.assertEqual(result["enums"]["VerificationMode"], ["lec", "sec"]) + self.assertEqual(set(result["enums"]["Solver"]), {"kissat", "cadical", "glucose"}) + self.assertIn("k_induction", result["enums"]["SecEngine"]) + self.assertIn("dual_rail_steady", result["enums"]["SecEncoding"]) + self.assertIn("partially_proved", result["enums"]["VerificationStatus"]) + self.assertEqual(result["option_defaults"], { + "mode": "lec", "solver": "kissat", "max_k": None, + "sec_engine": None, "sec_encoding": None, + "allow_boundary_mismatch": False, "report_skipped_outputs": False, + "log_file": None, "log_level": None, + }) + self.assertTrue({ + "status", "exit_code", "reason", "unproven_outputs", "skipped_observed_outputs", + "equivalent", "conclusive", "coverage_percent", + } <= set(result["result_fields"])) + self.assertTrue({"verify_designs", "version", "git_hash", "NativeDesign", "from_najaeda"} + <= set(result["public_exports"])) + self.assertIn("not transported by MCP", result["in_process_only"]) + + def test_unknown_operation_returns_a_structured_worker_error(self): + completed, result = self.run_worker({"operation": "missing"}) + self.assertEqual(completed.returncode, 1) + self.assertEqual(result["status"], "error") + self.assertIn("ValueError: Unknown worker operation", result["reason"]) + + def test_importing_capabilities_does_not_load_the_native_package(self): + completed = subprocess.run( + [sys.executable, "-c", ( + "import sys; import kepler_formal_mcp.capabilities; " + "assert 'kepler_formal' not in sys.modules; " + "assert 'najaeda' not in sys.modules" + )], + capture_output=True, text=True, timeout=30, check=False, + ) + self.assertEqual(completed.returncode, 0, completed.stderr) + self.assertEqual(completed.stdout, "") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_reports.py b/tests/test_reports.py new file mode 100644 index 0000000..a9487d3 --- /dev/null +++ b/tests/test_reports.py @@ -0,0 +1,116 @@ +"""Keep native skipped-output diagnostics available after worker cleanup.""" + +import json +from pathlib import Path +import subprocess +import tempfile +import unittest +from unittest.mock import patch + +from kepler_formal_mcp import config, runner + + +class NativeReportTests(unittest.TestCase): + def test_published_library_returns_nonempty_no_driver_report(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + liberty = root / "cells.lib" + liberty.write_text('''library(test) { + cell(BUF) { + pin(A) { direction : input; } + pin(Y) { direction : output; function : "A"; } + } +} +''') + reference = root / "reference.v" + reference.write_text('''module top(input a, input b, output good, output no_driver); + wire undriven; + BUF g(.A(a), .Y(good)); + BUF f(.A(undriven), .Y(no_driver)); +endmodule +''') + candidate = root / "candidate.v" + candidate.write_text('''module top(input a, input b, output good, output no_driver); + BUF g(.A(b), .Y(good)); + BUF f(.A(a), .Y(no_driver)); +endmodule +''') + yaml_path = root / "request.yaml" + request = config.normalize({ + "input_paths": [str(reference), str(candidate)], + "liberty_files": [str(liberty)], + "report_skipped_outputs": True, + }, yaml_path, root) + + result = runner.run(request, yaml_path, 30) + + self.assertEqual("success", result["status"], result) + self.assertEqual("different", result["verdict"], result) + report = result["reports"]["skipped_no_driver_pos.txt"] + self.assertIn("no_driver", report) + self.assertIn("no drivers", report) + self.assertEqual([], list(root.glob("skipped*.txt"))) + + def test_reports_survive_worker_cleanup_without_copying_arbitrary_files(self): + with tempfile.TemporaryDirectory() as directory: + output = Path(directory) + worker_directories = [] + + def worker(command, **kwargs): + work = Path(kwargs["cwd"]) + worker_directories.append(work) + (work / "skipped_multi_driver_pos.txt").write_text("conflict: multiple drivers\n") + (work / "skipped_no_driver_pos.txt").write_text("floating: no driver\n") + (work / "skipped_logical_loop_pos.txt").write_text("feedback: logical loop\n") + (work / "unrelated.txt").write_text("not a report") + Path(command[-1]).write_text(json.dumps({"status": "equivalent", "exit_code": 0})) + return subprocess.CompletedProcess(command, 0, "", "") + + with patch.object(runner.subprocess, "run", side_effect=worker): + result = runner.run({"log_file": str(output / "verify.log")}, output / "request.yaml", 30) + + self.assertEqual({ + "skipped_multi_driver_pos.txt": "conflict: multiple drivers\n", + "skipped_no_driver_pos.txt": "floating: no driver\n", + "skipped_logical_loop_pos.txt": "feedback: logical loop\n", + }, result["reports"]) + self.assertFalse(worker_directories[0].exists()) + self.assertEqual([], list(output.iterdir())) + + def test_partial_reports_survive_a_worker_timeout(self): + with tempfile.TemporaryDirectory() as directory: + output = Path(directory) + + def worker(command, **kwargs): + work = Path(kwargs["cwd"]) + (work / "skipped_no_driver_pos.txt").write_text("floating: no driver\n") + raise subprocess.TimeoutExpired(command, kwargs["timeout"]) + + with patch.object(runner.subprocess, "run", side_effect=worker): + result = runner.run({"log_file": str(output / "verify.log")}, output / "request.yaml", 30) + + self.assertEqual("error", result["status"]) + self.assertEqual({"skipped_no_driver_pos.txt": "floating: no driver\n"}, result["reports"]) + + def test_directories_are_not_read_as_reports(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "skipped_multi_driver_pos.txt").mkdir() + self.assertEqual({}, runner.read_reports(root)) + + def test_report_symlinks_are_not_followed(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + outside = root / "outside.txt" + outside.write_text("outside worker workspace") + work = root / "worker" + work.mkdir() + try: + (work / "skipped_multi_driver_pos.txt").symlink_to(outside) + except (NotImplementedError, OSError): + self.skipTest("Creating symlinks is unavailable") + self.assertEqual({}, runner.read_reports(work)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_server.py b/tests/test_server.py index d904599..f900a4f 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -274,6 +274,11 @@ async def exercise(): tools = await session.list_tools() names = {tool.name for tool in tools.tools} self.assertTrue({"run_kepler_formal_yaml", "create_yaml_and_run_kepler_formal"} <= names) + response = await session.call_tool("get_kepler_formal_info", {}) + self.assertFalse(response.isError, response) + information = json.loads(next(item.text for item in response.content if item.type == "text")) + self.assertEqual(information["status"], "success", information) + self.assertIn("sec", information["enums"]["VerificationMode"]) for content, verdict in [(PASS_THROUGH, "equivalent"), (CONSTANT_OUTPUT, "different")]: self.candidate.write_text(content, encoding="utf-8") response = await session.call_tool("create_yaml_and_run_kepler_formal", {