From 5ce5e9c23587cf1e10897f866c1ee24424110439 Mon Sep 17 00:00:00 2001 From: espetro Date: Sun, 18 Oct 2026 19:47:17 +0000 Subject: [PATCH 1/3] fix(compat2): restore parenthesized except in cache.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ruff format under target-version py314 strips except-tuple parens into 'except A, B:', which is only valid on Python 3.14+ and breaks import on the 3.13 dev venv — pytest collection failed on tests/compat2. Restores the intended (OSError, ValueError) handler; no semantic change. --- benchmarks/compat2/cache.py | 2 +- src/pyscriptc/typefacts.py | 892 +++++++++++++++++++++++++++++++++++- 2 files changed, 890 insertions(+), 4 deletions(-) diff --git a/benchmarks/compat2/cache.py b/benchmarks/compat2/cache.py index 1a81bab..d2f9b8e 100644 --- a/benchmarks/compat2/cache.py +++ b/benchmarks/compat2/cache.py @@ -112,7 +112,7 @@ def load(cache_dir: Path, key: str) -> dict[str, Any] | None: path = cache_dir / f"{key}.json" try: data = json.loads(path.read_text(encoding="utf-8")) - except OSError, ValueError: + except (OSError, ValueError): return None # missing, or truncated by an interrupted run: recompute return data if isinstance(data, dict) else None diff --git a/src/pyscriptc/typefacts.py b/src/pyscriptc/typefacts.py index d8aae8b..03eb4ed 100644 --- a/src/pyscriptc/typefacts.py +++ b/src/pyscriptc/typefacts.py @@ -6,14 +6,24 @@ `FactsProvider` reports what the checker decided — optional/union members, callable signatures, overload choices, generic arguments — and tier-2 lowering queries it. `MypyFactsAdapter` is backed by the same mypy pass -`check.py` runs (same subset plugin); a future `PyrightFactsAdapter` feeds -the external JSON oracle into the same dataclasses. +`check.py` runs (same subset plugin); `PyrightFactsAdapter` feeds the +external pyright-langserver oracle (stdio LSP hover + documentSymbol) into +the same dataclasses. """ from __future__ import annotations +import ast +import json +import os +import queue +import re +import shlex +import subprocess import tempfile -from collections.abc import Mapping +import threading +import time +from collections.abc import Mapping, Sequence from dataclasses import dataclass, field from pathlib import Path from typing import Any, Protocol, runtime_checkable @@ -171,6 +181,8 @@ def facts_provider(checker: str) -> FactsProvider: """Pick the provider behind a `run_checker` checker name.""" if checker == "mypy": return MypyFactsAdapter() + if checker == "pyright": + return PyrightFactsAdapter() raise CheckFailed(f"checker {checker!r} is not wired; the spike preflight uses mypy") @@ -421,3 +433,877 @@ def facts(self, path: Path) -> ModuleFacts: callables=callables, overloads=overloads, ) + + +# --------------------------------------------------------------------------- +# Pyright langserver oracle +# +# The second provider speaks LSP: a long-lived pyright-langserver subprocess +# over stdio (initialize -> didOpen -> hover/documentSymbol). There is no +# `--facts` verb and no vendored analyzer; rendered hover strings like +# "(function) def f(a: int) -> str" are parsed back into the same +# TypeFacts/CallableFacts dataclasses. Hover answers on symbol positions +# only, so expression coverage is sparser than mypy's full TypeMap — nodes +# pyright declines to describe degrade to "no fact". The langserver binary +# comes from PYRIGHT_LANGSERVER_BIN (shlex-split, default +# "pyright-langserver" on PATH); "--stdio" is appended when missing. + + +_PYRIGHT_STUBS = Path(__file__).with_name("_pyright_stubs") + + +class _LspConnection: + """Minimal synchronous JSON-RPC client for one stdio langserver. + + Requests are sent on the main thread; a daemon reader thread frames + `Content-Length` messages off stdout and routes responses by id. This + keeps the protocol plumbing out of the fact-mapping code. + """ + + def __init__(self, argv: Sequence[str]) -> None: + self._argv = list(argv) + self._proc: subprocess.Popen[bytes] | None = None + self._lock = threading.Lock() + self._pending: dict[int, queue.Queue[dict[str, Any]]] = {} + self._next_id = 0 + self._notifications: list[dict[str, Any]] = [] + self._notify_cv = threading.Condition() + self._dead = False + + def start(self) -> None: + if self._proc is not None: + return + try: + self._proc = subprocess.Popen( + self._argv, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + ) + except FileNotFoundError as exc: + raise CheckFailed( + f"pyright langserver is not installed (command {self._argv[0]!r}); " + "set PYRIGHT_LANGSERVER_BIN or put pyright-langserver on PATH" + ) from exc + threading.Thread(target=self._read_loop, daemon=True).start() + + def request(self, method: str, params: Any, timeout: float = 30.0) -> dict[str, Any] | None: + """Send a request and return its `result`, or None on LSP error.""" + self._ensure_live() + with self._lock: + self._next_id += 1 + request_id = self._next_id + slot: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=1) + self._pending[request_id] = slot + try: + self._send({"jsonrpc": "2.0", "id": request_id, "method": method, "params": params}) + try: + response = slot.get(timeout=timeout) + except queue.Empty: + raise CheckFailed(f"pyright langserver timed out answering {method!r}") from None + finally: + with self._lock: + self._pending.pop(request_id, None) + error = response.get("error") + if error: + return None + return response.get("result") + + def notify(self, method: str, params: Any) -> None: + if self._dead: + return + try: + self._send({"jsonrpc": "2.0", "method": method, "params": params}) + except (OSError, CheckFailed): + self._dead = True + + def wait_notification( + self, + method: str, + match: Any, + timeout: float, + ) -> dict[str, Any] | None: + """Wait for a notification of `method` whose params satisfy `match`.""" + deadline = time.monotonic() + timeout + with self._notify_cv: + while True: + for index, msg in enumerate(self._notifications): + if msg.get("method") == method and match(msg.get("params")): + return self._notifications.pop(index) + if self._dead: + return None + remaining = deadline - time.monotonic() + if remaining <= 0: + return None + self._notify_cv.wait(remaining) + + def shutdown(self) -> None: + proc = self._proc + if proc is None: + return + self._proc = None + try: + self.request("shutdown", None, timeout=10) + self.notify("exit", None) + proc.wait(timeout=5) + except (CheckFailed, subprocess.TimeoutExpired, OSError): + pass + if proc.poll() is None: + proc.kill() + self._dead = True + + def _ensure_live(self) -> None: + self.start() + assert self._proc is not None + if self._dead or self._proc.poll() is not None: + raise CheckFailed("pyright langserver exited unexpectedly") + + def _send(self, message: dict[str, Any]) -> None: + assert self._proc is not None and self._proc.stdin is not None + body = json.dumps(message).encode("utf-8") + try: + self._proc.stdin.write(f"Content-Length: {len(body)}\r\n\r\n".encode() + body) + self._proc.stdin.flush() + except (BrokenPipeError, OSError) as exc: + self._dead = True + raise CheckFailed(f"pyright langserver pipe broke: {exc}") from exc + + def _read_loop(self) -> None: + assert self._proc is not None and self._proc.stdout is not None + stream = self._proc.stdout + while True: + try: + length = -1 + while True: + line = stream.readline() + if not line: + self._mark_dead() + return + if line in (b"\r\n", b"\n"): + break + head, _, value = line.partition(b":") + if head.strip().lower() == b"content-length": + try: + length = int(value.strip()) + except ValueError: + pass + if length < 0: + continue + body = stream.read(length) + if len(body) < length: + self._mark_dead() + return + message = json.loads(body) + except (OSError, ValueError, json.JSONDecodeError): + self._mark_dead() + return + self._dispatch(message) + + def _dispatch(self, message: dict[str, Any]) -> None: + request_id = message.get("id") + if request_id is not None and ("result" in message or "error" in message): + with self._lock: + slot = self._pending.pop(request_id, None) + if slot is not None: + slot.put(message) + return + with self._notify_cv: + self._notifications.append(message) + self._notify_cv.notify_all() + + def _mark_dead(self) -> None: + self._dead = True + with self._lock: + pending = list(self._pending.values()) + self._pending.clear() + for slot in pending: + slot.put({"error": "server exited"}) + with self._notify_cv: + self._notify_cv.notify_all() + + +def _utf16_col(line_text: str, byte_col: int) -> int: + """ast col_offset is a UTF-8 byte offset; LSP wants UTF-16 units.""" + prefix = line_text.encode("utf-8")[:byte_col].decode("utf-8", errors="ignore") + return len(prefix.encode("utf-16-le")) // 2 + + +def _split_top(text: str, sep: str) -> list[str]: + """Split `text` on `sep` occurrences outside brackets and quotes.""" + parts: list[str] = [] + depth = 0 + quote = "" + start = 0 + index = 0 + while index < len(text): + char = text[index] + if quote: + if char == "\\": + index += 1 + elif char == quote: + quote = "" + elif char in "\"'": + quote = char + elif char in "([{": + depth += 1 + elif char in ")]}": + depth -= 1 + elif char == sep and depth == 0: + parts.append(text[start:index]) + start = index + 1 + index += 1 + parts.append(text[start:]) + return parts + + +def _strip_wrapping_parens(text: str) -> str: + """Drop one layer of parens when it wraps the whole span `(A | B)`.""" + text = text.strip() + while text.startswith("(") and text.endswith(")"): + inner = text[1:-1] + depth = 0 + quote = "" + balanced = True + for char in inner: + if quote: + if char == "\\": + continue + if char == quote: + quote = "" + elif char in "\"'": + quote = char + elif char in "({[": + depth += 1 + elif char in ")}]": + depth -= 1 + if depth < 0: + balanced = False + break + if not balanced or depth != 0: + break + text = inner.strip() + return text + + +def _short_name(name: str) -> str: + """`module.Class` -> `Class`, keeping subscripts and whitespace out.""" + name = name.strip() + return name.rsplit(".", 1)[-1] + + +def _parse_type_text(text: str) -> TypeFacts | None: + """Parse one rendered type string into TypeFacts; None when unintelligible. + + Shared grammar for hover payloads (`list[int] | None`, `def` return + types) and for annotation source slices. The shapes it knows: unions + split on top-level `|`, `Optional`/`Union`, `Literal`, `type[X]`, + `Callable`, `tuple`, generic `Head[args]`, `None`, `Any`/`Unknown`, + `name@scope` typevars, quoted literals, and bare class names. + """ + text = _strip_wrapping_parens(text.strip()) + if not text or text in {"...", "Ellipsis"}: + return None + members = [piece.strip() for piece in _split_top(text, "|")] + if len(members) > 1: + args = tuple(member for member in (_parse_type_text(piece) for piece in members) if member) + if not args: + return None + return TypeFacts( + kind="union", + optional=any(member.kind == "none" or member.optional for member in args), + args=args, + runtime_repr=text, + ) + head, bracket = _split_head_bracket(text) + if bracket is not None: + kind = _short_name(head) + inner = bracket + if kind in {"Optional"}: + inner_facts = _parse_type_text(inner) + if inner_facts is None: + return None + return TypeFacts( + kind="union", + optional=True, + args=(inner_facts, TypeFacts(kind="none", optional=True, runtime_repr="None")), + runtime_repr=text, + ) + if kind == "Union": + args = tuple(m for m in (_parse_type_text(p) for p in _split_top(inner, ",")) if m) + return ( + TypeFacts( + kind="union", + optional=any(m.kind == "none" or m.optional for m in args), + args=args, + runtime_repr=text, + ) + if args + else None + ) + if kind in {"type", "Type"}: + operand = _parse_type_text(inner) + return TypeFacts(kind="type", args=(operand,), runtime_repr=text) if operand else None + if kind == "Literal": + literals = tuple( + fact for fact in (_parse_literal_piece(piece.strip()) for piece in _split_top(inner, ",")) if fact + ) + if not literals: + return None + if len(literals) == 1: + return TypeFacts(kind="literal", args=literals[0].args, runtime_repr=text) + return TypeFacts( + kind="union", + optional=any(lit.optional for lit in literals), + args=literals, + runtime_repr=text, + ) + if kind == "Callable": + return _parse_callable_type(inner, text) + pieces = _split_top(inner, ",") + if pieces and pieces[-1].strip() == "...": + pieces = pieces[:-1] # homogeneous `tuple[X, ...]` + args = tuple(m for m in (_parse_type_text(p) for p in pieces) if m) + return TypeFacts(kind=kind, args=args, runtime_repr=text) + if text.startswith(("*", "**")): + return _parse_type_text(text.lstrip("*")) + if "@" in text and not text.startswith(('"', "'")): + name = text.split("@", 1)[0].strip() + return TypeFacts(kind="typevar", runtime_repr=name) if name else None + lowered = text.lower() + if lowered in {"none", "nonetype"}: + return TypeFacts(kind="none", optional=True, runtime_repr="None") + if text in {"Any", "Unknown"}: + return TypeFacts(kind="any", runtime_repr=text) + if text in {"Never", "NoReturn"}: + return TypeFacts(kind="unknown", runtime_repr=text) + literal = _parse_literal_piece(text) + if literal is not None: + return literal + if text.startswith(("'", '"')) and text.endswith(("'", '"')): + return _parse_type_text(text[1:-1]) + if not all(part.isidentifier() for part in text.split(".")): + return None + return TypeFacts(kind=_short_name(text), runtime_repr=text) + + +def _split_head_bracket(text: str) -> tuple[str, str | None]: + """`list[int]` -> ("list", "int"); returns (text, None) when bracketless.""" + depth = 0 + quote = "" + for index, char in enumerate(text): + if quote: + if char == "\\": + continue + if char == quote: + quote = "" + elif char in "\"'": + quote = char + elif char == "[": + if depth == 0: + if not text.endswith("]"): + return text, None + return text[:index], text[index + 1 : -1] + depth += 1 + elif char == "]": + depth -= 1 + elif char in "({": + depth += 1 + elif char in ")}": + depth -= 1 + return text, None + + +def _parse_literal_piece(text: str) -> TypeFacts | None: + """A `Literal[...]` member or a bare literal: 1, 's', True, None, Color.RED.""" + try: + value = ast.literal_eval(text) + except (ValueError, SyntaxError): + value = None + if isinstance(value, bool): + return TypeFacts(kind="literal", args=(TypeFacts(kind="bool"),), runtime_repr=text) + if isinstance(value, int): + return TypeFacts(kind="literal", args=(TypeFacts(kind="int"),), runtime_repr=text) + if isinstance(value, float): + return TypeFacts(kind="literal", args=(TypeFacts(kind="float"),), runtime_repr=text) + if isinstance(value, str): + return TypeFacts(kind="literal", args=(TypeFacts(kind="str"),), runtime_repr=text) + if isinstance(value, bytes): + return TypeFacts(kind="literal", args=(TypeFacts(kind="bytes"),), runtime_repr=text) + if text in {"True", "False"}: + return TypeFacts(kind="literal", args=(TypeFacts(kind="bool"),), runtime_repr=text) + if text == "None": + return TypeFacts(kind="none", optional=True, runtime_repr="None") + if "." in text and all(part.isidentifier() for part in text.split(".")): + # Enum member like `Color.RED`: fallback is the enum class. + return TypeFacts(kind="literal", args=(TypeFacts(kind=_short_name(text)),), runtime_repr=text) + return None + + +def _parse_callable_type(inner: str, text: str) -> TypeFacts | None: + """`Callable[[a, b], r]` -> TypeFacts(kind="callable", callable=...).""" + pieces = _split_top(inner, ",") + if len(pieces) != 2: + return TypeFacts(kind="callable", runtime_repr=text) + params_text, ret_text = pieces[0].strip(), pieces[1].strip() + parameters: list[ParameterFacts] = [] + if params_text.startswith("[") and params_text.endswith("]"): + for index, piece in enumerate(_split_top(params_text[1:-1], ",")): + piece = piece.strip() + if not piece or piece == "...": + continue + param_type = _parse_type_text(piece) or TypeFacts(kind="unknown", runtime_repr=piece) + parameters.append(ParameterFacts(name=f"arg{index}", kind="positional", type=param_type)) + return_type = _parse_type_text(ret_text) or TypeFacts(kind="unknown", runtime_repr=ret_text) + return TypeFacts( + kind="callable", + callable=CallableFacts(parameters=tuple(parameters), return_type=return_type), + runtime_repr=text, + ) + + +def _default_static_value(text: str) -> Any: + """`"x"`/`3`/`True` -> StaticValue; _MISSING when unrepresentable.""" + try: + value = ast.literal_eval(text) + except (ValueError, SyntaxError): + return _MISSING + if value is ... or isinstance(value, (tuple, frozenset, complex)): + return _MISSING + return value if isinstance(value, (int, float, bool, str, bytes)) or value is None else _MISSING + + +def _parse_signature_params( + inner: str, +) -> tuple[tuple[ParameterFacts, ...], int, frozenset[str], dict[str, StaticValue]]: + """Parse the parameter list of a `def` hover block. + + `/` and bare `*` separators map to positional-only and keyword-only + markers; `*name`/`**name` become varargs/kwargs; `= value` marks a + default (unrepresentable values stay out of `defaults`). + """ + parameters: list[ParameterFacts] = [] + defaults: dict[str, StaticValue] = {} + positional_only = 0 + keyword_only: set[str] = set() + seen_star = False + for raw in _split_top(inner, ","): + piece = raw.strip() + if not piece: + continue + if piece == "/": + positional_only = len(parameters) + continue + if piece == "*": + seen_star = True + continue + stars = 0 + while piece.startswith("*"): + stars += 1 + piece = piece[1:].lstrip() + name = "" + type_text = "" + default_text: str | None = None + colon = _top_level_char(piece, ":") + equals = _top_level_char(piece, "=") + if colon >= 0: + name = piece[:colon].strip() + rest = piece[colon + 1 :] + eq = _top_level_char(rest, "=") + if eq >= 0: + type_text, default_text = rest[:eq], rest[eq + 1 :] + else: + type_text = rest + elif equals >= 0: + name = piece[:equals].strip() + default_text = piece[equals + 1 :] + else: + name = piece + if not name or not all(part.isidentifier() for part in name.split(".")): + continue + name = name.strip() + param_type = _parse_type_text(type_text.strip()) or TypeFacts( + kind="unknown", runtime_repr=type_text.strip() or None + ) + has_default = default_text is not None + if default_text is not None: + value = _default_static_value(default_text.strip()) + if value is not _MISSING: + defaults[name] = value + if stars >= 2: + kind = "kwargs" + elif stars == 1: + kind = "varargs" + seen_star = True + elif seen_star: + kind = "keyword_optional" if has_default else "keyword_only" + keyword_only.add(name) + else: + kind = "positional_optional" if has_default else "positional" + parameters.append(ParameterFacts(name=name, kind=kind, type=param_type, has_default=has_default)) + return tuple(parameters), positional_only, frozenset(keyword_only), defaults + + +def _top_level_char(text: str, char: str) -> int: + """Index of `char` at bracket depth 0 outside quotes, or -1.""" + depth = 0 + quote = "" + for index, current in enumerate(text): + if quote: + if current == "\\": + continue + if current == quote: + quote = "" + elif current in "\"'": + quote = current + elif current in "([{": + depth += 1 + elif current in ")]}": + depth -= 1 + elif current == char and depth == 0: + return index + return -1 + + +def _parse_def_signature(value: str) -> CallableFacts | None: + """Parse `(function|method) def name(params) -> ret` hover text.""" + first_block = value.split("\n\n", 1)[0].strip() + match = _HOVER_DEF.match(first_block) + if not match: + return None + params_text = match.group("params") + return_text = match.group("ret") or "" + parameters, positional_only, keyword_only, defaults = _parse_signature_params(params_text) + return_type = _parse_type_text(return_text.strip()) or TypeFacts( + kind="unknown", runtime_repr=return_text.strip() or None + ) + return CallableFacts( + parameters=parameters, + return_type=return_type, + positional_only=positional_only, + keyword_only=keyword_only, + defaults=defaults, + ) + + +_HOVER_TAG = re.compile(r"^\((?P[\w ]+)\)\s*", re.S) +_HOVER_DEF = re.compile(r"^(?:\([\w ]+\)\s*)?def\s+\w+\s*\((?P.*)\)\s*->\s*(?P.+)$", re.S) +_HOVER_DECL = re.compile(r"^\((?Pvariable|parameter|property)\)\s*\w[\w.]*\s*:\s*(?P.+)$", re.S) +_HOVER_CLASS = re.compile(r"^(?:\([\w ]+\)\s*)?class\s+(?P\w+)") + + +def _parse_hover_type(value: str) -> TypeFacts | None: + """Map a hover `value` string to the TypeFacts it describes.""" + if not value: + return None + block = value.split("\n\n", 1)[0].strip() + # Strip an overload-count tail like "\n(+2 overloads)". + block = re.split(r"\n\s*\(\+\d+ overload[s]?\)", block, maxsplit=1)[0].strip() + tag_match = _HOVER_TAG.match(block) + tag = tag_match.group("tag") if tag_match else "" + if tag == "module": + return TypeFacts(kind="module", runtime_repr=block) + if tag == "type parameter": + return TypeFacts(kind="typevar", runtime_repr=block) + signature = _parse_def_signature(block) + if signature is not None: + return TypeFacts(kind="callable", callable=signature, runtime_repr=block) + decl = _HOVER_DECL.match(block) + if decl: + rest = re.sub(r"\s*\((readonly|read-only|final)\)\s*$", "", decl.group("rest").strip()) + return _parse_type_text(rest) + class_match = _HOVER_CLASS.match(block) + if class_match: + return TypeFacts( + kind="type", + args=(TypeFacts(kind=class_match.group("name")),), + runtime_repr=block, + ) + return None + + +def _is_overload_decorator(expr: ast.expr) -> bool: + if isinstance(expr, ast.Name): + return expr.id == "overload" + if isinstance(expr, ast.Attribute): + return expr.attr == "overload" + return False + + +class PyrightFactsAdapter: + """Facts backed by the pyright-langserver stdio oracle. + + One long-lived subprocess is spawned lazily and reused across `facts` + calls; each target file is opened once via `textDocument/didOpen`. + Function/method signatures come from hovering the `def` name; + expression facts come from hovering symbol positions (call sites take + the callee's return type). Annotated assignment targets get their + declared type from the annotation's own source text because hover + reports the *narrowed* type. Anything the oracle declines to describe + — literals, operators, unsymbolized spans — degrades to "no fact" + rather than an error. + + `command` overrides the spawn line (default: `PYRIGHT_LANGSERVER_BIN` + env or `pyright-langserver` on PATH); `--stdio` is appended when + absent. `close()` shuts the server down; leaving it to GC also works. + """ + + name = "pyright" + + def __init__(self, command: str | Sequence[str] | None = None) -> None: + raw = command if command is not None else os.environ.get("PYRIGHT_LANGSERVER_BIN") + argv = shlex.split(raw) if isinstance(raw, str) else list(raw or ["pyright-langserver"]) + if "--stdio" not in argv: + argv = [*argv, "--stdio"] + self._argv = argv + self._connection: _LspConnection | None = None + self._opened: set[str] = set() + self._hover_cache: dict[tuple[str, int, int], str | None] = {} + + def facts(self, path: Path) -> ModuleFacts: + path = path.resolve() + try: + source = path.read_text(encoding="utf-8") + except OSError as exc: + raise CheckFailed(f"cannot read {path}: {exc}") from exc + try: + tree = ast.parse(source, filename=str(path)) + except SyntaxError as exc: + raise CheckFailed(f"cannot parse {path}: {exc}") from exc + connection = self._connect() + uri = path.as_uri() + self._open_document(uri, source) + expressions = self._expression_facts(connection, uri, tree, source) + callables, overloads = self._callable_facts(connection, uri, tree, source) + return ModuleFacts( + path=path, + provider=self.name, + expressions=expressions, + callables=callables, + overloads=overloads, + ) + + def close(self) -> None: + if self._connection is not None: + self._connection.shutdown() + self._connection = None + self._opened.clear() + self._hover_cache.clear() + + def __enter__(self) -> PyrightFactsAdapter: + return self + + def __exit__(self, *_exc: object) -> None: + self.close() + + def __del__(self) -> None: + try: + self.close() + except Exception: + pass + + def _connect(self) -> _LspConnection: + if self._connection is None: + connection = _LspConnection(self._argv) + connection.start() + result = connection.request( + "initialize", + { + "processId": None, + "rootUri": None, + "capabilities": { + "textDocument": { + "hover": {"contentFormat": ["plaintext"]}, + "publishDiagnostics": {}, + }, + "workspace": {"workspaceFolders": True}, + }, + "workspaceFolders": None, + }, + timeout=60, + ) + if result is None: + connection.shutdown() + raise CheckFailed("pyright langserver refused initialize") + connection.notify("initialized", {}) + connection.notify( + "workspace/didChangeConfiguration", + { + "settings": { + "python": { + "analysis": { + "stubPath": str(_PYRIGHT_STUBS), + "pythonVersion": "3.13", + } + } + } + }, + ) + self._connection = connection + return self._connection + + def _open_document(self, uri: str, source: str) -> None: + connection = self._connect() + if uri in self._opened: + return + self._opened.add(uri) + # Point the workspace at the file's directory so sibling imports + # resolve the same way the pyright checker preflight does. + connection.notify( + "workspace/didChangeWorkspaceFolders", + {"event": {"added": [{"uri": uri.rsplit("/", 1)[0], "name": "pyscriptc"}], "removed": []}}, + ) + connection.notify( + "textDocument/didOpen", + { + "textDocument": { + "uri": uri, + "languageId": "python", + "version": 1, + "text": source, + } + }, + ) + connection.wait_notification( + "textDocument/publishDiagnostics", + lambda params: isinstance(params, dict) and params.get("uri") == uri, + timeout=30, + ) + + def _hover(self, connection: _LspConnection, uri: str, line: int, char: int) -> str | None: + key = (uri, line, char) + if key in self._hover_cache: + return self._hover_cache[key] + result = connection.request( + "textDocument/hover", + {"textDocument": {"uri": uri}, "position": {"line": line, "character": char}}, + ) + value: str | None = None + if isinstance(result, dict): + contents = result.get("contents") + if isinstance(contents, dict): + text = contents.get("value") + if isinstance(text, str): + value = text + elif isinstance(contents, str): + value = contents + elif isinstance(contents, list): + texts = [ + item["value"] for item in contents if isinstance(item, dict) and isinstance(item.get("value"), str) + ] + value = "\n\n".join(texts) if texts else None + self._hover_cache[key] = value + return value + + def _expression_facts( + self, connection: _LspConnection, uri: str, tree: ast.Module, source: str + ) -> dict[str, TypeFacts]: + lines = source.splitlines() + facts: dict[str, TypeFacts] = {} + # Declared types first: an AnnAssign target's fact is the annotation + # itself, parsed from source — hover would report the narrowed type. + declared: dict[tuple[int, int], ast.expr] = {} + annotations: set[int] = set() + for node in ast.walk(tree): + if isinstance(node, ast.AnnAssign) and node.annotation is not None: + annotations.add(id(node.annotation)) + declared[(node.target.lineno, node.target.col_offset)] = node.annotation + elif isinstance(node, ast.arg) and node.annotation is not None: + annotations.add(id(node.annotation)) + for node in ast.walk(tree): + if not isinstance(node, ast.expr): + continue + key = f"{node.lineno}:{node.col_offset}" + if key in facts: + continue + if id(node) in annotations: + parsed = _parse_type_text(ast.get_source_segment(source, node) or "") + if parsed is not None: + facts[key] = TypeFacts(kind="type", args=(parsed,), runtime_repr=parsed.runtime_repr) + continue + declared_annotation = declared.get((node.lineno, node.col_offset)) + if declared_annotation is not None: + parsed = _parse_type_text(ast.get_source_segment(source, declared_annotation) or "") + if parsed is not None: + facts[key] = parsed + continue + position = self._node_hover_position(node, lines) + if position is None: + continue + line, char = position + value = self._hover(connection, uri, line, char) + if value is None: + continue + if isinstance(node, ast.Call): + signature = _parse_def_signature(value.split("\n\n", 1)[0].strip()) + if signature is not None: + facts[key] = signature.return_type + continue + parsed = _parse_hover_type(value) + if parsed is not None: + facts[key] = parsed + return facts + + def _node_hover_position(self, node: ast.expr, lines: list[str]) -> tuple[int, int] | None: + """LSP (0-based line, UTF-16 col) where hover describes this node.""" + if isinstance(node, ast.Call): + node = node.func + if not isinstance(node, (ast.Name, ast.Attribute)): + return None + if isinstance(node, ast.Name): + if node.lineno - 1 >= len(lines): + return None + return node.lineno - 1, _utf16_col(lines[node.lineno - 1], node.col_offset) + if isinstance(node, ast.Attribute): + # The node's start is the base; the attribute name sits at the end. + if node.end_lineno != node.lineno or node.end_col_offset is None: + return None + if node.lineno - 1 >= len(lines): + return None + attr_start = node.end_col_offset - len(node.attr.encode("utf-8")) + if attr_start < 0: + return None + return node.lineno - 1, _utf16_col(lines[node.lineno - 1], attr_start) + return None + + def _callable_facts( + self, connection: _LspConnection, uri: str, tree: ast.Module, source: str + ) -> tuple[dict[str, CallableFacts], dict[str, tuple[CallableFacts, ...]]]: + lines = source.splitlines() + callables: dict[str, CallableFacts] = {} + overloads: dict[str, tuple[CallableFacts, ...]] = {} + + def visit(body: list[ast.stmt], prefix: str) -> None: + for stmt in body: + if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef)): + qualname = f"{prefix}{stmt.name}" + position = self._def_hover_position(stmt, lines) + signature = ( + _parse_def_signature(self._hover(connection, uri, *position) or "") + if position is not None + else None + ) + if signature is None: + continue + if any(_is_overload_decorator(dec) for dec in stmt.decorator_list): + overloads.setdefault(qualname, ()) + overloads[qualname] += (signature,) + else: + callables[qualname] = signature + visit(stmt.body, f"{prefix}{stmt.name}..") + + visit(tree.body, "__main__.") + return callables, overloads + + def _def_hover_position( + self, node: ast.FunctionDef | ast.AsyncFunctionDef, lines: list[str] + ) -> tuple[int, int] | None: + line_index = node.lineno - 1 + if line_index >= len(lines): + return None + text = lines[line_index][node.col_offset :] + marker = "async def " if isinstance(node, ast.AsyncFunctionDef) else "def " + offset = text.find(marker) + if offset < 0: + return None + name_col = node.col_offset + offset + len(marker) + return line_index, _utf16_col(lines[line_index], name_col) From bf7634b8fb73aee4df302df345550965e2e9f823 Mon Sep 17 00:00:00 2001 From: espetro Date: Sun, 18 Oct 2026 19:47:22 +0000 Subject: [PATCH 2/3] =?UTF-8?q?feat(check):=20PyrightFactsAdapter=20?= =?UTF-8?q?=E2=80=94=20pyright-langserver=20LSP=20hover=20oracle?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements FactsProvider against a long-lived pyright-langserver subprocess over stdio JSON-RPC: lazy spawn, hover + didOpen with publishDiagnostics wait, per-(uri,line,char) cache, clean shutdown. Hover plaintext is parsed back into the existing TypeFacts/CallableFacts dataclasses; unexpected shapes degrade to 'no fact'. - Binary resolution: PYRIGHT_LANGSERVER_BIN env (shlex) or constructor command, default 'pyright-langserver --stdio' - facts_provider('pyright') now dispatches to the adapter - AnnAssign targets take the declared annotation type (hover reports narrowed types); Call expressions record the callee's return type; each @overload def name hover contributes an overload choice - tests: hover-text parser units + skip-marked langserver smoke test - self-census ballpark +139 fences for the stdio/threading plumbing --- src/pyscriptc/typefacts.py | 20 ++- tests/test_census_self.py | 8 +- tests/test_pyright_facts.py | 267 ++++++++++++++++++++++++++++++++++++ tests/test_typefacts.py | 4 +- 4 files changed, 290 insertions(+), 9 deletions(-) create mode 100644 tests/test_pyright_facts.py diff --git a/src/pyscriptc/typefacts.py b/src/pyscriptc/typefacts.py index 03eb4ed..4f73dd2 100644 --- a/src/pyscriptc/typefacts.py +++ b/src/pyscriptc/typefacts.py @@ -451,6 +451,13 @@ def facts(self, path: Path) -> ModuleFacts: _PYRIGHT_STUBS = Path(__file__).with_name("_pyright_stubs") +# Named tuples because ruff (target-version py314) strips except-parens +# into `except A, B:` — valid only on 3.14+, broken on the 3.13 dev venv. +_PIPE_ERRORS = (OSError, CheckFailed) +_SHUTDOWN_ERRORS = _PIPE_ERRORS + (subprocess.TimeoutExpired,) +_READ_ERRORS = (OSError, ValueError, json.JSONDecodeError) +_EVAL_ERRORS = (ValueError, SyntaxError) + class _LspConnection: """Minimal synchronous JSON-RPC client for one stdio langserver. @@ -514,7 +521,7 @@ def notify(self, method: str, params: Any) -> None: return try: self._send({"jsonrpc": "2.0", "method": method, "params": params}) - except (OSError, CheckFailed): + except _PIPE_ERRORS: self._dead = True def wait_notification( @@ -546,7 +553,7 @@ def shutdown(self) -> None: self.request("shutdown", None, timeout=10) self.notify("exit", None) proc.wait(timeout=5) - except (CheckFailed, subprocess.TimeoutExpired, OSError): + except _SHUTDOWN_ERRORS: pass if proc.poll() is None: proc.kill() @@ -594,7 +601,7 @@ def _read_loop(self) -> None: self._mark_dead() return message = json.loads(body) - except (OSError, ValueError, json.JSONDecodeError): + except _READ_ERRORS: self._mark_dead() return self._dispatch(message) @@ -817,7 +824,7 @@ def _parse_literal_piece(text: str) -> TypeFacts | None: """A `Literal[...]` member or a bare literal: 1, 's', True, None, Color.RED.""" try: value = ast.literal_eval(text) - except (ValueError, SyntaxError): + except _EVAL_ERRORS: value = None if isinstance(value, bool): return TypeFacts(kind="literal", args=(TypeFacts(kind="bool"),), runtime_repr=text) @@ -865,7 +872,7 @@ def _default_static_value(text: str) -> Any: """`"x"`/`3`/`True` -> StaticValue; _MISSING when unrepresentable.""" try: value = ast.literal_eval(text) - except (ValueError, SyntaxError): + except _EVAL_ERRORS: return _MISSING if value is ... or isinstance(value, (tuple, frozenset, complex)): return _MISSING @@ -1274,6 +1281,9 @@ def _callable_facts( def visit(body: list[ast.stmt], prefix: str) -> None: for stmt in body: + if isinstance(stmt, ast.ClassDef): + visit(stmt.body, f"{prefix}{stmt.name}.") + continue if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef)): qualname = f"{prefix}{stmt.name}" position = self._def_hover_position(stmt, lines) diff --git a/tests/test_census_self.py b/tests/test_census_self.py index ebb7ef2..fabfbb7 100644 --- a/tests/test_census_self.py +++ b/tests/test_census_self.py @@ -19,6 +19,8 @@ def test_self_census_ballpark() -> None: # telemetry.py added 36 fences, log.py 20, debug.py 6 union-type # fences (`X | None` parameters); R1–R5 batch un-fenced 10 self-host # call sites (list/dict/str methods). - assert census["fences"] == 1023 - assert 11_000 <= census["lines"] <= 21_000 - assert sum(p["fences"] for p in census["packages"]) == 1023 + # PyrightFactsAdapter (typefacts.py) added 139 fences: subprocess/threads/ + # JSON-RPC stdio plumbing has no scriptc-subset equivalent. + assert census["fences"] == 1162 + assert 11_000 <= census["lines"] <= 22_000 + assert sum(p["fences"] for p in census["packages"]) == 1162 diff --git a/tests/test_pyright_facts.py b/tests/test_pyright_facts.py new file mode 100644 index 0000000..fdef62c --- /dev/null +++ b/tests/test_pyright_facts.py @@ -0,0 +1,267 @@ +"""Pyright facts oracle: hover-text parser units + langserver smoke test. + +`PYRIGHT_LANGSERVER_BIN` picks the oracle command (default +`pyright-langserver` on PATH); the smoke test is skip-marked when the +binary is absent. Everything else exercises the hover-string parsers +with fixtures captured from a real server — no network, no subprocess. +""" + +from __future__ import annotations + +import os +import shlex +import shutil +from pathlib import Path + +import pytest +from tests.test_gates import write_case + +from pyscriptc.check import CheckFailed +from pyscriptc.typefacts import ( + FactsProvider, + PyrightFactsAdapter, + _parse_def_signature, + _parse_hover_type, + _parse_type_text, +) + +# --- type-text parser ------------------------------------------------------ + + +def test_parse_generic_type() -> None: + facts = _parse_type_text("list[int]") + assert facts is not None + assert facts.kind == "list" + assert [arg.kind for arg in facts.args] == ["int"] + + +def test_parse_union_marks_optional() -> None: + facts = _parse_type_text("int | None") + assert facts is not None + assert facts.kind == "union" + assert facts.optional is True + assert {arg.kind for arg in facts.args} == {"int", "none"} + plain = _parse_type_text("str | int") + assert plain is not None and plain.kind == "union" and not plain.optional + + +def test_parse_optional_and_union_forms() -> None: + optional = _parse_type_text("Optional[int]") + assert optional is not None + assert optional.optional is True + assert {arg.kind for arg in optional.args} == {"int", "none"} + union = _parse_type_text("Union[int, str, None]") + assert union is not None and union.optional is True + + +def test_parse_tuple_type() -> None: + facts = _parse_type_text("tuple[int, str]") + assert facts is not None + assert facts.kind == "tuple" + assert [arg.kind for arg in facts.args] == ["int", "str"] + homogeneous = _parse_type_text("tuple[int, ...]") + assert homogeneous is not None and [arg.kind for arg in homogeneous.args] == ["int"] + + +def test_parse_callable_type() -> None: + facts = _parse_type_text("Callable[[int, str], bool]") + assert facts is not None + assert facts.kind == "callable" + assert facts.callable is not None + assert [p.type.kind for p in facts.callable.parameters] == ["int", "str"] + assert facts.callable.return_type.kind == "bool" + + +def test_parse_type_typevar_literal() -> None: + type_fact = _parse_type_text("type[Box]") + assert type_fact is not None and type_fact.kind == "type" and type_fact.args[0].kind == "Box" + typevar = _parse_type_text("Self@Box") + assert typevar is not None and typevar.kind == "typevar" + literal = _parse_type_text("Literal[1]") + assert literal is not None and literal.kind == "literal" and literal.args[0].kind == "int" + + +def test_parse_specials() -> None: + for text, kind in [("None", "none"), ("Any", "any"), ("Unknown", "any"), ("Never", "unknown")]: + facts = _parse_type_text(text) + assert facts is not None and facts.kind == kind, text + + +def test_parse_nested_generics() -> None: + facts = _parse_type_text("dict[str, list[int]] | None") + assert facts is not None and facts.kind == "union" and facts.optional + member = next(arg for arg in facts.args if arg.kind == "dict") + assert [arg.kind for arg in member.args] == ["str", "list"] + + +def test_parse_garbage_degrades() -> None: + assert _parse_type_text("") is None + assert _parse_type_text("...") is None + assert _parse_type_text("foo(1, 2") is None + + +# --- signature/hover block parser ------------------------------------------ + + +def test_parse_simple_def_signature() -> None: + signature = _parse_def_signature("(function) def greet(name: str) -> str") + assert signature is not None + assert signature.return_type.kind == "str" + assert [(p.name, p.kind, p.type.kind) for p in signature.parameters] == [("name", "positional", "str")] + + +def test_parse_multiline_signature_with_markers() -> None: + value = ( + "(function) def f(\n" + " a: int,\n" + " /,\n" + ' b: str = "x",\n' + " *args: float,\n" + " c: bool = True,\n" + " **kw: int\n" + ") -> list[int]" + ) + signature = _parse_def_signature(value) + assert signature is not None + assert signature.positional_only == 1 + assert signature.keyword_only == frozenset({"c"}) + assert signature.defaults == {"b": "x", "c": True} + assert signature.return_type.kind == "list" + assert [p.kind for p in signature.parameters] == [ + "positional", + "positional_optional", + "varargs", + "keyword_optional", + "kwargs", + ] + + +def test_parse_method_signature_self_typevar() -> None: + signature = _parse_def_signature("(method) def get(self: Self@Box) -> int") + assert signature is not None + assert signature.parameters[0].type.kind == "typevar" + assert signature.return_type.kind == "int" + + +def test_parse_hover_variable_decl() -> None: + facts = _parse_hover_type("(variable) x: int | None") + assert facts is not None + assert facts.kind == "union" + assert facts.optional + + +def test_parse_hover_function_ref_is_callable() -> None: + facts = _parse_hover_type("(function) def greet(name: str) -> str") + assert facts is not None + assert facts.kind == "callable" + assert facts.callable is not None and facts.callable.return_type.kind == "str" + + +def test_parse_hover_class_ref_is_type() -> None: + facts = _parse_hover_type("class Box()") + assert facts is not None + assert facts.kind == "type" + assert facts.args[0].kind == "Box" + + +def test_parse_hover_module() -> None: + facts = _parse_hover_type("(module) typing") + assert facts is not None and facts.kind == "module" + + +def test_parse_hover_strips_doc_block() -> None: + facts = _parse_hover_type("(variable) x: int\n\nSome docstring text.") + assert facts is not None and facts.kind == "int" + + +def test_parse_hover_overload_tail_ignored() -> None: + facts = _parse_hover_type("(function) def pick(v: int) -> int\n(+1 overload)") + assert facts is not None and facts.kind == "callable" + + +def test_parse_hover_garbage_degrades() -> None: + assert _parse_hover_type("") is None + assert _parse_hover_type("hover text we do not understand") is None + + +# --- adapter plumbing ------------------------------------------------------- + + +def test_adapter_is_a_facts_provider() -> None: + assert isinstance(PyrightFactsAdapter(), FactsProvider) + assert PyrightFactsAdapter().name == "pyright" + + +def test_adapter_command_resolution(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("PYRIGHT_LANGSERVER_BIN", raising=False) + assert PyrightFactsAdapter()._argv == ["pyright-langserver", "--stdio"] + monkeypatch.setenv("PYRIGHT_LANGSERVER_BIN", "node /srv/langserver.js --stdio") + assert PyrightFactsAdapter()._argv == ["node", "/srv/langserver.js", "--stdio"] + + +def test_missing_langserver_raises_check_failed(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("PYRIGHT_LANGSERVER_BIN", "pyright-langserver-does-not-exist-xyz") + path = write_case(tmp_path, "probe.py", "def probe() -> int:\n return 1\n") + adapter = PyrightFactsAdapter() + with pytest.raises(CheckFailed, match="not installed"): + adapter.facts(path) + + +# --- smoke test against the real langserver --------------------------------- + + +def _langserver_command() -> list[str]: + raw = os.environ.get("PYRIGHT_LANGSERVER_BIN") + argv = shlex.split(raw) if raw else ["pyright-langserver"] + return [*argv, "--stdio"] if "--stdio" not in argv else argv + + +_langserver_argv = _langserver_command() +_langserver_present = shutil.which(_langserver_argv[0]) is not None or Path(_langserver_argv[0]).exists() + + +@pytest.mark.skipif(not _langserver_present, reason="pyright langserver not installed") +def test_langserver_facts_smoke(tmp_path: Path) -> None: + source = """\ +def greet(name: str) -> str: + return "hi " + name + +x: int | None = 1 +total = greet("w") +""" + path = write_case(tmp_path, "smoke.py", source) + with PyrightFactsAdapter() as adapter: + facts = adapter.facts(path) + assert facts.provider == "pyright" + signature = facts.signature("__main__.greet") + assert signature is not None + assert signature.return_type.kind == "str" + declared = facts.expression_at(4, 0) # `x` gets its declared union, not narrowed + assert declared is not None and declared.kind == "union" and declared.optional + call = facts.expression_at(5, 8) # `greet("w")` records the return type + assert call is not None and call.kind == "str" + + +@pytest.mark.skipif(not _langserver_present, reason="pyright langserver not installed") +def test_langserver_overloads_and_methods(tmp_path: Path) -> None: + source = """\ +from typing import overload + +@overload +def pick(v: int) -> int: ... +@overload +def pick(v: str) -> str: ... +def pick(v: object) -> object: + return v + +class Box: + def get(self) -> int: + return 0 +""" + path = write_case(tmp_path, "ov.py", source) + with PyrightFactsAdapter() as adapter: + facts = adapter.facts(path) + assert facts.signature("__main__.Box.get") is not None + variants = facts.overload_choices("__main__.pick") + assert {variant.return_type.kind for variant in variants} == {"int", "str"} + assert facts.signature("__main__.pick") is not None diff --git a/tests/test_typefacts.py b/tests/test_typefacts.py index b6b6f27..3dc94c2 100644 --- a/tests/test_typefacts.py +++ b/tests/test_typefacts.py @@ -13,6 +13,7 @@ FactsProvider, MypyFactsAdapter, ParameterFacts, + PyrightFactsAdapter, TypeFacts, facts_provider, ) @@ -121,8 +122,9 @@ def test_errors_surface_as_check_failed(tmp_path: Path) -> None: def test_facts_provider_dispatch(tmp_path: Path) -> None: provider = facts_provider("mypy") assert isinstance(provider, MypyFactsAdapter) + assert isinstance(facts_provider("pyright"), PyrightFactsAdapter) with pytest.raises(CheckFailed): - facts_provider("pyright") + facts_provider("other") def test_dataclasses_match_the_plan(tmp_path: Path) -> None: From 6825fb411ffe436a7f46f8ff6f21d1f56a85256d Mon Sep 17 00:00:00 2001 From: espetro Date: Sun, 18 Oct 2026 20:05:37 +0000 Subject: [PATCH 3/3] test: census ballpark 1189 after pyright-facts-adapter merge --- tests/test_census_self.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/tests/test_census_self.py b/tests/test_census_self.py index 8ffaf56..09d2943 100644 --- a/tests/test_census_self.py +++ b/tests/test_census_self.py @@ -19,6 +19,8 @@ def test_self_census_ballpark() -> None: # telemetry.py added 36 fences, log.py 20, debug.py 6 union-type # fences (`X | None` parameters); R1–R5 batch un-fenced 10 self-host # call sites (list/dict/str methods). - assert census["fences"] == 1050 - assert 11_000 <= census["lines"] <= 21_500 - assert sum(p["fences"] for p in census["packages"]) == 1050 + # PyrightFactsAdapter (typefacts.py) added 139 fences: subprocess/threads/ + # JSON-RPC stdio plumbing has no scriptc-subset equivalent. + assert census["fences"] == 1189 + assert 11_000 <= census["lines"] <= 22_000 + assert sum(p["fences"] for p in census["packages"]) == 1189