"""Tool-registry catalog generator for the interactive shell. The OpenSRE tool registry (:mod:`tools.registry`) auto-discovers tools under ``tools/`` and exposes them via :func:`get_registered_tools`. Each :class:`~tools.registered_tool.RegisteredTool` carries rich metadata — ``name``, ``description``, ``surfaces``, ``input_schema``, ``source``, and the module it was discovered in — but none of this is currently visible to the interactive-shell user. This module turns the registry snapshot into a compact catalog suitable for two independent consumers: 1. The ``/tools`` slash command — users can see what tools are wired into their build of the shell. 2. Future codebase-aware grounding — the catalog text can be injected into the LLM prompt so the assistant can answer "what tools can the chat agent call?" accurately. (Wiring lives in a separate later issue; ``format_tool_catalog_text`` is shaped to support both surfaces today.) Two pure functions: - :func:`build_tool_catalog` — wraps :func:`get_registered_tools` and returns :class:`ToolCatalogEntry` records with the fields needed for display and prompt injection. - :func:`format_tool_catalog_text` — renders entries as compact Markdown-ish text grouped by surface, suitable for both terminal display and LLM grounding. """ from __future__ import annotations import importlib from dataclasses import dataclass from pathlib import Path from typing import Any from config.constants.paths import REPO_ROOT from core.domain.types.tools import ToolSurface from core.tool_framework.registered_tool import RegisteredTool from tools.registry import get_registered_tools # Cap one-line schema summaries so wide registries don't break terminal # wrapping or balloon prompt injection. The cap is generous — tools with # many params get a trailing ellipsis rather than an unwieldy line. _MAX_SCHEMA_SUMMARY_CHARS = 200 @dataclass(frozen=True) class ToolCatalogEntry: """Display-shaped projection of a :class:`RegisteredTool`.""" name: str """Registered tool name, e.g. ``"search_github_code"``.""" surfaces: tuple[str, ...] """Surfaces the tool is exposed on (``"investigation"`` and/or ``"chat"``).""" description: str """One-line description from the tool's metadata, trimmed for display.""" source_file: str """Repo-relative path (forward slashes) when ``__file__`` lives under the checkout root, otherwise the resolved absolute POSIX path when the defining module is outside the repo (e.g. an installed editable or site-packages shim). Empty string only when ``origin_module`` is missing, import fails, or the loaded module exposes no ``__file__``.""" input_schema_summary: str """One-line render of top-level params, e.g. ``"query: string, limit?: integer"``. ``"(no params)"`` when the tool takes no inputs.""" def _resolve_source_file(tool: RegisteredTool) -> str: """Best-effort relative path to the tool's defining file. The tool registry stores ``origin_module`` (a dotted module path); we import it to read ``__file__``. Failures here are non-fatal — the catalog still surfaces the tool, just without a source pointer. """ module_name = tool.origin_module if not module_name: return "" try: module = importlib.import_module(module_name) except Exception: return "" file_attr = getattr(module, "__file__", None) if not file_attr: return "" try: return Path(file_attr).resolve().relative_to(REPO_ROOT).as_posix() except ValueError: # Module lives outside the repo root (installed package, namespace # package). Fall back to the absolute path so users can still open # it; ``ValueError`` here means ``relative_to`` couldn't compute a # relative path, not that the file is missing. return Path(file_attr).as_posix() def _summarize_input_schema(input_schema: dict[str, Any]) -> str: """Render top-level params as a one-line ``name: type`` list. Required params render as ``name: type``; optional params get a ``?`` suffix (``name?: type``). Untyped properties (no ``type`` key) render as ``any`` so the user can see the param exists. Returns ``"(no params)"`` for empty schemas. """ properties = input_schema.get("properties") or {} if not properties: return "(no params)" required = set(input_schema.get("required") or ()) parts: list[str] = [] for name, info in properties.items(): info_dict = info if isinstance(info, dict) else {} type_label = str(info_dict.get("type") or "any") suffix = "" if name in required else "?" parts.append(f"{name}{suffix}: {type_label}") rendered = ", ".join(parts) if len(rendered) > _MAX_SCHEMA_SUMMARY_CHARS: return rendered[: _MAX_SCHEMA_SUMMARY_CHARS - 1].rstrip(", ") + "…" return rendered def _entry_from_tool(tool: RegisteredTool) -> ToolCatalogEntry: return ToolCatalogEntry( name=tool.name, surfaces=tuple(tool.surfaces), description=(tool.description or "").strip(), source_file=_resolve_source_file(tool), input_schema_summary=_summarize_input_schema(tool.input_schema), ) def build_tool_catalog(surface: ToolSurface | None = None) -> list[ToolCatalogEntry]: """Return :class:`ToolCatalogEntry` records for tools registered on ``surface``. Pass ``surface=None`` (default) to get every registered tool, or ``"investigation"`` / ``"chat"`` to filter. The order matches :func:`get_registered_tools` (alphabetical by tool name). """ return [_entry_from_tool(tool) for tool in get_registered_tools(surface)] def _surface_sort_key(surface: str) -> tuple[int, str]: """Stable surface ordering — investigation first, chat second, others alphabetical.""" priority = {"investigation": 0, "chat": 1} return (priority.get(surface, 99), surface) def format_tool_catalog_text(entries: list[ToolCatalogEntry]) -> str: """Render entries as compact Markdown-ish text grouped by surface. Each tool may belong to multiple surfaces and will appear under each one so the user can see at a glance which tools the chat agent versus the investigation pipeline can reach. Returns ``""`` for an empty catalog. """ if not entries: return "" by_surface: dict[str, list[ToolCatalogEntry]] = {} for entry in entries: for surface in entry.surfaces: by_surface.setdefault(surface, []).append(entry) lines: list[str] = [] for surface in sorted(by_surface.keys(), key=_surface_sort_key): bucket = by_surface[surface] lines.append(f"## {surface} ({len(bucket)} tool{'s' if len(bucket) != 1 else ''})") lines.append("") for entry in bucket: lines.append(f"- **{entry.name}** — {entry.description}") if entry.source_file: lines.append(f" - source: `{entry.source_file}`") lines.append(f" - params: `{entry.input_schema_summary}`") lines.append("") return "\n".join(lines).rstrip() + "\n" __all__ = [ "ToolCatalogEntry", "build_tool_catalog", "format_tool_catalog_text", ]