diff --git a/README.md b/README.md index 79bcccb..03447b9 100644 --- a/README.md +++ b/README.md @@ -204,6 +204,16 @@ need it and trust every client that can reach the endpoint. --- +## Inspect An Existing Python Session + +Naja-Scope MCP can attach to designs already loaded in a Python or Jupyter +process. Select each design by its native database, library and design IDs; +switch between references without dumping or reloading. The bridge shares the +owner's editing/verification lock and only exposes read-only inspection. + +See the [live-session binding guide](examples/live-session.md) for owner setup, +MCP calls, design selection and lifecycle rules. + ## Requirements - Python 3.10+ diff --git a/README_PyPI.md b/README_PyPI.md index 7efb15c..ac69b08 100644 --- a/README_PyPI.md +++ b/README_PyPI.md @@ -163,6 +163,13 @@ Once a design is loaded, your assistant can: --- +## Inspect An Existing Python Session + +The MCP server can attach to a live Python/Jupyter process and inspect designs +by native Naja database, library and design IDs, without reloading. The owner +provides a shared lock; attached tools are read-only. See the +[live-session binding guide](examples/live-session.md). + ## Requirements - Python 3.10+ diff --git a/examples/live-session.md b/examples/live-session.md new file mode 100644 index 0000000..42f8505 --- /dev/null +++ b/examples/live-session.md @@ -0,0 +1,103 @@ +# Inspect Designs In A Live Python Or Jupyter Session + +The binding runs existing MCP inspection tools inside the process that owns the +Naja universe. It does not load, copy, dump, edit, or take ownership of designs. +Use the same installed Naja-Scope version in the owner and MCP server. + +## Start In The Owner + +After loading your designs using the raw `from najaeda import naja` API, +start a bridge in that same interpreter: + +```python +from naja_scope.binding import SessionBridge + +# existing_design_a and existing_design_b are raw naja.SNLDesign objects. +# shared_lock is the existing threading.RLock used by every editor/verifier. +scope = SessionBridge(lock=shared_lock, source_files=source_files).start() +golden_ref = scope.design_reference(existing_design_a) +edited_ref = scope.design_reference(existing_design_b) +connection_file = str(scope.connection_file) +``` + +If no other component has a lock, omit `lock=` and hold `scope.lock` around +every edit, load, top-selection change, and destruction. The bridge must share +the *exact same lock object* as any other in-process verifier or editor. +For a Kepler MCP bridge this can be its public `lock`, passed to `SessionBridge`. +Naja-Scope does not import or require Kepler or 22b. + +`source_files` is an optional sequence of paths used to resolve relative source +locations. It does not reload RTL. The warm native AST/intent link is reused if +available; attachment never re-elaborates a design to manufacture intent data. + +The references contain `session_id`, `db_id`, `library_id`, and `design_id`. +These are the same native-coordinate fields used by Kepler MCP. Each bridge +has its own session ID/connection credential even when both bridges share one +interpreter and universe. Obtain the Naja-Scope reference from this bridge; +do not send a Kepler bridge's session ID to it. No design-name map exists. +The variable names `golden_ref` and `edited_ref` have no special API meaning. + +## Attach And Select Through MCP + +Launch the usual `naja-scope-mcp` server on the same machine and user account. +The following are MCP tool names and their arguments, not a new Python client: + +```python +attach_session(connection_file=connection_file) +get_session_binding() + +# Explicit per-query selection. Paths are relative to the chosen design's top. +get_drivers(design=golden_ref, path="top.y") +get_drivers(design=edited_ref, path="top.y") +get_loads(design=edited_ref, path="top.a") +get_stats(design=edited_ref) + +# Optional default for subsequent calls on this MCP server. +set_session_design(design=golden_ref) +get_drivers(path="top.y") +``` + +All read-only inspection tools accept `design`. A per-query reference overrides +the selected default for that call only. Prefer explicit per-query references +when multiple agents share an MCP server; its default is not client-private. +Attaching without a design allows discovery but requires selection before a +design query. `attach_session(..., design=golden_ref)` can set the initial default. + +Discovery lists at most 200 non-primitive designs and reports +`designs_truncated`. This does not limit lookup: the owner can produce a reference +for any raw SNL design through `scope.design_reference(design)`. + +Each query resolves the native IDs afresh and executes under the shared lock +with a scoped inspection root. The owner's global and per-database top selections +are preserved; Naja-Scope's temporary metadata is restored even on failure. +Native cone tracing temporarily changes top selection internally, so the bridge +restores both selections under the shared lock. `trace_cone` requires an existing +global top and a top in the target database: the raw binding cannot restore an +unset top. If either is missing, it rejects tracing before mutation; the owner +can select the tops first. Other inspection tools also work with unset tops. +Results retain their normal fields and add `binding` with the owner process, +session and exact design reference. Queries see the current connectivity after +each completed edit. Naja-Scope inspection is not a formal equivalence proof. + +## Lifetime And Trust + +- Keep both the owner process and bridge alive. `detach_session()` disconnects + only the MCP server. `scope.close()` removes only this bridge and its private + connection file; neither destroys designs. +- A missing design, wrong session, invalid/out-of-range ID, or replaced universe + is rejected. There is no fallback to a local loaded design. +- Native IDs are coordinates, not generation counters. Close all bridges before + destroying/reloading designs or databases, then obtain fresh session references. + Delete-and-recreate may reuse native IDs and cannot always be detected. +- Load, reset, snapshot/export, intent-loading and arbitrary-Python tools are + blocked while attached, including when `NAJA_SCOPE_ENABLE_PYTHON` is enabled. +- Busy owners reject queries. A timeout does not cancel native work; wait for + the shared lock before editing, closing or destroying anything. +- The connection file is a credential: keep it private, never commit or log its + contents. It grants read access to designs in the owner's universe, not just + the currently selected one. The bridge binds loopback only and authenticates + requests. It is for trusted same-user clients, not an OS security sandbox. +- The owner's direct Python code can still mutate Naja. MCP read-only enforcement + does not constrain arbitrary code running in that interpreter. + +Without attachment, the existing load/query/reset workflow is unchanged. diff --git a/src/naja_scope/binding.py b/src/naja_scope/binding.py new file mode 100644 index 0000000..55e4068 --- /dev/null +++ b/src/naja_scope/binding.py @@ -0,0 +1,452 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Authenticated, read-only access to raw SNL designs in a caller's process. + +No pickle, Python evaluation, netlist transfer, or native ownership transfer. +The owner must use the same lock for edits, verification and destruction. +""" + +from __future__ import annotations + +import hmac +import inspect +import json +import os +from pathlib import Path +import secrets +import socket +import socketserver +import stat +import struct +import tempfile +import threading +import time +from uuid import uuid4 + +from .errors import ScopeError +from .design_reference import DesignReference, reference_from_design + + +PROTOCOL = "naja-scope-session-v1" +MAX_REQUEST = 1024 * 1024 +MAX_RESPONSE = 8 * 1024 * 1024 +READ_ONLY_TOOLS = frozenset({ + "status", "resolve", "find", "get_hierarchy", "get_drivers", "get_loads", + "trace_cone", "get_source", "get_module_card", "get_stats", "get_intent", +}) +# Serializes native inspection and process-global api.SESSION across bridges. +NATIVE_LOCK = threading.RLock() + + +def _receive(sock, maximum, deadline): + def exact(size): + result = bytearray() + while len(result) < size: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("Session request timed out") + sock.settimeout(remaining) + part = sock.recv(size - len(result)) + if not part: + raise ScopeError("Session connection closed before a complete response") + result.extend(part) + return result + + size = struct.unpack("!I", exact(4))[0] + if not 0 < size <= maximum: + raise ScopeError("Session message exceeds its size limit") + value = json.loads(exact(size)) + if not isinstance(value, dict): + raise ScopeError("Session message must be a JSON object") + return value + + +def _send(sock, value, maximum): + data = json.dumps(value, allow_nan=False).encode("utf-8") + if len(data) > maximum: + raise ScopeError("Session response exceeds its size limit; use smaller query limits") + sock.sendall(struct.pack("!I", len(data)) + data) + + +class _Server(socketserver.ThreadingTCPServer): + allow_reuse_address = False + daemon_threads = True + + def __init__(self, bridge): + self.bridge = bridge + self.slots = threading.BoundedSemaphore(8) + super().__init__(("127.0.0.1", 0), _Handler) + + def process_request(self, request, client_address): + if not self.slots.acquire(blocking=False): + self.shutdown_request(request) + return + try: + super().process_request(request, client_address) + except BaseException: + self.slots.release() + raise + + def process_request_thread(self, request, client_address): + try: + super().process_request_thread(request, client_address) + finally: + self.slots.release() + + +class _Handler(socketserver.BaseRequestHandler): + def handle(self): + self.request.settimeout(5) + try: + request = _receive(self.request, MAX_REQUEST, time.monotonic() + 5) + reply = self.server.bridge._dispatch(request) + _send(self.request, reply, MAX_RESPONSE) + except (OSError, TypeError, ValueError, ScopeError): + try: + _send(self.request, {"error": "Invalid or incomplete session request"}, MAX_RESPONSE) + except (OSError, ScopeError): + pass + + +class SessionBridge: + """Host caller-owned raw SNLDesign handles for read-only MCP inspection. + + Pass the owner's existing reentrant lock when another editor/verifier shares + the universe. Otherwise hold bridge.lock around every native mutation. + Closing stops this bridge only; it never resets or destroys the universe. + """ + + def __init__(self, *, lock=None, source_files=()): + from najaeda import naja + from .session import Session + + self.lock = lock if lock is not None else NATIVE_LOCK + self.session_id = uuid4().hex + self.pid = os.getpid() + self._token = secrets.token_hex(32) + self._universe = naja.NLUniverse.get() + if self._universe is None: + raise ScopeError("A bridge requires an existing Naja universe") + if isinstance(source_files, (str, bytes)): + raise TypeError("source_files must be a sequence of paths") + self._metadata = Session() + self._metadata._record_sources([os.fspath(path) for path in source_files]) + self._server = self._thread = None + self._closed = False + self._directory = None + self.connection_file = None + + def _require_open(self): + from najaeda import naja + + if self._closed or naja.NLUniverse.get() is not self._universe: + raise ScopeError("The bound design session is closed or no longer valid") + + def _resolve(self, value): + reference = DesignReference.model_validate(value) + if reference.session_id != self.session_id: + raise ScopeError("Design reference belongs to a different session") + design = self._universe.getSNLDesign(reference.native_key()) + if design is None: + raise ScopeError("Native design ID does not exist in this session") + if reference_from_design(self.session_id, design) != reference: + raise ScopeError("Native lookup returned a different design identity") + return reference, design + + def design_reference(self, design): + """Return native coordinates without registering or retaining a design.""" + with self.lock, NATIVE_LOCK: + self._require_open() + reference = reference_from_design(self.session_id, design) + self._resolve(reference) + return reference.model_dump() + + def start(self): + self._require_open() + if self._server is not None: + return self + server = _Server(self) + directory = Path(tempfile.mkdtemp(prefix="naja-scope-session-")) + connection = directory / "connection.json" + try: + with connection.open("x", encoding="utf-8") as stream: + if os.name == "posix": + os.fchmod(stream.fileno(), 0o600) + json.dump({"protocol": PROTOCOL, "host": "127.0.0.1", + "port": server.server_address[1], "token": self._token, + "session_id": self.session_id, "pid": self.pid}, stream) + self._server, self._directory, self.connection_file = server, directory, connection + self._thread = threading.Thread(target=lambda: server.serve_forever(poll_interval=0.05), + daemon=True, name="naja-scope-session") + self._thread.start() + except BaseException: + server.server_close() + connection.unlink(missing_ok=True) + directory.rmdir() + raise + return self + + def _inspect(self): + designs = [] + truncated = False + for database in self._universe.getUserDBs(): + for library in database.getLibraries(): + if library.isPrimitives(): + continue + for design in library.getSNLDesigns(): + if design.isPrimitive(): + continue + if len(designs) == 200: + truncated = True + break + designs.append({"name": design.getName(), + "reference": reference_from_design( + self.session_id, design).model_dump()}) + if truncated: + break + if truncated: + break + return {"session_id": self.session_id, "pid": self.pid, + "design_addressing": "native-id-v1", "designs": designs, + "designs_truncated": truncated} + + def _dispatch(self, request): + token = request.get("token") + if (not isinstance(token, str) or not token.isascii() + or not hmac.compare_digest(token, self._token)): + return {"error": "Session authentication failed"} + if request.get("protocol") != PROTOCOL or request.get("session_id") != self.session_id: + return {"error": "Session identity or protocol mismatch"} + if not self.lock.acquire(blocking=False): + return {"error": "Design session is busy; retry when the owner is idle"} + try: + if not NATIVE_LOCK.acquire(blocking=False): + return {"error": "Native inspection is busy; retry when idle"} + try: + from . import api, snl + + self._require_open() + operation = request.get("operation") + if operation == "inspect": + return self._inspect() + if operation == "select": + reference, _ = self._resolve(request.get("design")) + return {"session_id": self.session_id, "pid": self.pid, + "design": reference.model_dump()} + if operation != "query" or request.get("tool") not in READ_ONLY_TOOLS: + raise ScopeError("Attached sessions allow only typed read-only inspection tools") + reference, design = self._resolve(request.get("design")) + arguments = request.get("arguments", {}) + if not isinstance(arguments, dict): + raise ScopeError("Query arguments must be a JSON object") + function = getattr(api, request["tool"]) + inspect.signature(function).bind(**arguments) + previous_top = self._universe.getTopDesign() + target_db = design.getDB() + previous_db_top = target_db.getTopDesign() + # Native LogicCone changes top selection internally. The raw + # binding cannot restore an unset top, so fail before mutation. + if request["tool"] == "trace_cone" and (previous_top is None or previous_db_top is None): + raise ScopeError("trace_cone requires existing global and target-database tops; " + "select them in the owner before tracing") + previous_session = api.SESSION + try: + api.SESSION = self._metadata + with snl.inspection_design(design): + result = function(**arguments) + finally: + api.SESSION = previous_session + if previous_db_top is not None and target_db.getTopDesign() != previous_db_top: + self._universe.setTopDesign(previous_db_top) + if previous_top is not None and self._universe.getTopDesign() != previous_top: + self._universe.setTopDesign(previous_top) + return {"session_id": self.session_id, "pid": self.pid, + "design": reference.model_dump(), "result": result} + except ScopeError as error: + return error.to_dict() + except (RuntimeError, ReferenceError, TypeError, ValueError) as error: + return {"error": f"Bound inspection failed: {error}"} + finally: + NATIVE_LOCK.release() + finally: + self.lock.release() + + def close(self): + if not self.lock.acquire(blocking=False): + raise ScopeError("Design session is busy; wait before closing its bridge") + try: + if not NATIVE_LOCK.acquire(blocking=False): + raise ScopeError("Native inspection is busy; wait before closing its bridge") + try: + if self._closed: + return + self._closed = True + finally: + NATIVE_LOCK.release() + finally: + self.lock.release() + if self._server is not None: + self._server.shutdown() + self._server.server_close() + self._thread.join(timeout=5) + if self.connection_file is not None: + self.connection_file.unlink(missing_ok=True) + # Delete only our private connection directory, never caller data. + try: + self._directory.rmdir() + except OSError: + pass + + def __enter__(self): + return self.start() + + def __exit__(self, *args): + self.close() + + +class _Client: + def __init__(self, connection_file, timeout): + path = Path(connection_file).expanduser().resolve(strict=True) + if path.stat().st_size > 8192: + raise ScopeError("Invalid session connection file") + if os.name == "posix": + info = path.stat() + if info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) & 0o077: + raise ScopeError("Session connection file must be private and owned by this user") + self.connection = json.loads(path.read_text()) + value = self.connection + if (not isinstance(value, dict) or value.get("protocol") != PROTOCOL + or value.get("host") != "127.0.0.1" + or type(value.get("port")) is not int or not 0 < value["port"] < 65536 + or not isinstance(value.get("token"), str) or len(value["token"]) != 64 + or not isinstance(value.get("session_id"), str) + or type(value.get("pid")) is not int or value["pid"] <= 0): + raise ScopeError("Invalid local session connection settings") + self.timeout = timeout + + def call(self, operation, **arguments): + value = self.connection + deadline = time.monotonic() + self.timeout + try: + with socket.create_connection((value["host"], value["port"]), timeout=self.timeout) as sock: + _send(sock, {"protocol": PROTOCOL, "session_id": value["session_id"], + "token": value["token"], "operation": operation, **arguments}, MAX_REQUEST) + reply = _receive(sock, MAX_RESPONSE, deadline) + except TimeoutError as error: + raise ScopeError("Attached query timed out; it may still be running in the owner. " + "Do not edit until the shared lock is available.") from error + except (OSError, ValueError) as error: + raise ScopeError("Attached session is unavailable; no local fallback was used") from error + if "error" in reply: + raise ScopeError(reply["error"], reply.get("suggestions")) + if reply.get("session_id") != value["session_id"] or reply.get("pid") != value["pid"]: + raise ScopeError("Attached session response identity mismatch") + return reply + + +class SessionBinding: + """One active borrowed design for one MCP server, separate from local state.""" + + def __init__(self): + self._operation = threading.Lock() + self._client = None + self._design = None + + def _acquire(self): + if not self._operation.acquire(blocking=False): + raise ScopeError("Another session operation is running; retry when idle") + + def attach(self, connection_file, design=None, timeout_seconds=30): + self._acquire() + try: + if type(timeout_seconds) not in (int, float) or not 0 < timeout_seconds <= 300: + raise ScopeError("Session timeout must be positive and at most 300 seconds") + try: + client = _Client(connection_file, timeout_seconds) + except (OSError, ValueError) as error: + raise ScopeError("Cannot read a valid private session connection file") from error + info = client.call("inspect") + if info.get("design_addressing") != "native-id-v1": + raise ScopeError("The owner does not support native design references") + selected = self._select(client, design) if design is not None else None + self._client = client + self._design = dict(selected) if selected is not None else None + return {**info, "attached": True, "selected_design": selected} + finally: + self._operation.release() + + @staticmethod + def _reference(client, design): + try: + reference = DesignReference.model_validate(design) + except ValueError as error: + raise ScopeError("Provide a native design reference with session, DB, library and design IDs") from error + if reference.session_id != client.connection["session_id"]: + raise ScopeError("Design reference belongs to a different session") + return reference.model_dump() + + @classmethod + def _select(cls, client, design): + reference = cls._reference(client, design) + reply = client.call("select", design=reference) + if reply.get("design") != reference: + raise ScopeError("Attached response identifies a different native design") + return reference + + def select(self, design): + self._acquire() + try: + if self._client is None: + raise ScopeError("No attached session") + selected = self._select(self._client, design) + self._design = dict(selected) + return {"session_id": self._client.connection["session_id"], + "pid": self._client.connection["pid"], + "attached": True, "selected_design": selected} + finally: + self._operation.release() + + def inspect(self): + self._acquire() + try: + if self._client is None: + return {"attached": False} + return {**self._client.call("inspect"), "attached": True, + "selected_design": dict(self._design) if self._design is not None else None} + finally: + self._operation.release() + + def detach(self): + self._acquire() + try: + self._client = self._design = None + return {"attached": False, "caller_designs_preserved": True} + finally: + self._operation.release() + + def invoke(self, tool, arguments, local, design=None): + self._acquire() + try: + if self._client is None: + if design is not None: + raise ScopeError("A native design reference requires an attached session") + with NATIVE_LOCK: + return local() + if tool not in READ_ONLY_TOOLS: + raise ScopeError("Attached session is read-only; detach before local loading, " + "reset, export or Python execution") + selected = design if design is not None else self._design + if selected is None: + raise ScopeError("Supply a native design reference or call set_session_design first") + reference = self._reference(self._client, selected) + reply = self._client.call("query", tool=tool, design=reference, arguments=arguments) + if reply.get("design") != reference or not isinstance(reply.get("result"), dict): + raise ScopeError("Attached query response does not match the requested native design") + result = dict(reply["result"]) + result["binding"] = {"session_id": reply["session_id"], "pid": reply["pid"], + "design": reference} + return result + finally: + self._operation.release() + + +BINDING = SessionBinding() diff --git a/src/naja_scope/design_reference.py b/src/naja_scope/design_reference.py new file mode 100644 index 0000000..f82f728 --- /dev/null +++ b/src/naja_scope/design_reference.py @@ -0,0 +1,27 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Native Naja design coordinates, scoped to the bridge that resolves them.""" + +from pydantic import BaseModel, ConfigDict, Field, StrictInt, StrictStr + + +class DesignReference(BaseModel): + """Same native-ID shape as Kepler MCP; no Kepler dependency or alias map.""" + + model_config = ConfigDict(extra="forbid") + session_id: StrictStr = Field(min_length=1, description="Owning Naja-Scope bridge session ID.") + db_id: StrictInt = Field(ge=0, le=255, description="Native Naja database ID (8-bit).") + library_id: StrictInt = Field(ge=0, le=65535, description="Native Naja library ID (16-bit).") + design_id: StrictInt = Field(ge=0, le=4294967295, description="Native Naja design ID (32-bit).") + + def native_key(self): + return self.db_id, self.library_id, self.design_id + + +def reference_from_design(session_id, design): + from najaeda import naja + + if not isinstance(design, naja.SNLDesign): + raise TypeError("Expected a raw naja.SNLDesign") + identity = design.getNLID() + return DesignReference(session_id=session_id, db_id=identity.getDBID(), + library_id=identity.getLibraryID(), design_id=identity.getDesignID()) diff --git a/src/naja_scope/server.py b/src/naja_scope/server.py index 9b99aa3..80c1ad2 100644 --- a/src/naja_scope/server.py +++ b/src/naja_scope/server.py @@ -8,6 +8,8 @@ from __future__ import annotations import functools +import asyncio +import inspect import os import sys from typing import Annotated, Any, Callable, Dict, List, Literal, Optional @@ -23,6 +25,11 @@ from . import api from .errors import ScopeError +from .binding import BINDING +from .design_reference import DesignReference + +DesignSelector = Annotated[Optional[DesignReference], Field( + description="Attached-session native design reference; overrides set_session_design for this query only. Omit in standalone mode.")] mcp = FastMCP("naja-scope") @@ -68,14 +75,19 @@ ) -def _tool(*, annotations: ToolAnnotations) -> Callable: +def _tool(*, annotations: ToolAnnotations, binding_control: bool = False) -> Callable: """Register fn as an MCP tool; ScopeErrors become structured responses.""" def decorator(fn: Callable) -> Callable: @functools.wraps(fn) - def wrapper(*args, **kwargs) -> Dict[str, Any]: + async def wrapper(*args, **kwargs) -> Dict[str, Any]: try: - return fn(*args, **kwargs) + if binding_control: + return await asyncio.to_thread(fn, *args, **kwargs) + arguments = dict(inspect.signature(fn).bind(*args, **kwargs).arguments) + design = arguments.pop("design", None) + return await asyncio.to_thread(BINDING.invoke, fn.__name__, arguments, + lambda: fn(**arguments), design) except ScopeError as e: return e.to_dict() @@ -84,8 +96,43 @@ def wrapper(*args, **kwargs) -> Dict[str, Any]: return decorator +@_tool(annotations=SESSION_MUTATION, binding_control=True) +def attach_session( + connection_file: Annotated[str, Field(description="Private connection JSON created by SessionBridge in the design-owning Python process.")], + design: Annotated[Optional[DesignReference], Field(description="Native design reference to select; omit to discover designs before choosing one.")] = None, + timeout_seconds: Annotated[float, Field(description="Query timeout in seconds, greater than zero and at most 300.", gt=0, le=300)] = 30, +) -> dict: + """Bind existing read-only MCP tools to a caller's live design without + loading, copying or dumping it. Mutating tools are blocked while attached. + The connection credential must remain private. Timeout does not kill the owner.""" + return BINDING.attach(connection_file, design, timeout_seconds) + + +@_tool(annotations=SESSION_MUTATION, binding_control=True) +def set_session_design( + design: Annotated[DesignReference, Field(description="Native session, database, library and design IDs from the attached owner.")], +) -> dict: + """Select a native live design for subsequent queries, without + changing the owner's selected top outside those queries.""" + return BINDING.select(design) + + +@_tool(annotations=READ_ONLY, binding_control=True) +def get_session_binding() -> dict: + """Report the attached owner, available designs and selected design. + Returns attached=false in the unchanged standalone/local mode.""" + return BINDING.inspect() + + +@_tool(annotations=SESSION_MUTATION, binding_control=True) +def detach_session() -> dict: + """Stop using the live attachment and resume local tools. Never destroys, + resets or closes the caller's designs or bridge.""" + return BINDING.detach() + + @_tool(annotations=READ_ONLY) -def status() -> dict: +def status(*, design: DesignSelector = None) -> dict: """Inspect the current in-memory session without changing it. Use this before design queries to confirm a design is loaded and whether get_intent is live (`intent_loaded`) or can be reloaded (`intent_loadable`). Returns @@ -255,6 +302,7 @@ def resolve( Optional[int], Field(description="Maximum matches to return; defaults to 20 and is capped at 200."), ] = None, + *, design: DesignSelector = None, ) -> dict: """Resolve a known hierarchical object path to instance, term, or net descriptors with source references. The final segment accepts a glob and @@ -283,6 +331,7 @@ def find( Field(description="Opaque next_cursor from the previous response; " "omit for the first page."), ] = None, + *, design: DesignSelector = None, ) -> dict: """Search case-sensitive object names across the loaded design with a glob. Use this when an exact path is unknown; use resolve once a path is known or @@ -310,6 +359,7 @@ def get_hierarchy( Optional[str], Field(description="Opaque root-level next_cursor from a previous response."), ] = None, + *, design: DesignSelector = None, ) -> dict: """Browse the instance tree below `path` (or the top instance). Use this for structural children; use find for design-wide name search or get_stats @@ -331,6 +381,7 @@ def get_drivers( Optional[int], Field(description="Maximum endpoint entries; defaults to 50 and is capped at 200."), ] = None, + *, design: DesignSelector = None, ) -> dict: """List the immediate upstream endpoints that drive a term or net across hierarchy. Use this for direct sources; use get_loads for downstream readers @@ -354,6 +405,7 @@ def get_loads( Optional[int], Field(description="Maximum endpoint entries; defaults to 50 and is capped at 200."), ] = None, + *, design: DesignSelector = None, ) -> dict: """List the immediate downstream endpoints that consume a term or net across hierarchy. Use this for direct readers; use get_drivers for upstream @@ -380,6 +432,7 @@ def trace_cone( int, Field(description="Maximum listed endpoints per frontier kind; clamped to 1..200."), ] = 50, + *, design: DesignSelector = None, ) -> dict: """Trace the combinational fanin/fanout cone of a term/net via naja's LogicCone. Use this for transitive logic reachability; use get_drivers or @@ -405,6 +458,7 @@ def get_source( int, Field(description="Extra lines before and after the source range; clamped to 0..20."), ] = 3, + *, design: DesignSelector = None, ) -> dict: """Read the bounded SystemVerilog source excerpt that produced an object (for example, an FF instance maps to its `always_ff` block). Use after @@ -421,6 +475,7 @@ def get_module_card( str, Field(description="Elaborated module/model name, not an instance path."), ], + *, design: DesignSelector = None, ) -> dict: """Deterministic module summary: ports, instance counts by model, sequential count, source ref, plus clock/reset candidates — a name-based @@ -444,6 +499,7 @@ def get_stats( Field(description="Opaque next_cursor from the previous response; " "omit for the first page."), ] = None, + *, design: DesignSelector = None, ) -> dict: """Summarize instance population by model below `path` or the top design. Use this for aggregate leaf/sequential/model counts; use get_hierarchy for @@ -464,6 +520,7 @@ def get_intent( Literal["auto", "type", "fsm_states", "parameters"], Field(description="Intent fact to retrieve; auto selects from the reference."), ] = "auto", + *, design: DesignSelector = None, ) -> dict: """Retrieve source-level intent that netlist lowering erases (warm-only). Use when the answer is in the SystemVerilog *type/declaration*, not the diff --git a/src/naja_scope/snl.py b/src/naja_scope/snl.py index ff100b1..4b51b2a 100644 --- a/src/naja_scope/snl.py +++ b/src/naja_scope/snl.py @@ -24,6 +24,8 @@ import difflib import re +from contextlib import contextmanager +from contextvars import ContextVar from dataclasses import dataclass from typing import Iterator, List, Optional, Tuple @@ -50,21 +52,39 @@ # -- universe / designs ------------------------------------------------------ +_inspection_top = ContextVar("naja_scope_inspection_top", default=None) + + +@contextmanager +def inspection_design(design): + """Scope queries to an explicit root without changing native DB top state.""" + token = _inspection_top.set(design) + try: + yield + finally: + _inspection_top.reset(token) + + def universe(): return naja.NLUniverse.get() def has_top() -> bool: - u = universe() - return u is not None and u.getTopDesign() is not None + return top_design() is not None def top_design(): + selected = _inspection_top.get() + if selected is not None: + return selected u = universe() return u.getTopDesign() if u is not None else None def top_db(): + selected = _inspection_top.get() + if selected is not None: + return selected.getDB() u = universe() return u.getTopDB() if u is not None else None @@ -86,10 +106,17 @@ def design_names(include_primitives: bool = False) -> List[str]: def find_design(name: str): - for design in iter_designs(include_primitives=True): - if design.getName() == name: - return design - return None + # An explicitly selected native top wins over same-named models in other + # libraries. Never silently choose among remaining ambiguous models. + top = top_design() + if top is not None and top.getName() == name: + return top + matches = [design for design in iter_designs(include_primitives=True) + if design.getName() == name] + if len(matches) > 1: + from .errors import ScopeError + raise ScopeError(f"Ambiguous module name '{name}'; select its native design reference first.") + return matches[0] if matches else None def suggest_designs(name: str, n: int = 8) -> List[str]: diff --git a/tests/test_binding.py b/tests/test_binding.py new file mode 100644 index 0000000..2efe543 --- /dev/null +++ b/tests/test_binding.py @@ -0,0 +1,457 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Real in-memory designs and MCP transport, without a Kepler/22b dependency.""" + +import asyncio +import json +import os +from pathlib import Path +import socket +import struct +import sys +import threading +import time +from types import SimpleNamespace + +import pytest +from najaeda import naja +from pydantic import ValidationError + +from naja_scope import api, server +from naja_scope.binding import (MAX_REQUEST, READ_ONLY_TOOLS, + SessionBinding, SessionBridge, _Client, _receive) +from naja_scope.design_reference import DesignReference +from naja_scope.errors import ScopeError +from naja_scope.session import SESSION + + +def make_design(universe, driver="a"): + database = naja.NLDB.create(universe) + library = naja.NLLibrary.create(database, "designs") + design = naja.SNLDesign.create(library, "top") + for name in ("a", "b"): + net = naja.SNLScalarNet.create(design, name) + naja.SNLScalarTerm.create(design, naja.SNLTerm.Direction.Input, name).setNet(net) + naja.SNLScalarTerm.create(design, naja.SNLTerm.Direction.Output, "y").setNet( + design.getNet(driver)) + return design + + +@pytest.fixture +def live(): + SESSION.reset() + universe = naja.NLUniverse.create() + golden, edited = make_design(universe), make_design(universe, "b") + universe.setTopDesign(edited) + universe.setTopDesign(golden) + lock = threading.RLock() + bridge = SessionBridge(lock=lock).start() + binding = SessionBinding() + references = [bridge.design_reference(design) for design in (golden, edited)] + yield SimpleNamespace(universe=universe, golden=golden, edited=edited, + bridge=bridge, binding=binding, refs=references, lock=lock) + binding.detach() + bridge.close() + SESSION.reset() + + +def query(live, tool="get_drivers", *, design=None, **arguments): + return live.binding.invoke(tool, arguments, lambda: pytest.fail("No local fallback"), design) + + +def test_switch_same_named_designs_by_native_database_id(live): + first, second = live.refs + assert first["db_id"] != second["db_id"] + assert first["library_id"] == second["library_id"] + assert first["design_id"] == second["design_id"] + info = live.binding.attach(live.bridge.connection_file, first) + assert info["design_addressing"] == "native-id-v1" + assert [d["name"] for d in info["designs"]] == ["top", "top"] + assert not hasattr(live.bridge, "_designs") + for reference, expected in ((first, "top.a"), (second, "top.b"), (first, "top.a")): + live.binding.select(reference) + result = query(live, path="top.y") + assert result["top_drivers"] == [{"port": expected, "dir": "input"}] + assert result["binding"]["design"] == reference + assert live.universe.getTopDesign() == live.golden + assert live.universe.getTopDB() == live.golden.getDB() + assert api.SESSION is SESSION + + +def test_query_override_does_not_change_default_and_sees_cumulative_edits(live): + live.binding.attach(live.bridge.connection_file, live.refs[0]) + result = query(live, design=live.refs[1], path="top.y") + assert result["top_drivers"][0]["port"] == "top.b" + assert live.binding.inspect()["selected_design"] == live.refs[0] + with live.lock: + live.edited.getScalarTerm("y").setNet(live.edited.getNet("a")) + assert query(live, design=live.refs[1], path="top.y")["top_drivers"][0]["port"] == "top.a" + with live.lock: + live.edited.getScalarTerm("y").setNet(live.edited.getNet("b")) + assert query(live, design=live.refs[1], path="top.y")["top_drivers"][0]["port"] == "top.b" + assert query(live, path="top.y")["binding"]["design"] == live.refs[0] + + +@pytest.mark.parametrize("tool,arguments", [ + ("status", {}), ("resolve", {"path": "top.y"}), + ("find", {"pattern": "*"}), ("get_hierarchy", {}), + ("get_drivers", {"path": "top.y"}), ("get_loads", {"path": "top.b"}), + ("trace_cone", {"path": "top.y", "direction": "fanin"}), + ("get_stats", {}), ("get_module_card", {"module": "top"}), + ("get_intent", {"ref": "top.y"}), +]) +def test_all_typed_queries_match_direct_api_on_selected_design(live, tool, arguments): + live.binding.attach(live.bridge.connection_file) + with live.lock: + naja.SNLScalarTerm.create(live.edited, naja.SNLTerm.Direction.Input, "only_edited") + live.universe.setTopDesign(live.edited) + expected = getattr(api, tool)(**arguments) + live.universe.setTopDesign(live.golden) + result = query(live, tool, design=live.refs[1], **arguments) + result.pop("binding") + assert result == expected + assert live.universe.getTopDesign() == live.golden + + +def test_returned_selection_cannot_mutate_retained_default(live): + result = live.binding.attach(live.bridge.connection_file, live.refs[0]) + result["selected_design"]["db_id"] = live.refs[1]["db_id"] + result = live.binding.select(live.refs[0]) + result["selected_design"]["db_id"] = live.refs[1]["db_id"] + result = live.binding.inspect() + result["selected_design"]["db_id"] = live.refs[1]["db_id"] + assert query(live, path="top.y")["binding"]["design"] == live.refs[0] + + +def test_discover_then_require_explicit_selection(live): + info = live.binding.attach(live.bridge.connection_file) + assert info["selected_design"] is None + with pytest.raises(ScopeError, match="Supply a native"): + query(live, path="top.y") + assert query(live, design=live.refs[1], path="top.y")["top_drivers"][0]["port"] == "top.b" + + +def test_queries_preserve_all_native_top_selections_including_unset(live): + live.binding.attach(live.bridge.connection_file) + with live.lock: + unselected = make_design(live.universe) + reference = live.bridge.design_reference(unselected) + databases = list(live.universe.getUserDBs()) + before = [db.getTopDesign() for db in databases] + assert unselected.getDB().getTopDesign() is None + for tool, arguments in (("status", {}), ("get_drivers", {"path": "top.y"})): + query(live, tool, design=reference, **arguments) + assert [db.getTopDesign() for db in databases] == before + assert live.universe.getTopDesign() == live.golden + with pytest.raises(ScopeError, match="target-database tops"): + query(live, "trace_cone", design=reference, path="top.y", direction="fanin") + assert [db.getTopDesign() for db in databases] == before + + +def test_cone_restores_target_database_top_even_when_querying_another_model(live): + with live.lock: + extra = naja.SNLDesign.create(live.edited.getLibrary(), "extra") + n = naja.SNLScalarNet.create(extra, "i") + naja.SNLScalarTerm.create(extra, naja.SNLTerm.Direction.Input, "i").setNet(n) + naja.SNLScalarTerm.create(extra, naja.SNLTerm.Direction.Output, "o").setNet(n) + reference = live.bridge.design_reference(extra) + live.binding.attach(live.bridge.connection_file) + query(live, "trace_cone", design=reference, path="extra.o", direction="fanin") + assert live.universe.getTopDesign() == live.golden + assert live.edited.getDB().getTopDesign() == live.edited + + +def test_bridge_does_not_require_an_owner_top(): + SESSION.reset() + universe = naja.NLUniverse.create() + design = make_design(universe) + binding = SessionBinding() + try: + with SessionBridge() as bridge: + binding.attach(bridge.connection_file) + result = binding.invoke("get_drivers", {"path": "top.y"}, + lambda: pytest.fail("No local fallback"), bridge.design_reference(design)) + assert result["top_drivers"][0]["port"] == "top.a" + assert universe.getTopDesign() is None + assert design.getDB().getTopDesign() is None + with pytest.raises(ScopeError, match="global and target-database tops"): + binding.invoke("trace_cone", {"path": "top.y", "direction": "fanin"}, + lambda: pytest.fail("No local fallback"), bridge.design_reference(design)) + binding.detach() + finally: + SESSION.reset() + + +def test_new_designs_are_visible_without_registration_and_renames_keep_ids(live): + live.binding.attach(live.bridge.connection_file) + with live.lock: + another = make_design(live.universe) + reference = live.bridge.design_reference(another) + assert reference in [d["reference"] for d in live.binding.inspect()["designs"]] + with live.lock: + another.setName("renamed") + assert live.bridge.design_reference(another) == reference + result = query(live, design=reference, path="renamed.y") + assert result["top_drivers"][0]["port"] == "renamed.a" + + +def test_library_and_design_ids_select_exact_model_even_with_duplicate_names(live): + with live.lock: + library = naja.NLLibrary.create(live.golden.getDB(), "another_library") + other = naja.SNLDesign.create(library, "top") + naja.SNLScalarTerm.create(other, naja.SNLTerm.Direction.Input, "only_here") + third = naja.SNLDesign.create(library, "third") + second_ref, third_ref = [live.bridge.design_reference(d) for d in (other, third)] + assert second_ref["db_id"] == live.refs[0]["db_id"] + assert second_ref["library_id"] != live.refs[0]["library_id"] + assert second_ref["design_id"] == live.refs[0]["design_id"] + assert third_ref["library_id"] == second_ref["library_id"] + assert third_ref["design_id"] != second_ref["design_id"] + live.binding.attach(live.bridge.connection_file) + for reference, model in ((live.refs[0], live.golden), (second_ref, other), (third_ref, third)): + with live.lock: + live.universe.setTopDesign(model) + expected = api.get_module_card(model.getName()) + live.universe.setTopDesign(live.golden) + result = query(live, "get_module_card", design=reference, module=model.getName()) + result.pop("binding") + assert result == expected + with pytest.raises(ScopeError, match="Ambiguous"): + query(live, "get_module_card", design=third_ref, module="top") + assert live.universe.getTopDesign() == live.golden + + +@pytest.mark.parametrize("field,value", [ + ("db_id", -1), ("db_id", 256), ("library_id", 65536), ("design_id", 4294967296), + ("db_id", True), ("db_id", "1"), ("library_id", 0.0), ("session_id", ""), + ("session_id", 123), ("name", "golden"), +]) +def test_reference_rejects_coercion_truncation_and_aliases(live, field, value): + bad = dict(live.refs[0], **{field: value}) + with pytest.raises(ValidationError): + DesignReference.model_validate(bad) + live.binding.attach(live.bridge.connection_file) + with pytest.raises(ScopeError): + query(live, design=bad, path="top.y") + # The owner validates too, not just the public MCP/client schema. + client = _Client(live.bridge.connection_file, 2) + with pytest.raises(ScopeError): + client.call("query", tool="get_drivers", design=bad, arguments={"path": "top.y"}) + + +def test_other_session_missing_and_deleted_designs_never_fall_back(live): + live.binding.attach(live.bridge.connection_file, live.refs[0]) + for bad in (dict(live.refs[1], session_id="another-session"), + dict(live.refs[1], design_id=1000)): + with pytest.raises(ScopeError): + live.binding.select(bad) + assert live.binding.inspect()["selected_design"] == live.refs[0] + with live.lock: + live.edited.destroy() + with pytest.raises(ScopeError, match="does not exist"): + query(live, design=live.refs[1], path="top.y") + + +def test_replaced_universe_is_rejected(live): + live.binding.attach(live.bridge.connection_file, live.refs[0]) + with live.lock: + live.universe.destroy() + replacement = naja.NLUniverse.create() + replacement.setTopDesign(make_design(replacement)) + with pytest.raises(ScopeError, match="no longer valid"): + query(live, path="top.y") + + +@pytest.mark.parametrize("tool", ["reset_universe", "load_verilog", "load_systemverilog", + "load_liberty", "load_primitives", "load_snapshot", + "save_snapshot", "load_intent", "query_python"]) +def test_attached_mode_blocks_mutations_on_client_and_owner(live, tool): + live.binding.attach(live.bridge.connection_file, live.refs[0]) + with pytest.raises(ScopeError, match="read-only"): + query(live, tool) + with pytest.raises(ScopeError, match="read-only"): + _Client(live.bridge.connection_file, 2).call( + "query", tool=tool, design=live.refs[0], arguments={}) + assert live.universe.getTopDesign() == live.golden + + +def test_owner_lock_blocks_inspection_and_exceptions_restore_top(live, monkeypatch): + live.binding.attach(live.bridge.connection_file, live.refs[1]) + with live.lock: + with pytest.raises(ScopeError, match="busy"): + query(live, path="top.y") + with pytest.raises(ScopeError): + query(live, path="top.missing") + assert live.universe.getTopDesign() == live.golden + assert api.SESSION is SESSION + def broken(**kwargs): + raise RuntimeError("native query error") + monkeypatch.setattr(api, "get_drivers", broken) + with pytest.raises(ScopeError, match="native query error"): + query(live, path="top.y") + assert live.universe.getTopDesign() == live.golden + assert api.SESSION is SESSION + + +def test_timeout_keeps_owner_locked_until_query_finishes(live, monkeypatch): + entered, release, finished = threading.Event(), threading.Event(), threading.Event() + original = api.get_drivers + def slow(**kwargs): + entered.set() + try: + assert release.wait(3) + return original(**kwargs) + finally: + finished.set() + monkeypatch.setattr(api, "get_drivers", slow) + live.binding.attach(live.bridge.connection_file, live.refs[1]) + live.binding._client.timeout = 0.1 + try: + with pytest.raises(ScopeError, match="timed out"): + query(live, path="top.y") + assert entered.is_set() + acquired = live.lock.acquire(blocking=False) + if acquired: + live.lock.release() + assert not acquired + finally: + release.set() + assert finished.wait(3) + with live.lock: + pass + assert live.universe.getTopDesign() == live.golden + + +def test_close_detach_and_local_mode_preserve_owner_designs(live): + live.binding.attach(live.bridge.connection_file, live.refs[1]) + assert live.binding.detach()["caller_designs_preserved"] + assert live.binding.invoke("status", {}, api.status)["top"]["name"] == "top" + with pytest.raises(ScopeError, match="requires an attached"): + query(live, design=live.refs[0], path="top.y") + live.bridge.close() + assert not live.bridge.connection_file.exists() + assert naja.NLUniverse.get() is live.universe + assert live.golden.getName() == live.edited.getName() == "top" + with pytest.raises(ScopeError): + live.bridge.start() + + +def test_authentication_protocol_and_remote_host_rejection(live, tmp_path): + client = _Client(live.bridge.connection_file, 2) + client.connection["token"] = "0" * 64 + with pytest.raises(ScopeError, match="authentication"): + client.call("inspect") + connection = json.loads(live.bridge.connection_file.read_text()) + for changes in ({"protocol": "unknown"}, {"host": "example.com"}, {"port": True}): + path = tmp_path / "invalid.json" + path.write_text(json.dumps(dict(connection, **changes))) + path.chmod(0o600) + with pytest.raises(ScopeError): + _Client(path, 2) + if os.name == "posix": + live.bridge.connection_file.chmod(0o644) + with pytest.raises(ScopeError, match="private"): + _Client(live.bridge.connection_file, 2) + + +def test_oversized_request_rejected(live): + connection = json.loads(live.bridge.connection_file.read_text()) + with socket.create_connection(("127.0.0.1", connection["port"]), timeout=2) as sock: + sock.sendall(struct.pack("!I", MAX_REQUEST + 1)) + assert "error" in _receive(sock, 4096, time.monotonic() + 2) + + +def test_wrong_response_design_rejected(live, monkeypatch): + live.binding.attach(live.bridge.connection_file, live.refs[0]) + monkeypatch.setattr(live.binding._client, "call", lambda *a, **k: { + "session_id": live.bridge.session_id, "pid": os.getpid(), + "design": live.refs[1], "result": {"loaded": True}}) + with pytest.raises(ScopeError, match="does not match"): + query(live, path="top.y") + + +def test_dead_owner_does_not_fall_back_to_local(live): + live.binding.attach(live.bridge.connection_file, live.refs[0]) + live.bridge.close() + with pytest.raises(ScopeError, match="unavailable"): + query(live, path="top.y") + assert live.binding._client is not None + + +def test_existing_warm_source_and_intent_are_reused(uart_session): + source = Path(__file__).parent / "fixtures/uart.sv" + SESSION.reset() + SESSION.load_systemverilog([str(source)], keep_ast_link=True) + expected_source = api.get_source("uart_top.u_tx.u_div_cnt") + expected_intent = api.get_intent("uart_top.u_tx", want="parameters") + universe = naja.NLUniverse.get() + top = universe.getTopDesign() + binding = SessionBinding() + with SessionBridge(source_files=[source]) as bridge: + reference = bridge.design_reference(top) + binding.attach(bridge.connection_file, reference) + for tool, args, expected in ( + ("get_source", {"path": "uart_top.u_tx.u_div_cnt"}, expected_source), + ("get_intent", {"ref": "uart_top.u_tx", "want": "parameters"}, expected_intent), + ): + result = binding.invoke(tool, args, lambda: pytest.fail("No local fallback")) + result.pop("binding") + assert result == expected + binding.detach() + assert naja.NLUniverse.get() is universe and universe.getTopDesign() == top + + +def test_discovery_is_bounded_but_does_not_limit_native_resolution(live): + with live.lock: + library = live.edited.getLibrary() + for index in range(205): + last = naja.SNLDesign.create(library, f"extra_{index}") + info = live.binding.attach(live.bridge.connection_file) + assert len(info["designs"]) == 200 and info["designs_truncated"] + reference = live.bridge.design_reference(last) + assert query(live, "status", design=reference)["top"]["name"] == "extra_204" + + +def test_mcp_schemas_offer_typed_references_on_all_queries(): + tools = {tool.name: tool for tool in asyncio.run(server.mcp.list_tools())} + for name in READ_ONLY_TOOLS | {"attach_session", "set_session_design"}: + schema = tools[name].inputSchema + assert "design" in schema["properties"] + fields = schema["$defs"]["DesignReference"] + assert set(fields["required"]) == {"session_id", "db_id", "library_id", "design_id"} + assert fields["additionalProperties"] is False + + +def test_real_stdio_mcp_inspects_both_designs_in_owner_process(live): + from mcp import ClientSession, StdioServerParameters + from mcp.client.stdio import stdio_client + + async def run(): + env = dict(os.environ, PYTHONPATH=str(Path(__file__).resolve().parents[1] / "src"), + NAJA_SCOPE_ENABLE_PYTHON="1") + params = StdioServerParameters(command=sys.executable, args=["-m", "naja_scope.server"], env=env) + async with stdio_client(params) as streams: + async with ClientSession(*streams) as client: + await client.initialize() + async def call(name, **arguments): + response = await client.call_tool(name, arguments) + assert not response.isError, response + return response.structuredContent or json.loads(response.content[0].text) + attached = await call("attach_session", connection_file=str(live.bridge.connection_file)) + assert attached["pid"] == os.getpid() + for reference, expected in ((live.refs[0], "top.a"), (live.refs[1], "top.b")): + result = await call("get_drivers", design=reference, path="top.y") + assert result["top_drivers"][0]["port"] == expected + assert result["binding"]["design"] == reference + with live.lock: + live.edited.getScalarTerm("y").setNet(live.edited.getNet("a")) + await call("set_session_design", design=live.refs[1]) + result = await call("get_drivers", path="top.y") + assert result["top_drivers"][0]["port"] == "top.a" + assert "error" in await call("query_python", code="1+1") + assert "error" in await call("reset_universe") + invalid = await client.call_tool("get_drivers", {"path": "top.y", "design": dict(live.refs[0], db_id=256)}) + assert invalid.isError + await call("detach_session") + assert (await call("status"))["loaded"] is False + asyncio.run(run()) + assert live.universe.getTopDesign() == live.golden diff --git a/tests/test_zz_snapshot.py b/tests/test_zz_snapshot.py index 1a8e2bf..4a5c0de 100644 --- a/tests/test_zz_snapshot.py +++ b/tests/test_zz_snapshot.py @@ -154,6 +154,7 @@ async def _list(): "get_drivers", "get_loads", "trace_cone", "get_source", "get_module_card", "get_stats", "get_intent", "load_intent", + "attach_session", "set_session_design", "get_session_binding", "detach_session", } assert expected <= names, expected - names # query_python is opt-in (NAJA_SCOPE_ENABLE_PYTHON) and registered at import @@ -172,7 +173,7 @@ async def _list(): read_only = { "status", "resolve", "find", "get_hierarchy", "get_drivers", "get_loads", "trace_cone", "get_source", "get_module_card", - "get_stats", "get_intent", + "get_stats", "get_intent", "get_session_binding", } for name in read_only: assert tools_by_name[name].annotations.readOnlyHint is True