"""Read files, definitions, and references from installed SRHarness source.""" from __future__ import annotations import ast from pathlib import Path from typing import Any from .base_tool import BaseTool from ..core import ToolMetadata _SOURCE_ROOT = Path(__file__).resolve().parents[2] _ALLOWED_SUFFIXES = {".py", ".md", ".html", ".json", ".toml", ".yaml", ".yml"} _MAX_SOURCE_SIZE = 200_000 _MAX_SYMBOL_RESULTS = 200 def _resolve(path: str | None) -> Path: path = path or "." if path == "src": path = "." elif path.startswith("src/"): path = path[4:] target = (_SOURCE_ROOT / path).resolve() if target != _SOURCE_ROOT and _SOURCE_ROOT not in target.parents: raise ValueError("path must remain inside the installed SRHarness source tree") return target def _python_files(target: Path) -> list[Path]: if target.is_file(): if target.suffix != ".py": raise ValueError("symbol queries require a Python file or directory") return [target] return [item for item in sorted(target.rglob("*.py")) if "__pycache__" not in item.parts] def _matches_symbol(candidate: str, query: str) -> bool: return candidate == query or candidate.endswith("." + query) def _attribute_name(node: ast.AST) -> str | None: parts = [] while isinstance(node, ast.Attribute): parts.append(node.attr) node = node.value if isinstance(node, ast.Name): parts.append(node.id) return ".".join(reversed(parts)) return None def _symbol_results(path: Path, source: str, symbol: str, *, implementation: bool, references: bool) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: try: tree = ast.parse(source) except SyntaxError: return [], [] lines = source.splitlines() relative = str(path.relative_to(_SOURCE_ROOT)) definitions: list[dict[str, Any]] = [] usages: list[dict[str, Any]] = [] class Visitor(ast.NodeVisitor): def __init__(self): self.scope: list[str] = [] def _definition(self, node): qualified = ".".join([*self.scope, node.name]) if implementation and _matches_symbol(qualified, symbol): definitions.append({ "path": relative, "line": node.lineno, "qualified_name": qualified, "source": "\n".join(lines[node.lineno - 1:node.end_lineno]), }) self.scope.append(node.name) self.generic_visit(node) self.scope.pop() visit_ClassDef = _definition visit_FunctionDef = _definition visit_AsyncFunctionDef = _definition def visit_Attribute(self, node): name = _attribute_name(node) if references and name and _matches_symbol(name, symbol): usages.append({ "path": relative, "line": node.lineno, "column": node.col_offset, "reference": name, "source": lines[node.lineno - 1].strip(), }) self.generic_visit(node) def visit_Name(self, node): if references and "." not in symbol and node.id == symbol: usages.append({ "path": relative, "line": node.lineno, "column": node.col_offset, "reference": node.id, "source": lines[node.lineno - 1].strip(), }) def visit_ImportFrom(self, node): if references: for alias in node.names: imported = f"{node.module}.{alias.name}" if node.module else alias.name if _matches_symbol(imported, symbol) or _matches_symbol(alias.name, symbol): usages.append({ "path": relative, "line": node.lineno, "column": node.col_offset, "reference": imported, "source": lines[node.lineno - 1].strip(), }) self.generic_visit(node) Visitor().visit(tree) unique = {(item["path"], item["line"], item["column"], item["reference"]): item for item in usages} return definitions, list(unique.values()) @BaseTool.register("read_source") class ReadSourceTool(BaseTool): """Inspect installed source files and locate symbol definitions or references.""" metadata = ToolMetadata( name="read_source", description=( "List a directory, read a source file, or query a Python symbol's implementation " "and references. Paths are relative to src; omit path to search all SRHarness source." ), parameters={ "type": "object", "properties": { "path": {"type": "string", "description": "Optional path relative to src."}, "symbol": {"type": "string", "description": "Optional symbol such as DefaultEvaluator.evaluate."}, "include_implementation": {"type": "boolean", "description": "Return matching definitions; defaults to true."}, "include_references": {"type": "boolean", "description": "Return matching usages; defaults to false."}, }, "additionalProperties": False, }, ) def execute(self, path: str | None = None, symbol: str | None = None, include_implementation: bool = True, include_references: bool = False) -> dict[str, Any]: """Read a source path or locate a Python symbol. Args: path: File or directory relative to the installed ``src`` tree. symbol: Qualified or unqualified Python symbol to locate. include_implementation: Whether symbol queries return definitions. include_references: Whether symbol queries return references. Returns: A directory listing, file source, or symbol-query result. """ target = _resolve(path) if not target.exists(): raise ValueError(f"Source path does not exist: {path}") if symbol: if not include_implementation and not include_references: raise ValueError("a symbol query must request implementation, references, or both") definitions, references = [], [] for item in _python_files(target): source = item.read_text(encoding="utf-8") if len(source) > _MAX_SOURCE_SIZE: continue found_definitions, found_references = _symbol_results( item, source, symbol, implementation=include_implementation, references=include_references, ) definitions.extend(found_definitions) references.extend(found_references) if len(definitions) + len(references) >= _MAX_SYMBOL_RESULTS: break result = { "path": str(target.relative_to(_SOURCE_ROOT)), "type": "symbol", "symbol": symbol, "definitions": definitions[:_MAX_SYMBOL_RESULTS], "references": references[:max(0, _MAX_SYMBOL_RESULTS - len(definitions))], } if len(result["definitions"]) == 1: result["source"] = result["definitions"][0]["source"] return result if target.is_dir(): entries = [ str(item.relative_to(_SOURCE_ROOT)) for item in sorted(target.rglob("*")) if item.is_file() and item.suffix in _ALLOWED_SUFFIXES and "__pycache__" not in item.parts ] return {"path": str(target.relative_to(_SOURCE_ROOT)), "type": "directory", "entries": entries} if target.suffix not in _ALLOWED_SUFFIXES: raise ValueError(f"Unsupported source-file type: {target.suffix}") source = target.read_text(encoding="utf-8") if len(source) > _MAX_SOURCE_SIZE: raise ValueError(f"Source file exceeds {_MAX_SOURCE_SIZE} characters") return {"path": str(target.relative_to(_SOURCE_ROOT)), "type": "file", "source": source} @classmethod def format_result_dict(cls, result: dict[str, Any]) -> str: """Format a source inspection result for a language model. Args: result: Structured result returned by :meth:`execute`. Returns: Human-readable source text or query matches. """ if result["type"] == "directory": return "Source tree: " + result["path"] + "\n" + "\n".join(result["entries"]) if result["type"] == "file": return f"Source: {result['path']}\n```\n{result['source']}\n```" lines = [f"Symbol: {result['symbol']} (scope: {result['path'] or 'src'})"] for item in result["definitions"]: lines.append(f"Implementation: {item['path']}:{item['line']} · {item['qualified_name']}\n```python\n{item['source']}\n```") for item in result["references"]: lines.append(f"Reference: {item['path']}:{item['line']} · `{item['source']}`") if len(lines) == 1: lines.append("No matching implementation or reference found.") return "\n".join(lines)