Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
227 changes: 227 additions & 0 deletions src/vtk_prompt/completion.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,10 @@

from __future__ import annotations

import re
from functools import lru_cache
from types import ModuleType

from . import get_logger

logger = get_logger(__name__)
Expand Down Expand Up @@ -84,6 +88,15 @@ def complete_python(code: str, line: int, column: int, limit: int = 500) -> list
"""
if not _JEDI_OK or not isinstance(code, str):
return []

# Cheap path first, same reasoning as hover: a resolvable VTK receiver is
# answered from docstrings without waiting on jedi's inference.
cls = _resolve_receiver(code, line, column, allow_jedi=False)
if cls is not None:
members = _members_of(cls, limit)
if members:
return members

try:
completions = jedi.Interpreter(code, [_NS]).complete(line, column)
except Exception as exc: # jedi can raise on malformed/partial input
Expand All @@ -97,6 +110,13 @@ def complete_python(code: str, line: int, column: int, limit: int = 500) -> list
except Exception:
detail = ""
out.append({"label": c.name, "kind": c.type, "detail": detail})

if not out:
# jedi gives nothing after a VTK call ("GetPointIds()."), so retry with
# jedi allowed to infer the chain's root.
cls = _resolve_receiver(code, line, column, allow_jedi=True)
if cls is not None:
return _members_of(cls, limit)
return out


Expand All @@ -111,6 +131,195 @@ def _doc_prose(doc: str, name: str) -> str:
return "\n".join(kept).strip()


# --------------------------------------------------------------------------
# Docstring-based return-type resolution
#
# jedi cannot infer through VTK's C-extension methods. It *shows* the return
# type because it parses the docstring, but it does not use it for inference, so
# any chained expression dies at the first call: `tetra.GetPointIds().SetId` has
# no completions and no hover, even though `tetra.GetPointIds` resolves fine.
#
# VTK's docstrings do carry the type ("GetPointIds(self) -> vtkIdList"), so the
# chain can be walked manually: resolve the root with jedi, then step through
# each `.Method()` by reading the annotation and looking the class up in the vtk
# namespace. Only VTK object returns resolve, which is the intent - `-> int` or
# `-> None` correctly yields nothing to chain from.
# --------------------------------------------------------------------------

# "-> vtkIdList", "-> ('vtkIdList', ...)", "-> None"
_RETURN_ANNOTATION = re.compile(r"->\s*\(?\s*'?([A-Za-z_][\w.]*)'?")
# Trailing `name.Method().Method()` chain immediately left of the cursor. The
# final identifier is optional: completion fires straight after the "." with
# nothing typed, while hover sits inside a name that is already there.
_CHAIN = re.compile(r"([A-Za-z_]\w*)\s*((?:\.\s*[A-Za-z_]\w*\s*\([^()]*\)\s*)+)\.\s*\w*$")
_CHAIN_STEP = re.compile(r"\.\s*([A-Za-z_]\w*)\s*\([^()]*\)")
# `name.token` with no intervening call - the direct-attribute case.
_DIRECT = re.compile(r"([A-Za-z_]\w*)\s*\.\s*\w*$")


def _assigned_vtk_class(code: str, root: str) -> type | None:
"""Class from a literal ``root = vtk.vtkFoo()`` assignment in the source.

Covers the dominant shape of generated VTK scripts without paying for jedi
inference, which costs hundreds of milliseconds even with a warm cache.
"""
m = re.search(
rf"^\s*{re.escape(root)}\s*=\s*(?:[\w.]*\.)?(vtk[A-Za-z0-9_]*)\s*\(",
code,
re.MULTILINE,
)
return _vtk_class(m.group(1)) if m else None


def _live_class(root: str) -> type | None:
"""Class of an injected runtime object (``renderer``, ``render_window``).

Modules and classes are excluded: ``vtk`` is in the namespace too, and
``type()`` of a module is useless here.
"""
live = _NS.get(root)
if live is None or isinstance(live, (type, ModuleType)):
return None
cls = type(live)
return cls if cls.__name__.startswith("vtk") else None


@lru_cache(maxsize=4096)
def _return_class(cls: type, method: str) -> type | None:
"""Class returned by ``cls.method()`` per its docstring, or None.

Cached because hover fires on mouse movement; the parsing is cheap but not
free at that rate.
"""
func = getattr(cls, method, None)
doc = getattr(func, "__doc__", "") or ""
for line in doc.splitlines():
if not line.strip().startswith(method + "("):
continue
m = _RETURN_ANNOTATION.search(line)
if not m:
continue
name = m.group(1).split(".")[-1]
if not name.startswith("vtk"):
return None # int/float/None etc: nothing to chain from
return _vtk_class(name)
return None


def _vtk_class(name: str) -> type | None:
"""Look a VTK class up by bare name in the seeded namespace."""
for container in (_NS.get("vtk"), _NS.get("vtkmodules")):
obj = getattr(container, name, None)
if isinstance(obj, type):
return obj
obj = _NS.get(name)
return obj if isinstance(obj, type) else None


def _root_class(code: str, line: int, column: int, root: str, allow_jedi: bool) -> type | None:
"""Class of the chain's root variable, cheapest source first.

Injected runtime objects and literal ``x = vtk.vtkFoo()`` assignments are
resolved without jedi, which is the whole point: jedi's inference is what
makes hover slow enough for the editor to give up waiting.
"""
cls = _live_class(root) or _assigned_vtk_class(code, root)
if cls is not None or not allow_jedi or not _JEDI_OK:
return cls
try:
for d in jedi.Interpreter(code, [_NS]).infer(line, column):
obj = _vtk_class(d.name)
if obj is not None:
return obj
except Exception as exc:
logger.debug("jedi root inference error: %s", exc)
return None


def _resolve_receiver(code: str, line: int, column: int, allow_jedi: bool = True) -> type | None:
"""Class of the expression immediately left of the cursor, or None.

Handles both ``tetra.GetPointIds().SetId`` (walking the chain by docstring
return type) and plain ``renderer.AddActor``.
"""
lines = code.splitlines()
if line < 1 or line > len(lines):
return None
text = lines[line - 1]
prefix = text[: column + 1]
# Include the identifier the cursor sits inside, not just what precedes it.
trailing = re.match(r"\w*", text[column + 1 :])
prefix += trailing.group(0) if trailing else ""

m = _CHAIN.search(prefix)
steps_text = m.group(2) if m else ""
if m is None:
m = _DIRECT.search(prefix)
if m is None:
return None
root = m.group(1)
# Column of the root token, so jedi infers the variable and not the chain.
cls = _root_class(code, line, prefix.index(root) + len(root), root, allow_jedi)
if cls is None:
return None
for step in _CHAIN_STEP.findall(steps_text):
cls = _return_class(cls, step)
if cls is None:
return None
return cls


def _members_of(cls: type, limit: int) -> list[dict]:
"""Completion candidates for a resolved class, in jedi's output shape."""
out: list[dict] = []
for name in dir(cls):
if name.startswith("_"):
continue
try:
attr = getattr(cls, name, None)
doc = (getattr(attr, "__doc__", "") or "").splitlines()
detail = next((ln.strip() for ln in doc if ln.strip()), "")[:80]
kind = "function" if callable(attr) else "instance"
except Exception:
detail, kind = "", "instance"
out.append({"label": name, "kind": kind, "detail": detail})
if len(out) >= limit:
break
return out


def _hover_from_class(cls: type, name: str) -> dict | None:
"""Hover payload for ``cls.name``, matching hover_python's shape."""
attr = getattr(cls, name, None)
if attr is None:
return None
doc = getattr(attr, "__doc__", "") or ""
if not doc:
return None
signatures = [ln.strip() for ln in doc.splitlines() if ln.strip().startswith(name + "(")]
return {
"name": name,
"type": "function" if callable(attr) else "instance",
"signatures": signatures,
"prose": _doc_prose(doc, name),
}


def _token_at(code: str, line: int, column: int) -> str:
"""Return the identifier the cursor sits in or next to."""
lines = code.splitlines()
if line < 1 or line > len(lines):
return ""
text = lines[line - 1]
start = column
while start > 0 and (text[start - 1].isalnum() or text[start - 1] == "_"):
start -= 1
end = column
while end < len(text) and (text[end].isalnum() or text[end] == "_"):
end += 1
return text[start:end]


def hover_python(code: str, line: int, column: int) -> dict | None:
"""Return hover info (signature + docstring) for the symbol at the cursor.

Expand All @@ -120,12 +329,30 @@ def hover_python(code: str, line: int, column: int) -> dict | None:
"""
if not _JEDI_OK or not isinstance(code, str):
return None

# Try the docstring route before jedi. For a VTK receiver it answers in
# ~1ms where jedi's help() takes hundreds of milliseconds warm and seconds
# cold - long enough that the editor cancels the request and shows nothing,
# which looked like "hover is broken" rather than "hover is slow".
token = _token_at(code, line, column)
if token:
cls = _resolve_receiver(code, line, column, allow_jedi=False)
if cls is not None:
info = _hover_from_class(cls, token)
if info is not None:
return info

try:
defs = jedi.Interpreter(code, [_NS]).help(line, column)
except Exception as exc:
logger.debug("jedi hover error: %s", exc)
return None
if not defs:
# Nothing from jedi either: retry the chain with jedi allowed to infer
# the root, for receivers the cheap paths cannot see.
cls = _resolve_receiver(code, line, column, allow_jedi=True)
if cls is not None and token:
return _hover_from_class(cls, token)
return None
d = defs[0]
try:
Expand Down
98 changes: 95 additions & 3 deletions src/vtk_prompt/controllers/configuration.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,20 +9,112 @@

import yaml

from .. import get_logger
from ..provider_utils import DEFAULT_PROVIDER, get_default_model, supports_temperature

logger = get_logger(__name__)

# Settings that belong to the install rather than to a conversation, and so are
# written back to the config file. Everything else in that file is left alone:
# `model` and `base_url` follow the active conversation now, and `retries` /
# `modelParameters` are per-conversation too, so writing them here would let
# whichever conversation happened to be open redefine the startup defaults.
_GLOBAL_SETTING_KEYS = (
"mcp_url",
"top_k",
"log_tool_calls",
"agentic_retrieval",
"data_root",
)


def _global_settings_path():
"""Return the config file to write global settings to.

Prefers whichever file was discovered at startup so edits land back in the
file they came from, and falls back to the per-user location on a fresh
install where no config exists yet.
"""
from pathlib import Path

from ..utils.env_config import _config_home, discover_config_file

found = discover_config_file()
if found:
return Path(found)
return _config_home() / "config.yml"


def persist_global_settings(app: Any) -> None:
"""Write the install-wide settings back to the config file.

Read-modify-write rather than a full dump: the file is hand-maintained and
holds keys this app never sets (``name``, ``description``), so only the keys
in ``_GLOBAL_SETTING_KEYS`` are touched and existing order is preserved.
"""
path = _global_settings_path()
try:
existing = {}
if path.is_file():
loaded = yaml.safe_load(path.read_text())
if isinstance(loaded, dict):
existing = loaded

updated = dict(existing)
for key in _GLOBAL_SETTING_KEYS:
if not hasattr(app.state, key):
continue
value = getattr(app.state, key)
if key == "top_k":
value = int(value)
elif key in ("log_tool_calls", "agentic_retrieval"):
value = bool(value)
else:
value = (value or "").strip()
if not value and key not in existing:
continue # don't add empty keys the file never had
updated[key] = value

if updated == existing:
return # nothing changed; leave the file's mtime alone

path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(yaml.safe_dump(updated, sort_keys=False))
logger.info("Saved global settings to %s", path)
except (OSError, yaml.YAMLError, TypeError, ValueError) as e:
# Never let a settings write break closing the dialog.
logger.warning("Could not save global settings to %s: %s", path, e)


def on_tab_change(app: Any, tab_index: int, **_: Any) -> None:
"""Handle tab change to sync use_cloud_models state."""
app.state.use_cloud_models = tab_index == 0


def on_model_change(app: Any, **_: Any) -> None:
"""Handle model change to update temperature support."""
"""Sync temperature support for the selected model.

Models that ignore temperature need it pinned to 1, but that used to
overwrite the user's value with no way back: switching to o3 and back left
the conversation at 1 forever. The pre-clamp value is stashed instead, and
restored on the way out.
"""
current_model = app._get_model()
app.state.temperature_supported = supports_temperature(current_model)
if not app.state.temperature_supported:
supported = supports_temperature(current_model)
app.state.temperature_supported = supported

if not supported:
# Stash only on the way in, or a second model change while clamped
# would overwrite the stash with the clamped value itself.
if not str(getattr(app.state, "temperature_pref", "") or ""):
app.state.temperature_pref = str(app.state.temperature)
app.state.temperature = 1
return

stashed = str(getattr(app.state, "temperature_pref", "") or "")
if stashed:
app.state.temperature = stashed
app.state.temperature_pref = ""


def on_provider_change(app: Any, provider: str, **kwargs: Any) -> None:
Expand Down
Loading
Loading