orchestrata_ultimate.tools.lsp_tool

Reference › orchestrata_ultimate › tools › lsp_tool

LSPTool (SPEC-065) gives volks code navigation and diagnostics for Python files. It is a BaseTool named lsp in the search category with priority P0, registered by agents.tool_registry and exported from the package __init__. tests/tools/test_lsp_tool.py covers it.

The module records a deliberate backend choice. No language-server binary is available, so it does not implement a JSON-RPC LSP client. Diagnostics come from PyrightDiagnosticsBackend, which runs the batch pyright command with JSON output. Definition, references, document symbols and hover come from PythonSymbolIndex, an AST index that resolves symbols by name, not by type inference. This is a stated trade-off.

execute takes an action, a file uri and, for navigation, a position. The actions are definition, references, diagnostics, hover and document_symbols. It converts the URI to a path, requires a Python file for non-diagnostic actions, finds the identifier at the position, and searches the paths resolved from the arguments. Data classes Position, Range, LSPLocation, DocumentSymbolInfo, LSPHover, LSPDiagnostic and LSPDiagnostics carry results, and LSPError with LSPErrorCode reports failures. execute never raises: errors become failed ToolResult values. It carries @traces for REQ-065-001 and related requirements.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

LSPErrorCode

Line 78

Error codes for LSPTool (spec.md Error Conditions, ERR-00N).

LSPError

Bases: Exception · Line 101

Raised by the LSP pipeline on any LSPErrorCode condition.

Position (dataclass)

Decorators: dataclass(slots=True, frozen=True) · Line 129

A zero-based line/character cursor position (spec.md Position).

Range (dataclass)

Decorators: dataclass(slots=True, frozen=True) · Line 140

A start/end Position span (spec.md Range).

LSPLocation (dataclass)

Decorators: traces(spec='065', story='US-065-003', reqs=['REQ-065-001', 'REQ-065-002']), dataclass(slots=True, frozen=True) · Line 152

A URI + range reference (spec.md LSPLocation).

DocumentSymbolInfo (dataclass)

Decorators: traces(spec='065', story='US-065-003', reqs=['REQ-065-003']), dataclass(slots=True, frozen=True) · Line 165

One entry of textDocument/documentSymbol (FR-005 subset).

LSPHover (dataclass)

Decorators: traces(spec='065', story='US-065-004', reqs=['REQ-065-004']), dataclass(slots=True, frozen=True) · Line 179

Hover content + optional target range (spec.md LSPHover).

LSPDiagnostic (dataclass)

Decorators: traces(spec='065', story='US-065-002', reqs=['REQ-065-006']), dataclass(slots=True, frozen=True) · Line 195

One diagnostic item (spec.md LSPDiagnostics.diagnostics[]).

LSPDiagnostics (dataclass)

Decorators: traces(spec='065', story='US-065-002', reqs=['REQ-065-006']), dataclass(slots=True) · Line 220

Diagnostics for one document (spec.md LSPDiagnostics).

PythonSymbolIndex

Decorators: traces(spec='065', story='US-065-003', reqs=['REQ-065-001', 'REQ-065-002', 'REQ-065-003', 'REQ-065-004']) · Line 438

Token-precise, name-resolution navigation over Python source (MVP backend).

PyrightDiagnosticsBackend

Decorators: traces(spec='065', story='US-065-002', reqs=['REQ-065-006', 'REQ-065-007']) · Line 601

Wraps the real pyright CLI (a declared dev dependency) for diagnostics.

LSPTool

Bases: BaseTool · Decorators: traces(spec='065', story='US-065-003', reqs=['REQ-065-016']) · Line 774

LSPTool — code navigation + diagnostics for Python targets (SPEC-065).

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/tools/lsp_tool.py

"""LSPTool — code navigation for Python targets (SPEC-065).

Backend reality check (see the brief this module was built from) before any
line of implementation:

- ``pyright`` IS a declared dev dependency (``pyproject.toml``, "pyright>=1.1.410")
  and IS runnable in this repo — ``scripts/pyright_ratchet.py`` already shells
  out to it as ``pyright <source> --outputjson`` for the pyright-error-count
  gate. That binary is a **batch type-checker**: one-shot ``pyright <path>
  --outputjson`` produces a diagnostics report. It is NOT the same binary as
  ``pyright-langserver``, which speaks LSP JSON-RPC over stdio — that
  long-lived, stateful server binary is not declared/installed here.
- Neither ``python-lsp-server`` (pylsp) nor ``jedi`` is declared in
  ``pyproject.toml``. No dependency was added to get either (per the brief:
  "Do NOT add dependencies").

**Chosen backend (the documented fork):** a JSON-RPC LSP client (initialize/
textDocument-definition/references) was rejected for this MVP because there
is no LSP *server* binary available on PATH to talk to (only the batch
``pyright`` CLI, and no ``jedi``/``pylsp``). Building the client without a
compliant server to test it against would be untested vaporware. Instead:

- **Diagnostics** (US-002) uses the REAL ``pyright`` CLI already wired
  elsewhere in this repo (:class:`PyrightDiagnosticsBackend`), batch mode
  (``--outputjson``), not push/debounced streaming — a faithful but
  synchronous subset of FR-004.
- **Definition / references / document-symbols / hover** (US-003, US-004,
  and the navigation half of FR-005/FR-006) are implemented over this
  repo's existing AST-based code-intelligence engine style (the same
  token-precise approach used by ``scripts/serialize_codebase_ast.py``'s
  ``find_references``/``list_symbols`` and
  ``orchestrata_ultimate.tools.ast_query``), reimplemented locally in
  :class:`PythonSymbolIndex` so this module has no dependency on the
  9000+ line toolkit script's internals. It resolves symbols by NAME
  match within the AST (not full type inference — no ``jedi``/pyright
  language-server available to resolve by type), which is a stated,
  intentional trade-off for the MVP.
- **Completion (FR-003), code actions (FR-007), document formatting
  (FR-008), rename (FR-009), signature help, and persistent multi-server
  lifecycle/session state (FR-001/FR-002)** are NOT implemented — they
  need a live, stateful LSP session (edits, incremental sync, applying
  workspace edits) that a one-shot AST read or a batch pyright invocation
  cannot provide. The hardening path is a real stdio JSON-RPC client
  against ``pyright-langserver`` (or ``jedi``/``python-lsp-server`` once
  declared as a dependency), gated at runtime by
  ``shutil.which(...)`` exactly like :class:`PyrightDiagnosticsBackend`
  below already gates ``pyright``.

Only Python targets are supported (SPEC-065's other listed servers —
typescript-language-server / rust-analyzer / gopls / clangd — are neither
installed nor declared here; requesting a non-``.py`` file raises
``LSPErrorCode.UNSUPPORTED_LANGUAGE``, ERR-011).
"""

from __future__ import annotations

import ast
import json
import re
import shutil
import subprocess
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.tools.base import (
    BaseTool,
    ToolCategory,
    ToolMetadata,
    ToolPriority,
    ToolResult,
)

# ── error codes ──────────────────────────────────────────────────────────


class LSPErrorCode:
    """Error codes for LSPTool (spec.md Error Conditions, ERR-00N).

    Only the subset reachable by this MVP's stateless, one-shot backend is
    represented — spec.md conditions that require a persistent LSP server
    process (crash/restart, memory limits, protocol-level JSON-RPC errors,
    initialization timeout) do not apply here and are documented as
    deferred in ``specs/065-lsp-tool/checklists/requirements.md``.
    """

    SERVER_NOT_FOUND = "server_not_found"  # ERR-001
    REQUEST_TIMEOUT = "request_timeout"  # ERR-004
    INVALID_URI = "invalid_uri"  # ERR-005
    SERVER_ERROR = "server_error"  # ERR-007
    UNSUPPORTED_LANGUAGE = "unsupported_language"  # ERR-011
    FILE_READ_ERROR = "file_read_error"  # ERR-012
    # Not spec-numbered ERR-00N conditions — framework-level input
    # validation / lookup-miss codes needed by any tool dispatch, same
    # convention as ToolSearchTool's non-ERR codes (e.g. TOOL_NOT_FOUND).
    INVALID_REQUEST = "invalid_request"
    SYMBOL_NOT_FOUND = "symbol_not_found"


class LSPError(Exception):
    """Raised by the LSP pipeline on any ``LSPErrorCode`` condition.

    Mirrors ``ToolSearchError`` (tools/search.py): carries a code + message
    and knows how to render itself as a failed :class:`ToolResult` so
    :meth:`LSPTool.execute` never has to translate error codes itself.
    """

    def __init__(self, code: str, message: str) -> None:
        super().__init__(message)
        self.code = code
        self.message = message

    @traces(spec="065", story="US-065-003", reqs=["REQ-065-016"])
    def to_result(self, tool_name: str) -> ToolResult:
        """Render this error as a failed ``ToolResult`` for ``tool_name``."""
        return ToolResult(
            tool_name=tool_name,
            success=False,
            output="",
            error=f"{self.code}: {self.message}",
        )


# ── data models (spec.md Data Models, trimmed to the implemented subset) ──


@dataclass(slots=True, frozen=True)
class Position:
    """A zero-based line/character cursor position (spec.md ``Position``)."""

    line: int
    character: int

    def to_dict(self) -> dict[str, int]:
        return {"line": self.line, "character": self.character}


@dataclass(slots=True, frozen=True)
class Range:
    """A start/end ``Position`` span (spec.md ``Range``)."""

    start: Position
    end: Position

    def to_dict(self) -> dict[str, Any]:
        return {"start": self.start.to_dict(), "end": self.end.to_dict()}


@traces(spec="065", story="US-065-003", reqs=["REQ-065-001", "REQ-065-002"])
@dataclass(slots=True, frozen=True)
class LSPLocation:
    """A URI + range reference (spec.md ``LSPLocation``)."""

    uri: str
    range: Range

    @traces(spec="065", story="US-065-003", reqs=["REQ-065-001", "REQ-065-002"])
    def to_dict(self) -> dict[str, Any]:
        return {"uri": self.uri, "range": self.range.to_dict()}


@traces(spec="065", story="US-065-003", reqs=["REQ-065-003"])
@dataclass(slots=True, frozen=True)
class DocumentSymbolInfo:
    """One entry of ``textDocument/documentSymbol`` (FR-005 subset)."""

    name: str
    kind: str
    range: Range

    @traces(spec="065", story="US-065-003", reqs=["REQ-065-003"])
    def to_dict(self) -> dict[str, Any]:
        return {"name": self.name, "kind": self.kind, "range": self.range.to_dict()}


@traces(spec="065", story="US-065-004", reqs=["REQ-065-004"])
@dataclass(slots=True, frozen=True)
class LSPHover:
    """Hover content + optional target range (spec.md ``LSPHover``)."""

    contents: str
    range: Range | None = None

    @traces(spec="065", story="US-065-004", reqs=["REQ-065-004"])
    def to_dict(self) -> dict[str, Any]:
        payload: dict[str, Any] = {"contents": self.contents}
        if self.range is not None:
            payload["range"] = self.range.to_dict()
        return payload


@traces(spec="065", story="US-065-002", reqs=["REQ-065-006"])
@dataclass(slots=True, frozen=True)
class LSPDiagnostic:
    """One diagnostic item (spec.md ``LSPDiagnostics.diagnostics[]``)."""

    range: Range
    message: str
    severity: int = 1
    code: str | None = None
    source: str | None = None

    @traces(spec="065", story="US-065-002", reqs=["REQ-065-006"])
    def to_dict(self) -> dict[str, Any]:
        payload: dict[str, Any] = {
            "range": self.range.to_dict(),
            "message": self.message,
            "severity": self.severity,
        }
        if self.code is not None:
            payload["code"] = self.code
        if self.source is not None:
            payload["source"] = self.source
        return payload


@traces(spec="065", story="US-065-002", reqs=["REQ-065-006"])
@dataclass(slots=True)
class LSPDiagnostics:
    """Diagnostics for one document (spec.md ``LSPDiagnostics``)."""

    uri: str
    diagnostics: list[LSPDiagnostic] = field(default_factory=list)

    @traces(spec="065", story="US-065-002", reqs=["REQ-065-006"])
    def to_dict(self) -> dict[str, Any]:
        return {
            "uri": self.uri,
            "diagnostics": [d.to_dict() for d in self.diagnostics],
        }


#: spec.md "1=Error, 2=Warning, 3=Info, 4=Hint" vs. pyright's own string
#: severities (``--outputjson``'s ``generalDiagnostics[].severity``).
_PYRIGHT_SEVERITY_MAP = {
    "error": 1,
    "warning": 2,
    "information": 3,
    "hint": 4,
}

_VALID_ACTIONS = frozenset(
    {"definition", "references", "diagnostics", "hover", "document_symbols"}
)


# ── URI / path helpers ─────────────────────────────────────────────────────


def _uri_to_path(uri: str) -> Path:
    """Resolve a ``file://`` URI or bare path to a :class:`Path`.

    Raises:
        LSPError: ``INVALID_URI`` (ERR-005) on an empty or non-``file``
            scheme URI.
    """
    if not uri or not uri.strip():
        raise LSPError(LSPErrorCode.INVALID_URI, "uri must not be blank")
    if uri.startswith("file://"):
        raw = uri[len("file://") :]
    elif "://" in uri:
        raise LSPError(LSPErrorCode.INVALID_URI, f"unsupported URI scheme in {uri!r}")
    else:
        raw = uri
    if not raw:
        raise LSPError(LSPErrorCode.INVALID_URI, f"malformed uri: {uri!r}")
    return Path(raw)


def _path_to_uri(path: Path) -> str:
    """Render a filesystem path as a ``file://`` URI."""
    return f"file://{path}"


@traces(spec="065", reqs=["REQ-065-012"])
def _ensure_python_file(path: Path) -> None:
    """Raise ``UNSUPPORTED_LANGUAGE`` (ERR-011) for any non-``.py`` target."""
    if path.suffix != ".py":
        raise LSPError(
            LSPErrorCode.UNSUPPORTED_LANGUAGE,
            f"only Python (.py) targets are supported; got {path.suffix or '<none>'!r}",
        )


@traces(spec="065", reqs=["REQ-065-013"])
def _read_source(path: Path) -> str:
    """Read ``path`` as text, raising ``FILE_READ_ERROR`` (ERR-012) on failure."""
    try:
        return path.read_text()
    except OSError as exc:
        raise LSPError(
            LSPErrorCode.FILE_READ_ERROR, f"cannot read {path}: {exc}"
        ) from exc


@traces(spec="065", story="US-065-003", reqs=["REQ-065-002"])
def _iter_python_files(root: Path) -> list[Path]:
    """Every ``.py`` file under ``root`` (or ``[root]`` if it's a file)."""
    if root.is_file():
        return [root]
    if not root.is_dir():
        return []
    return sorted(
        p
        for p in root.rglob("*.py")
        if "__pycache__" not in p.parts
        and not any(part.startswith(".") for part in p.relative_to(root).parts)
    )


# ── signature helpers (self-contained; no dependency on scripts/*) ────────


@traces(spec="065", story="US-065-004", reqs=["REQ-065-004"])
def _func_signature(node: ast.FunctionDef | ast.AsyncFunctionDef) -> str:
    """Render a Python function/method signature from its AST node."""
    try:
        args = ast.unparse(node.args)
    except ValueError, TypeError:
        args = "..."
    ret = ""
    if node.returns is not None:
        try:
            ret = f" -> {ast.unparse(node.returns)}"
        except ValueError, TypeError:
            ret = ""
    prefix = "async def " if isinstance(node, ast.AsyncFunctionDef) else "def "
    return f"{prefix}{node.name}({args}){ret}"


@traces(spec="065", story="US-065-004", reqs=["REQ-065-004"])
def _class_signature(node: ast.ClassDef) -> str:
    """Render a Python class signature (name + base list) from its AST node."""
    bases = []
    for base in node.bases:
        try:
            bases.append(ast.unparse(base))
        except ValueError, TypeError:
            continue
    inner = f"({', '.join(bases)})" if bases else ""
    return f"class {node.name}{inner}"


@traces(spec="065", story="US-065-003", reqs=["REQ-065-001", "REQ-065-002"])
def _name_token_span(line_text: str, name: str, keyword: str) -> tuple[int, int] | None:
    """Column span of ``name`` immediately after ``keyword`` on ``line_text``."""
    match = re.search(rf"\b{keyword}\s+({re.escape(name)})\b", line_text)
    if not match:
        return None
    return match.start(1), match.end(1)


@traces(spec="065", story="US-065-003", reqs=["REQ-065-001", "REQ-065-002"])
def _def_location(
    path: Path,
    lines: list[str],
    node: ast.ClassDef | ast.FunctionDef | ast.AsyncFunctionDef,
    name: str,
) -> LSPLocation:
    """Build an :class:`LSPLocation` for a ``ClassDef``/``FunctionDef`` name token.

    Falls back to the node's own ``col_offset`` (pointing at the ``class``/
    ``def`` keyword rather than the name) if the name can't be found via
    regex on its source line — e.g. unusual formatting.
    """
    lineno = node.lineno - 1
    keyword = "class" if isinstance(node, ast.ClassDef) else "def"
    line_text = lines[lineno] if 0 <= lineno < len(lines) else ""
    span = _name_token_span(line_text, name, keyword)
    if span is None:
        start_col = node.col_offset
        end_col = start_col + len(name)
    else:
        start_col, end_col = span
    return LSPLocation(
        uri=_path_to_uri(path),
        range=Range(
            start=Position(line=lineno, character=start_col),
            end=Position(line=lineno, character=end_col),
        ),
    )


@traces(spec="065", story="US-065-003", reqs=["REQ-065-001", "REQ-065-002"])
def _name_node_location(path: Path, node: ast.Name) -> LSPLocation:
    """Build an :class:`LSPLocation` for an ``ast.Name`` reference node."""
    return LSPLocation(
        uri=_path_to_uri(path),
        range=Range(
            start=Position(line=node.lineno - 1, character=node.col_offset),
            end=Position(
                line=(node.end_lineno or node.lineno) - 1,
                character=(
                    node.end_col_offset
                    if node.end_col_offset is not None
                    else node.col_offset + len(node.id)
                ),
            ),
        ),
    )


@traces(spec="065", story="US-065-003", reqs=["REQ-065-002"])
def _attribute_node_location(path: Path, node: ast.Attribute) -> LSPLocation:
    """Build an :class:`LSPLocation` for the ``.attr`` token of an Attribute node.

    ``ast.Attribute.end_col_offset`` marks the end of the whole ``a.b.c``
    expression, which — for the common single-line case — is exactly the end
    of the trailing attribute name; the attribute's own start column is
    derived by subtracting its length. Multi-line attribute chains are a
    known imprecision of this heuristic (documented MVP trade-off).
    """
    end_line = (node.end_lineno or node.lineno) - 1
    end_col = (
        node.end_col_offset
        if node.end_col_offset is not None
        else node.col_offset + len(node.attr)
    )
    start_col = max(0, end_col - len(node.attr))
    return LSPLocation(
        uri=_path_to_uri(path),
        range=Range(
            start=Position(line=end_line, character=start_col),
            end=Position(line=end_line, character=end_col),
        ),
    )


# ── AST-based navigation backend (definition / references / symbols / hover) ──


@traces(
    spec="065",
    story="US-065-003",
    reqs=["REQ-065-001", "REQ-065-002", "REQ-065-003", "REQ-065-004"],
)
class PythonSymbolIndex:
    """Token-precise, name-resolution navigation over Python source (MVP backend).

    See the module docstring for the full backend-choice rationale: this is
    NOT a type-resolving language server — it walks the target file(s)' AST
    and matches identifiers by NAME, the same approach as this repo's
    existing ``scripts/serialize_codebase_ast.py`` navigation helpers,
    reimplemented locally so this tool has no dependency on that script.
    """

    @traces(spec="065", story="US-065-003", reqs=["REQ-065-001"])
    def identifier_at(self, source: str, position: Position) -> str | None:
        """Return the identifier token spanning ``position`` in ``source``.

        Returns ``None`` when the position is out of range or does not fall
        on a Python identifier (``[A-Za-z_][A-Za-z0-9_]*``).
        """
        lines = source.splitlines()
        if not (0 <= position.line < len(lines)):
            return None
        line_text = lines[position.line]
        col = position.character
        if not (0 <= col <= len(line_text)):
            return None
        for match in re.finditer(r"[A-Za-z_][A-Za-z0-9_]*", line_text):
            if match.start() <= col <= match.end():
                return match.group(0)
        return None

    @traces(spec="065", story="US-065-003", reqs=["REQ-065-001"])
    def find_definition(
        self, name: str, search_paths: list[Path]
    ) -> LSPLocation | None:
        """First ``ClassDef``/``FunctionDef``/module-level assignment named ``name``.

        Searches ``search_paths`` in order and returns the first match; the
        target must be a Python file (see :func:`_ensure_python_file`).
        """
        for path in search_paths:
            try:
                source = _read_source(path)
                tree = ast.parse(source, filename=str(path))
            except LSPError, SyntaxError:
                continue
            lines = source.splitlines()
            for node in ast.walk(tree):
                if (
                    isinstance(
                        node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)
                    )
                    and node.name == name
                ):
                    return _def_location(path, lines, node, name)
                if isinstance(node, ast.Assign):
                    for target in node.targets:
                        if isinstance(target, ast.Name) and target.id == name:
                            return _name_node_location(path, target)
        return None

    @traces(spec="065", story="US-065-003", reqs=["REQ-065-002"])
    def find_references(self, name: str, search_paths: list[Path]) -> list[LSPLocation]:
        """Every ``Name``/``Attribute``/def-site use of ``name`` across ``search_paths``."""
        locations: list[LSPLocation] = []
        for path in search_paths:
            try:
                source = _read_source(path)
                tree = ast.parse(source, filename=str(path))
            except LSPError, SyntaxError:
                continue
            lines = source.splitlines()
            for node in ast.walk(tree):
                if isinstance(node, ast.Name) and node.id == name:
                    locations.append(_name_node_location(path, node))
                elif isinstance(node, ast.Attribute) and node.attr == name:
                    locations.append(_attribute_node_location(path, node))
                elif (
                    isinstance(
                        node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)
                    )
                    and node.name == name
                ):
                    locations.append(_def_location(path, lines, node, name))
        return locations

    @traces(spec="065", story="US-065-004", reqs=["REQ-065-003"])
    def document_symbols(self, path: Path) -> list[DocumentSymbolInfo]:
        """List every top-level class/function and method in ``path`` (FR-005)."""
        source = _read_source(path)
        tree = ast.parse(source, filename=str(path))
        symbols: list[DocumentSymbolInfo] = []

        def _visit(node: ast.AST, class_name: str | None) -> None:
            for child in ast.iter_child_nodes(node):
                if isinstance(child, ast.ClassDef):
                    symbols.append(_symbol_of(path, child, "class", child.name))
                    _visit(child, child.name)
                elif isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
                    qual = f"{class_name}.{child.name}" if class_name else child.name
                    kind = "method" if class_name else "function"
                    symbols.append(_symbol_of(path, child, kind, qual))
                    _visit(child, class_name)
                else:
                    _visit(child, class_name)

        _visit(tree, None)
        return symbols

    @traces(spec="065", story="US-065-004", reqs=["REQ-065-004"])
    def hover(self, name: str, search_paths: list[Path]) -> LSPHover | None:
        """Signature + docstring for the ``ClassDef``/``FunctionDef`` named ``name``.

        A lightweight subset of spec.md's hover content — no live type
        inference (no jedi/pyright-langserver session), so only what the AST
        itself carries: the declared signature and its docstring.
        """
        for path in search_paths:
            try:
                source = _read_source(path)
                tree = ast.parse(source, filename=str(path))
            except LSPError, SyntaxError:
                continue
            lines = source.splitlines()
            for node in ast.walk(tree):
                if (
                    isinstance(
                        node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)
                    )
                    and node.name == name
                ):
                    if isinstance(node, ast.ClassDef):
                        signature = _class_signature(node)
                    else:
                        signature = _func_signature(node)
                    doc = ast.get_docstring(node) or ""
                    contents = f"```python\n{signature}\n```"
                    if doc:
                        contents += f"\n\n{doc}"
                    location = _def_location(path, lines, node, name)
                    return LSPHover(contents=contents, range=location.range)
        return None


@traces(spec="065", story="US-065-003", reqs=["REQ-065-003"])
def _symbol_of(path: Path, node: ast.AST, kind: str, name: str) -> DocumentSymbolInfo:
    """Build a :class:`DocumentSymbolInfo` for a class/function/method AST node."""
    lineno = getattr(node, "lineno", 1) - 1
    end_lineno = getattr(node, "end_lineno", None) or (lineno + 1)
    return DocumentSymbolInfo(
        name=name,
        kind=kind,
        range=Range(
            start=Position(line=lineno, character=getattr(node, "col_offset", 0)),
            end=Position(
                line=end_lineno - 1, character=getattr(node, "end_col_offset", 0) or 0
            ),
        ),
    )


# ── diagnostics backend (real pyright CLI, batch mode) ─────────────────────


@traces(spec="065", story="US-065-002", reqs=["REQ-065-006", "REQ-065-007"])
class PyrightDiagnosticsBackend:
    """Wraps the real ``pyright`` CLI (a declared dev dependency) for diagnostics.

    Runs ``pyright <path> --outputjson`` as a one-shot subprocess — exactly
    the invocation ``scripts/pyright_ratchet.py`` already uses for its gate —
    NOT a persistent ``pyright-langserver --stdio`` JSON-RPC session. See the
    module docstring for the full backend-choice rationale.
    """

    def __init__(self, command: str = "pyright") -> None:
        self._command = command

    @traces(spec="065", reqs=["REQ-065-014"])
    def available(self) -> bool:
        """Whether the ``pyright`` executable can be found on ``PATH`` (FR-SEC-001)."""
        return shutil.which(self._command) is not None

    @traces(
        spec="065",
        story="US-065-002",
        reqs=["REQ-065-006", "REQ-065-008", "REQ-065-009", "REQ-065-011"],
    )
    def diagnostics(self, path: Path, timeout: float = 30.0) -> LSPDiagnostics:
        """Run pyright against ``path`` and return its diagnostics (US-002).

        Raises:
            LSPError: ``SERVER_NOT_FOUND`` (ERR-001) if pyright isn't on
                PATH, ``REQUEST_TIMEOUT`` (ERR-004) if the subprocess exceeds
                ``timeout``, or ``SERVER_ERROR`` (ERR-007) if its stdout
                isn't parseable JSON.
        """
        if not self.available():
            raise LSPError(
                LSPErrorCode.SERVER_NOT_FOUND,
                f"{self._command!r} not found on PATH; install the 'pyright' "
                "dev dependency (uv sync --dev)",
            )
        if not path.exists():
            raise LSPError(LSPErrorCode.FILE_READ_ERROR, f"file not found: {path}")
        try:
            proc = subprocess.run(
                [self._command, str(path), "--outputjson"],
                capture_output=True,
                text=True,
                timeout=timeout,
                check=False,
            )
        except subprocess.TimeoutExpired as exc:
            raise LSPError(
                LSPErrorCode.REQUEST_TIMEOUT,
                f"pyright timed out after {timeout}s on {path}",
            ) from exc
        try:
            report = json.loads(proc.stdout)
        except json.JSONDecodeError as exc:
            raise LSPError(
                LSPErrorCode.SERVER_ERROR,
                f"could not parse pyright --outputjson output: {exc}",
            ) from exc
        items = []
        for entry in report.get("generalDiagnostics", []):
            rng = entry.get("range") or {}
            start = rng.get("start") or {}
            end = rng.get("end") or {}
            items.append(
                LSPDiagnostic(
                    range=Range(
                        start=Position(
                            line=start.get("line", 0),
                            character=start.get("character", 0),
                        ),
                        end=Position(
                            line=end.get("line", 0),
                            character=end.get("character", 0),
                        ),
                    ),
                    message=entry.get("message", ""),
                    severity=_PYRIGHT_SEVERITY_MAP.get(entry.get("severity"), 1),
                    code=str(entry["rule"]) if entry.get("rule") else None,
                    source="pyright",
                )
            )
        return LSPDiagnostics(uri=_path_to_uri(path), diagnostics=items)


# ── argument parsing ────────────────────────────────────────────────────


def _parse_position(raw: Any) -> Position:
    """Parse a ``{"line": int, "character": int}`` mapping into a :class:`Position`."""
    if not isinstance(raw, dict):
        raise LSPError(
            LSPErrorCode.INVALID_REQUEST, "arguments['position'] is required"
        )
    try:
        return Position(line=int(raw["line"]), character=int(raw["character"]))
    except (KeyError, TypeError, ValueError) as exc:
        raise LSPError(
            LSPErrorCode.INVALID_REQUEST,
            f"malformed position (need integer 'line'/'character'): {exc}",
        ) from exc


@traces(spec="065", story="US-065-003", reqs=["REQ-065-002"])
def _resolve_search_paths(arguments: dict[str, Any], default: Path) -> list[Path]:
    """Resolve ``arguments['search_paths']`` into a flat, deduplicated file list.

    Defaults to ``[default]`` (the target document only) when omitted —
    references/definition then only search within the single document,
    matching a minimal single-file navigation request.
    """
    raw_paths = arguments.get("search_paths")
    if not raw_paths:
        return [default]
    if not isinstance(raw_paths, list):
        raise LSPError(
            LSPErrorCode.INVALID_REQUEST, "arguments['search_paths'] must be a list"
        )
    seen: list[Path] = []
    for raw in raw_paths:
        root = _uri_to_path(str(raw)) if isinstance(raw, str) else None
        if root is None:
            continue
        for candidate in _iter_python_files(root):
            if candidate not in seen:
                seen.append(candidate)
    return seen or [default]


# ── the tool itself ─────────────────────────────────────────────────────

_LSP_TOOL_PARAMETERS: dict[str, Any] = {
    "type": "object",
    "properties": {
        "action": {
            "type": "string",
            "enum": sorted(_VALID_ACTIONS),
            "description": "The navigation operation to perform.",
        },
        "uri": {
            "type": "string",
            "description": "file:// URI or filesystem path to the target Python file.",
        },
        "position": {
            "type": "object",
            "properties": {
                "line": {"type": "integer", "minimum": 0},
                "character": {"type": "integer", "minimum": 0},
            },
            "required": ["line", "character"],
            "description": (
                "Zero-based cursor position; required for definition/"
                "references/hover, ignored for diagnostics/document_symbols."
            ),
        },
        "search_paths": {
            "type": "array",
            "items": {"type": "string"},
            "description": (
                "Additional files/directories to search for definition/"
                "references; defaults to the target file only."
            ),
        },
    },
    "required": ["action", "uri"],
}


@traces(
    spec="065",
    story="US-065-003",
    reqs=["REQ-065-016"],
)
class LSPTool(BaseTool):
    """LSPTool — code navigation + diagnostics for Python targets (SPEC-065).

    Composes a :class:`PythonSymbolIndex` (definition/references/document
    symbols/hover — the AST-based MVP navigation backend) and a
    :class:`PyrightDiagnosticsBackend` (real ``pyright`` CLI diagnostics).
    See the module docstring for the full backend-choice rationale and the
    explicit list of spec.md capabilities NOT implemented in this MVP.
    """

    def __init__(
        self,
        symbol_index: PythonSymbolIndex | None = None,
        diagnostics_backend: PyrightDiagnosticsBackend | None = None,
    ) -> None:
        self._symbols = symbol_index or PythonSymbolIndex()
        self._diagnostics = diagnostics_backend or PyrightDiagnosticsBackend()
        self._metadata = ToolMetadata(
            name="lsp",
            description=(
                "Code navigation (go-to-definition, find-references, "
                "document symbols, hover) and pyright diagnostics for "
                "Python files."
            ),
            category=ToolCategory.SEARCH,
            parameters=_LSP_TOOL_PARAMETERS,
            priority=ToolPriority.P0,
            tags=("lsp", "navigation", "diagnostics", "python"),
        )

    @property
    def metadata(self) -> ToolMetadata:
        return self._metadata

    @traces(
        spec="065",
        story="US-065-003",
        reqs=[
            "REQ-065-001",
            "REQ-065-002",
            "REQ-065-003",
            "REQ-065-004",
            "REQ-065-006",
            "REQ-065-010",
            "REQ-065-012",
            "REQ-065-013",
        ],
    )
    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        """Dispatch one of ``definition``/``references``/``diagnostics``/``hover``/
        ``document_symbols`` from raw tool-call ``arguments``.

        Never raises: any :class:`LSPError` (or unforeseen exception)
        becomes a failed :class:`ToolResult` instead.
        """
        try:
            action = arguments.get("action")
            if action not in _VALID_ACTIONS:
                raise LSPError(
                    LSPErrorCode.INVALID_REQUEST,
                    f"unknown action {action!r}; valid: {sorted(_VALID_ACTIONS)}",
                )
            uri = arguments.get("uri")
            if not isinstance(uri, str) or not uri.strip():
                raise LSPError(LSPErrorCode.INVALID_URI, "arguments['uri'] is required")
            path = _uri_to_path(uri)

            if action == "diagnostics":
                result = self._diagnostics.diagnostics(path)
                return ToolResult(
                    tool_name=self.metadata.name,
                    success=True,
                    output=json.dumps(result.to_dict()),
                )

            _ensure_python_file(path)

            if action == "document_symbols":
                symbols = self._symbols.document_symbols(path)
                return ToolResult(
                    tool_name=self.metadata.name,
                    success=True,
                    output=json.dumps([s.to_dict() for s in symbols]),
                )

            position = _parse_position(arguments.get("position"))
            source = _read_source(path)
            name = self._symbols.identifier_at(source, position)
            if name is None:
                raise LSPError(
                    LSPErrorCode.SYMBOL_NOT_FOUND,
                    f"no identifier at line {position.line}, character {position.character}",
                )
            search_paths = _resolve_search_paths(arguments, path)

            if action == "definition":
                location = self._symbols.find_definition(name, search_paths)
                output = json.dumps(location.to_dict() if location else None)
                return ToolResult(
                    tool_name=self.metadata.name, success=True, output=output
                )

            if action == "references":
                locations = self._symbols.find_references(name, search_paths)
                output = json.dumps([loc.to_dict() for loc in locations])
                return ToolResult(
                    tool_name=self.metadata.name, success=True, output=output
                )

            # action == "hover"
            hover = self._symbols.hover(name, search_paths)
            output = json.dumps(hover.to_dict() if hover else None)
            return ToolResult(tool_name=self.metadata.name, success=True, output=output)
        except LSPError as exc:
            return exc.to_result(self.metadata.name)
        except Exception as exc:  # noqa: BLE001 - never propagate; report as ToolResult
            return ToolResult(
                tool_name=self.metadata.name, success=False, output="", error=str(exc)
            )