Add persistent verification sessions and keep MCP transport responsive

README.md

@@ -37,8 +37,11 @@ The existing `python server.py` launcher also works using the installed environm
 - `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.
+- `open_session`, `load_designs`, `verify_session`, `close_session`: load designs once and reuse one Python worker for repeated checks.
+- `attach_session`: bind to a bridge in a Python interpreter that already owns NajaEDA designs.
+- `set_session`, `list_sessions`: select and inspect reusable sessions.
 
-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`.
+The two file tools use a separate Python worker for each call. Session tools retain one worker or attach to a caller's interpreter, preserving loaded designs between checks. In both cases, native solver output is separate from MCP messages, and Kepler verifies the existing NajaEDA design handles. `get_kepler_formal_info` reuses the selected session when one exists.
 
 A minimal YAML file is:
 
@@ -61,6 +64,8 @@ Supported verification settings are `verification` (`lec` or `sec`; `mode` is an
 
 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.
 
+See [persistent sessions and attaching to live NajaEDA designs](docs/sessions.md) to reuse loaded netlists across calls or verify designs already owned by another Python process.
+
 ## Results
 
 The returned JSON separates tool execution from the verification verdict:

docs/python-api.md

@@ -6,7 +6,8 @@ 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:
+`create_yaml_and_run_kepler_formal`, `verify_session`, and existing YAML configurations
+(attached sessions cannot request process-relative skipped-output report files):
 
 | Python option | MCP argument | YAML key | Accepted values |
 | --- | --- | --- | --- |
@@ -66,11 +67,15 @@ 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.
+With an active session it queries that interpreter; otherwise it uses a temporary worker.
 
 `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.
+To reuse loaded designs or bind to an existing interpreter, use the
+[session tools and Python bridge](sessions.md). Verification then runs in the
+process that owns those designs.
 
 The tests compare MCP option names, enum choices, and returned fields against
 the installed Python API, so missing coverage is detected when the dependency

docs/sessions.md

@@ -0,0 +1,84 @@
+# Persistent verification sessions
+
+Sessions retain loaded designs between MCP calls. Use a managed session to load
+files once, or attach to a Python process that already owns NajaEDA designs.
+The existing YAML tools still run each comparison in an independent worker.
+
+## Managed sessions
+
+Call these MCP tools in order:
+
+1. `open_session(allowed_output_dir="/absolute/path/results")` creates and selects
+   a session. Keep its returned `session_id`.
+2. `load_designs(input_paths=["/absolute/reference.v", "/absolute/candidate.v"],
+   liberty_files=[])` loads the pair as `reference` and `candidate`.
+3. `verify_session()` checks those loaded designs. Repeated calls reuse the same
+   Python process and native netlists; input files are not reread.
+4. `close_session()` releases the session and its process.
+
+`verify_session` accepts the same verification settings as the file tools:
+`verification`, `solver`, `max_k`, `sec_engine`, `sec_encoding`,
+`allow_boundary_mismatch`, `report_skipped_outputs`, `log_level`, and
+`log_file_name`. For example, use `verification="sec", max_k=10,
+sec_engine="pdr", sec_encoding="binary"` for a bounded SEC run.
+
+Open more sessions for independent designs. `list_sessions()` shows them;
+`set_session(session_id=...)` changes the active one. Pass `session_id` directly
+to loading, verification, or closing to target a session without switching.
+Custom `names=["golden", "revised"]` in `load_designs` can be selected with
+`verify_session(design1="golden", design2="revised")`.
+
+## Attach to existing NajaEDA designs
+
+Run the bridge in the Python process that owns your designs, using the same
+installed Kepler Formal and NajaEDA packages as the MCP server:
+
+```python
+from pathlib import Path
+from kepler_formal_mcp.session_bridge import SessionBridge
+
+# reference and candidate are your already loaded NajaEDA design objects.
+# Raw SNLDesigns, NajaEDA Instances, and KF NativeDesign handles are accepted.
+with SessionBridge(output_dir=Path("results").resolve()) as bridge:
+    bridge.register_design("reference", reference)
+    bridge.register_design("candidate", candidate)
+    print("Attach MCP using:", bridge.connection_file)
+    input("Keep this process running; press Enter when finished. ")
+```
+
+Then call `attach_session(connection_file="/path/printed/by/the/bridge")` from
+MCP, followed by `verify_session()`. Registering captures the selected design:
+subsequent NajaEDA top-design changes do not retarget the registered handle.
+Verification runs inside the owning Python process on its existing pointers.
+Only authenticated local commands and results cross the connection; netlists
+are not serialized or copied between processes.
+
+The connection file contains a credential. Keep it private and attach only to
+a bridge you intend to control. The bridge offers registered-design operations,
+not arbitrary Python execution.
+
+Coordinate edits in the owning Python process with the bridge's lock:
+
+```python
+with bridge.lock:
+    # Edit reference or candidate with NajaEDA here.
+    ...
+```
+
+The next verification sees those edits. Do not destroy registered designs or
+their universe while the bridge uses them. `close_session()` detaches MCP from
+an attached session; it leaves the bridge and caller's netlists alive.
+`bridge.close()` stops the bridge and releases its borrowed references without
+destroying the caller's universe.
+
+## Timeouts and outputs
+
+Both session types restrict logs to their selected output directory. Managed
+sessions can produce skipped-output reports in their own working directory.
+Attached sessions reject `report_skipped_outputs=True`, because native reports
+would otherwise write into the owning application's working directory.
+
+A managed-session timeout terminates its process and invalidates that session;
+open and load a new one to retry. An attached-session timeout stops waiting but
+does not kill the owning application. Verification may still be running there;
+overlapping operations are rejected until it finishes.

kepler_formal_mcp/runner.py

@@ -59,7 +59,8 @@ def _invoke_worker(request: dict, timeout_seconds: int, yaml_path: Path | None =
         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",
+                cwd=work, stdin=subprocess.DEVNULL,
+                capture_output=True, text=True, encoding="utf-8", errors="replace",
                 timeout=timeout_seconds, check=False,
             )
         except subprocess.TimeoutExpired as error:

kepler_formal_mcp/server.py

@@ -12,12 +12,19 @@
 
 from . import config, runner
 from .options import LogLevel, Mode, SecEncoding, SecEngine, Solver
+from .tool_dispatch import threaded_tool
+from . import session_tools
+from .session_tools import (
+    attach_session, close_session, list_sessions, load_designs,
+    open_session, set_session, verify_session,
+)
 
 
 app = FastMCP("kepler-formal")
+session_tools.register(app)
 
 
-@app.tool()
+@threaded_tool(app)
 def get_kepler_formal_info() -> str:
     """Report the installed versions, build revision, and supported Python API.
 
@@ -26,13 +33,14 @@ def get_kepler_formal_info() -> str:
     handles are process-local Python objects and cannot be passed through MCP.
     """
     try:
-        result = runner.get_info()
+        result = (session_tools.manager.call({"operation": "info"}, timeout_seconds=30)
+                  if session_tools.manager.active_session_id is not None else runner.get_info())
     except (OSError, ValueError) as error:
         result = runner.error_result(str(error))
     return json.dumps(result, indent=2)
 
 
-@app.tool()
+@threaded_tool(app)
 def run_kepler_formal_yaml(
     yaml_file: str,
     timeout_seconds: int = 600,
@@ -58,7 +66,7 @@ def run_kepler_formal_yaml(
     return json.dumps(result, indent=2)
 
 
-@app.tool()
+@threaded_tool(app)
 def create_yaml_and_run_kepler_formal(
     input_paths: list[str],
     liberty_files: list[str],
@@ -120,4 +128,7 @@ def create_yaml_and_run_kepler_formal(
 def main() -> None:
     logging.basicConfig(level=logging.INFO, stream=sys.stderr,
                         format="[kepler-mcp] [%(levelname)s] %(message)s")
-    app.run()
+    try:
+        app.run()
+    finally:
+        session_tools.manager.close_all()

kepler_formal_mcp/session_backend.py

@@ -0,0 +1,188 @@
+"""Persistent design handles hosted beside their native NajaEDA universe."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from threading import RLock
+from typing import Any
+from uuid import uuid4
+
+from .runner import REPORT_FILENAMES, read_reports, tail
+from .verification import OPTION_NAMES, load_designs, verify_loaded
+
+
+# Naja's universe and verification caches are shared throughout an interpreter,
+# including when a caller starts more than one bridge in that interpreter.
+_NATIVE_LOCK = RLock()
+
+
+class DesignSession:
+    """Host managed netlists or borrow live caller designs without taking ownership.
+
+    Attached callers must hold ``lock`` while changing or destroying registered
+    designs, so edits cannot race a request received by the bridge.
+    """
+
+    def __init__(self, output_dir: str | Path, owned: bool = False):
+        from najaeda import naja
+
+        self.lock = _NATIVE_LOCK
+        self.session_id = str(uuid4())
+        self.kind = "managed" if owned else "attached"
+        self.pid = os.getpid()
+        self.output_dir = Path(output_dir).expanduser().resolve()
+        self.output_dir.mkdir(parents=True, exist_ok=True)
+        self._owned = owned
+        self._closed = False
+        self._designs: dict[str, Any] = {}
+        self._databases: list[Any] = []
+        self._input_files: set[Path] = set()
+        self._workspace = Path.cwd().resolve()
+        self._universe = None
+        if owned:
+            if naja.NLUniverse.get() is not None:
+                raise RuntimeError("A managed session requires a fresh Naja universe")
+            self._universe = naja.NLUniverse.create()
+
+    def _require_open(self):
+        if self._closed:
+            raise RuntimeError("The design session is closed")
+        if self._owned:
+            from najaeda import naja
+
+            if naja.NLUniverse.get() is not self._universe:
+                raise ReferenceError("The managed session's universe is no longer active")
+
+    def _name(self, name: Any) -> str:
+        if not isinstance(name, str) or not name.strip():
+            raise ValueError("Design names must be nonempty strings")
+        return name
+
+    def _metadata(self, name: str) -> dict[str, Any]:
+        result = {"name": name}
+        try:
+            result.update(design_name=self._designs[name].najaeda_design.getName(), valid=True)
+        except (RuntimeError, ReferenceError) as error:
+            result.update(valid=False, reason=str(error))
+        return result
+
+    def _inspect(self) -> dict[str, Any]:
+        return {
+            "status": "success", "session_id": self.session_id, "kind": self.kind,
+            "pid": self.pid, "output_dir": str(self.output_dir),
+            "designs": [self._metadata(name) for name in self._designs],
+        }
+
+    def register_design(self, name: str, design: Any) -> dict[str, Any]:
+        """Capture an existing raw design or NajaEDA Instance without copying it."""
+        from kepler_formal import NativeDesign, from_najaeda
+
+        with self.lock:
+            self._require_open()
+            name = self._name(name)
+            if name in self._designs:
+                raise ValueError(f"A design is already registered as {name!r}")
+            handle = design if isinstance(design, NativeDesign) else from_najaeda(design)
+            self._designs[name] = handle
+            return {"status": "success", "session_id": self.session_id, **self._metadata(name)}
+
+    def _load(self, request: dict[str, Any]) -> dict[str, Any]:
+        if not self._owned:
+            raise ValueError("Attached sessions borrow caller designs; load them in NajaEDA and register them")
+        names = request.get("names", ["reference", "candidate"])
+        if not isinstance(names, list) or len(names) != 2:
+            raise ValueError("names must contain exactly two design names")
+        names = [self._name(name) for name in names]
+        if names[0] == names[1] or any(name in self._designs for name in names):
+            raise ValueError("Design names must be distinct and not already registered")
+        inputs, libraries = request.get("input_paths"), request.get("liberty_files", [])
+        databases, designs = load_designs(self._universe, inputs, libraries)
+        try:
+            from kepler_formal import from_najaeda
+
+            handles = [from_najaeda(design) for design in designs]
+        except Exception:
+            for database in reversed(databases):
+                database.destroy()
+            raise
+        self._designs.update(zip(names, handles))
+        self._databases.extend(databases)
+        self._input_files.update(Path(path).resolve() for path in inputs + libraries)
+        return {**self._inspect(), "loaded": names}
+
+    def _verify(self, request: dict[str, Any]) -> dict[str, Any]:
+        names = [self._name(request.get(key)) for key in ("design1", "design2")]
+        for name in names:
+            if name not in self._designs:
+                raise ValueError(f"Unknown registered design: {name!r}")
+        raw_options = request.get("options", {})
+        if not isinstance(raw_options, dict):
+            raise ValueError("options must be a mapping")
+        unknown = set(raw_options) - set(OPTION_NAMES)
+        if unknown:
+            raise ValueError("Unsupported verification options: " + ", ".join(sorted(map(str, unknown))))
+        options = dict(raw_options)
+        reports_requested = options.get("report_skipped_outputs", False)
+        if not isinstance(reports_requested, bool):
+            raise ValueError("report_skipped_outputs must be a boolean")
+        if reports_requested and not self._owned:
+            raise ValueError("report_skipped_outputs is unavailable for attached sessions because it writes in the caller's working directory")
+
+        log_value = options.get("log_file")
+        if log_value is not None and (not isinstance(log_value, str) or not log_value.strip()):
+            raise ValueError("log_file must be a nonempty path string or null")
+        log_path = Path(log_value).expanduser() if log_value is not None else Path(f"verify-{uuid4().hex}.log")
+        log_path = (log_path if log_path.is_absolute() else self.output_dir / log_path).resolve()
+        if not log_path.is_relative_to(self.output_dir):
+            raise ValueError(f"Log path is outside the session output directory: {log_path}")
+        if log_path in self._input_files or log_path.is_dir():
+            raise ValueError("The log file must not overwrite an input file or directory")
+        options["log_file"] = str(log_path)
+        if self._owned:
+            if Path.cwd().resolve() != self._workspace:
+                raise RuntimeError("The managed session's private working directory changed")
+            for name in REPORT_FILENAMES:
+                (self._workspace / name).unlink(missing_ok=True)
+        log_path.parent.mkdir(parents=True, exist_ok=True)
+        result = verify_loaded(self._designs[names[0]], self._designs[names[1]], options)
+        return {
+            "status": "error" if result["status"] == "error" else "success",
+            "session_id": self.session_id, "exit_code": result["exit_code"],
+            "verdict": result["status"], "verification_result": result,
+            "generated_log_file": str(log_path),
+            "log_tail": tail(log_path.read_text(encoding="utf-8", errors="replace")) if log_path.is_file() else "",
+            "reports": read_reports(self._workspace) if self._owned else {},
+            "stdout_tail": "", "stderr_tail": "",
+        }
+
+    def dispatch(self, request: dict[str, Any]) -> dict[str, Any]:
+        with self.lock:
+            self._require_open()
+            if not isinstance(request, dict):
+                raise ValueError("Session requests must be mappings")
+            operation = request.get("operation")
+            if operation == "info":
+                from .capabilities import get_capabilities
+
+                return {**get_capabilities(), "session_id": self.session_id, "kind": self.kind, "pid": self.pid}
+            if operation == "inspect":
+                return self._inspect()
+            if operation == "load":
+                return self._load(request)
+            if operation == "verify":
+                return self._verify(request)
+            raise ValueError(f"Unknown session operation: {operation!r}")
+
+    def close(self) -> dict[str, Any]:
+        with self.lock:
+            if not self._closed:
+                self._designs.clear()
+                self._databases.clear()
+                if self._owned:
+                    from najaeda import naja
+
+                    if naja.NLUniverse.get() is self._universe:
+                        self._universe.destroy()
+                self._closed = True
+            return {"status": "success", "session_id": self.session_id, "closed": True}