orchestrata_ultimate.refactor.tools

Reference › orchestrata_ultimate › refactor › tools

The hands of the refactor agent: a small set of tools confined to one root directory. agent drives them and the package facade re-exports them. The module imports only core.traceability.

RefactorTools is built around a root directory, and _resolve turns every path argument into a path under that root. The observation tools are read_file, list_dir, grep (literal substring search) and ast_outline (a structural summary of a Python file). The mutation tools are edit_file, which replaces exactly one occurrence of old_string and reports an error when it is missing or appears more than once, and write_file. run_check runs a command from the root only if it starts with one of the entries in _CHECK_WHITELIST (ruff, py_compile, pytest, mypy), with a 120-second timeout and output cut to the last 4,000 characters, prefixed by the exit code. Tools never raise: failures come back as ERROR: ... strings the model can read. dispatch(name, arguments) is the single entry point and runs only the declared tools. TOOL_SCHEMAS, built with _schema, is the function-calling description of those tools sent to the model.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

RefactorTools

Decorators: traces(spec='103', story='US-103-001', reqs=['REQ-103-008', 'REQ-103-009', 'REQ-103-010']) · Line 38

File/AST/verification tools confined to root.

Functions

Source

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

"""Sandboxed, callable tools for the refactor agent (DeepSeek function-calling).

Every tool is confined to a root directory (path traversal is rejected), takes
JSON-serialisable arguments, and returns a string the model reads back. Tools
never raise into the loop — failures come back as ``ERROR: ...`` strings so the
model can recover. ``TOOL_SCHEMAS`` is the OpenAI/DeepSeek-compatible function
spec sent to the model; :meth:`RefactorTools.dispatch` runs a requested call.

The toolset is deliberately small and composable: read / list / grep / AST
outline give the model real knowledge of the existing code; ``edit_file``
(exact-match replace) and ``write_file`` mutate it; ``run_check`` verifies.
"""

from __future__ import annotations

import ast
import subprocess
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.traceability import traces

# run_check only permits read-only verification commands (no mutation, no net).
_CHECK_WHITELIST: tuple[str, ...] = (
    "ruff ",
    "python -m py_compile",
    "python -m pytest",
    "python -m mypy",
    "py_compile",
    "pytest",
)
_MAX_READ_BYTES = 100_000


@traces(
    spec="103", story="US-103-001", reqs=["REQ-103-008", "REQ-103-009", "REQ-103-010"]
)
class RefactorTools:
    """File/AST/verification tools confined to ``root``."""

    def __init__(self, root: str | Path) -> None:
        self.root = Path(root).resolve()
        if not self.root.is_dir():
            raise ValueError(f"root is not a directory: {self.root}")

    # ---- sandbox helper ---------------------------------------------------
    @traces(spec="103", story="US-103-001", reqs=["REQ-103-008"])
    def _resolve(self, rel_path: str) -> Path:
        target = (self.root / rel_path).resolve()
        if target != self.root and self.root not in target.parents:
            raise ValueError(f"path escapes sandbox root: {rel_path}")
        return target

    # ---- tools ------------------------------------------------------------
    @traces(spec="103", story="US-103-001", reqs=["REQ-103-011"])
    def read_file(self, path: str) -> str:
        """Return the raw contents of a file (so exact snippets can be copied)."""
        p = self._resolve(path)
        if not p.is_file():
            return f"ERROR: not a file: {path}"
        data = p.read_text(encoding="utf-8", errors="replace")
        if len(data) > _MAX_READ_BYTES:
            return data[:_MAX_READ_BYTES] + "\n... [truncated]"
        return data

    @traces(spec="103", story="US-103-001", reqs=["REQ-103-012"])
    def list_dir(self, path: str = ".") -> str:
        """List entries under a directory (one per line, dirs marked with /)."""
        p = self._resolve(path)
        if not p.is_dir():
            return f"ERROR: not a directory: {path}"
        out = []
        for entry in sorted(p.iterdir()):
            if entry.name.startswith(".") or entry.name == "__pycache__":
                continue
            rel = entry.relative_to(self.root)
            out.append(f"{rel}/" if entry.is_dir() else str(rel))
        return "\n".join(out) or "(empty)"

    @traces(spec="103", story="US-103-001", reqs=["REQ-103-012"])
    def grep(self, pattern: str, path: str = ".") -> str:
        """Search for a substring across .py files under path; return path:line: text."""
        base = self._resolve(path)
        files = [base] if base.is_file() else sorted(base.rglob("*.py"))
        hits: list[str] = []
        for f in files:
            if "__pycache__" in f.parts:
                continue
            try:
                for i, line in enumerate(
                    f.read_text(encoding="utf-8", errors="replace").splitlines(), 1
                ):
                    if pattern in line:
                        hits.append(
                            f"{f.relative_to(self.root)}:{i}: {line.strip()[:160]}"
                        )
                        if len(hits) >= 200:
                            return "\n".join(hits) + "\n... [more matches truncated]"
            except OSError:
                continue
        return "\n".join(hits) or f"(no matches for {pattern!r})"

    @traces(spec="103", story="US-103-001", reqs=["REQ-103-013"])
    def ast_outline(self, path: str) -> str:
        """Structural outline of a Python file: classes/functions + signatures + lines."""
        p = self._resolve(path)
        if not p.is_file():
            return f"ERROR: not a file: {path}"
        try:
            tree = ast.parse(p.read_text(encoding="utf-8", errors="replace"))
        except SyntaxError as exc:
            return f"ERROR: SyntaxError: {exc}"
        lines: list[str] = []

        def sig(node: ast.AST) -> str:
            args = ", ".join(a.arg for a in node.args.args)  # type: ignore[attr-defined]
            return f"{node.name}({args})"  # type: ignore[attr-defined]

        for node in tree.body:
            if isinstance(node, ast.ClassDef):
                lines.append(f"L{node.lineno} class {node.name}:")
                for sub in node.body:
                    if isinstance(sub, ast.FunctionDef | ast.AsyncFunctionDef):
                        lines.append(f"  L{sub.lineno} def {sig(sub)}")
            elif isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
                lines.append(f"L{node.lineno} def {sig(node)}")
        return "\n".join(lines) or "(no classes or functions)"

    @traces(spec="103", story="US-103-001", reqs=["REQ-103-014"])
    def edit_file(self, path: str, old_string: str, new_string: str) -> str:
        """Replace one exact, unique occurrence of old_string with new_string."""
        p = self._resolve(path)
        if not p.is_file():
            return f"ERROR: not a file: {path}"
        text = p.read_text(encoding="utf-8")
        count = text.count(old_string)
        if count == 0:
            return "ERROR: old_string not found (it must match the file exactly)"
        if count > 1:
            return f"ERROR: old_string is not unique ({count} matches); add context"
        p.write_text(text.replace(old_string, new_string), encoding="utf-8")
        return f"OK: edited {path}"

    @traces(spec="103", story="US-103-001", reqs=["REQ-103-015"])
    def write_file(self, path: str, content: str) -> str:
        """Create or overwrite a file with the given content."""
        p = self._resolve(path)
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(content, encoding="utf-8")
        return f"OK: wrote {path} ({len(content)} bytes)"

    @traces(spec="103", story="US-103-001", reqs=["REQ-103-016"])
    def run_check(self, command: str) -> str:
        """Run a whitelisted, read-only verification command from the sandbox root."""
        if not any(command.strip().startswith(p) for p in _CHECK_WHITELIST):
            return (
                f"ERROR: command not allowed. Permitted: {', '.join(_CHECK_WHITELIST)}"
            )
        try:
            proc = (
                subprocess.run(  # noqa: S602 - whitelisted, no shell metachars expected
                    command,
                    shell=True,
                    cwd=self.root,
                    capture_output=True,
                    text=True,
                    timeout=120,
                )
            )
        except subprocess.TimeoutExpired:
            return "ERROR: command timed out (120s)"
        out = (proc.stdout + proc.stderr).strip()
        return (
            f"exit={proc.returncode}\n{out[-4000:]}"
            if out
            else f"exit={proc.returncode}"
        )

    # ---- dispatch ---------------------------------------------------------
    @traces(spec="103", story="US-103-001", reqs=["REQ-103-010"])
    def dispatch(self, name: str, arguments: dict[str, Any]) -> str:
        """Execute a tool call by name with keyword arguments; never raises."""
        fn = getattr(self, name, None)
        if fn is None or name.startswith("_") or name not in _TOOL_NAMES:
            return f"ERROR: unknown tool {name!r}"
        try:
            return str(fn(**arguments))
        except TypeError as exc:
            return f"ERROR: bad arguments for {name}: {exc}"
        except Exception as exc:  # tools must not crash the loop
            return f"ERROR: {type(exc).__name__}: {exc}"


def _schema(name: str, desc: str, props: dict[str, Any], required: list[str]) -> dict:
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": desc,
            "parameters": {
                "type": "object",
                "properties": props,
                "required": required,
            },
        },
    }


_STR = {"type": "string"}

TOOL_SCHEMAS: list[dict[str, Any]] = [
    _schema(
        "read_file",
        "Read a file's raw contents.",
        {"path": {**_STR, "description": "Path relative to the project root."}},
        ["path"],
    ),
    _schema(
        "list_dir",
        "List entries under a directory.",
        {"path": {**_STR, "description": "Directory relative to root (default '.')."}},
        [],
    ),
    _schema(
        "grep",
        "Search a substring across .py files; returns path:line: text.",
        {
            "pattern": {**_STR, "description": "Literal substring to find."},
            "path": {**_STR, "description": "File or dir to search (default '.')."},
        },
        ["pattern"],
    ),
    _schema(
        "ast_outline",
        "Outline a Python file: classes/functions, signatures, line numbers.",
        {"path": {**_STR, "description": "Python file relative to root."}},
        ["path"],
    ),
    _schema(
        "edit_file",
        "Replace ONE exact, unique occurrence of old_string with new_string. "
        "old_string must match the file byte-for-byte and be unique; include "
        "enough surrounding context to be unique.",
        {
            "path": _STR,
            "old_string": {
                **_STR,
                "description": "Exact text to replace (must be unique).",
            },
            "new_string": {**_STR, "description": "Replacement text."},
        },
        ["path", "old_string", "new_string"],
    ),
    _schema(
        "write_file",
        "Create or overwrite a file with content.",
        {"path": _STR, "content": _STR},
        ["path", "content"],
    ),
    _schema(
        "run_check",
        "Run a whitelisted read-only verification command (ruff / py_compile / "
        "pytest / mypy) from the project root.",
        {
            "command": {
                **_STR,
                "description": "e.g. 'ruff check path' or 'python -m py_compile path'",
            }
        },
        ["command"],
    ),
]

_TOOL_NAMES = frozenset(s["function"]["name"] for s in TOOL_SCHEMAS)