nisse/nisse-dev/src/nisse_dev/tools.py

"""Executor tools (spec A15: reduced schema, flat arguments).

Deliberately unsandboxed. The VM is the trust boundary (B3) and tool use is
auto-approved inside an approved unit of work (A4), so per-command
confinement would add failure modes without adding safety. The workspace is
where work happens — the default for relative paths and the command cwd — not
a wall. Protecting agent state from the agent is a mount-layout concern
(A2/B3), not this module's job.

What *is* enforced here is operational: commands cannot hang the loop, cannot
consume the context window with megabytes of build log, and do not inherit
the agent's own credentials.
"""

import os
import shutil
import signal
import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import Any

# Tools that change or verify something. Reading a file or listing a
# directory is a model looking around, not doing the work, so neither counts
# as evidence that work was carried out.
MUTATING_TOOLS = frozenset({"write_file", "run_command"})

DEFAULT_TIMEOUT_S = 120
MAX_OUTPUT_CHARS = 4000

EXECUTOR_TOOLS: list[dict[str, Any]] = [
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "Read a UTF-8 text file.",
            "parameters": {
                "type": "object",
                "properties": {"path": {"type": "string"}},
                "required": ["path"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "write_file",
            "description": "Write a UTF-8 text file, creating parent directories.",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string"},
                    "content": {"type": "string"},
                },
                "required": ["path", "content"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "list_dir",
            "description": "List the entries of a directory.",
            "parameters": {
                "type": "object",
                "properties": {"path": {"type": "string"}},
                "required": ["path"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "run_command",
            "description": (
                "Run a shell command in the workspace. Returns its exit code "
                "and output; a non-zero exit is a result, not an error."
            ),
            "parameters": {
                "type": "object",
                "properties": {"command": {"type": "string"}},
                "required": ["command"],
            },
        },
    },
]


def _child_env() -> dict[str, str]:
    """The project's build and test code is arbitrary; it has no business
    reading the agent's LiteLLM, Matrix, or Gitea credentials."""
    return {k: v for k, v in os.environ.items() if not k.startswith("NISSE_")}


@dataclass(frozen=True)
class ToolOutcome:
    result: str
    success: bool


class Workspace:
    def __init__(
        self,
        root: Path,
        *,
        timeout_s: int = DEFAULT_TIMEOUT_S,
        max_output_chars: int = MAX_OUTPUT_CHARS,
    ) -> None:
        self.root = Path(root)
        self.root.mkdir(parents=True, exist_ok=True)
        self._timeout_s = timeout_s
        self._max_output_chars = max_output_chars

    def _truncate(self, text: str) -> str:
        limit = self._max_output_chars
        if len(text) <= limit:
            return text
        dropped = len(text) - limit
        return f"{text[:limit]}\n[... {dropped} characters truncated]"

    def _resolve(self, path: str) -> Path:
        candidate = Path(path)
        return candidate if candidate.is_absolute() else self.root / candidate

    def _inside(self, target: Path) -> bool:
        try:
            return target.resolve().is_relative_to(self.root.resolve())
        except OSError:
            return False

    def dispatch(self, tool: str, arguments: dict[str, Any]) -> ToolOutcome:
        try:
            match tool:
                case "read_file":
                    return self.read_file(str(arguments["path"]))
                case "write_file":
                    return self.write_file(
                        str(arguments["path"]), str(arguments["content"])
                    )
                case "list_dir":
                    return self.list_dir(str(arguments.get("path", ".")))
                case "run_command":
                    return self.run_command(str(arguments["command"]))
                case _:
                    return ToolOutcome(f"no such tool: {tool}", success=False)
        except KeyError as exc:
            return ToolOutcome(f"missing argument {exc} for {tool}", success=False)

    def read_file(self, path: str) -> ToolOutcome:
        target = self._resolve(path)
        try:
            return ToolOutcome(self._truncate(target.read_text()), success=True)
        except (OSError, UnicodeDecodeError) as exc:
            return ToolOutcome(f"could not read {path}: {exc}", success=False)

    def write_file(self, path: str, content: str) -> ToolOutcome:
        target = self._resolve(path)
        # Not a sandbox — the VM is the trust boundary (B3). This is about the
        # work existing at all: a file written outside the repo is invisible
        # to git, so the milestone would publish an empty PR while the model
        # truthfully reports success. Models do guess plausible absolute paths
        # like /workspace, so say where the workspace actually is.
        if not self._inside(target):
            return ToolOutcome(
                f"refusing to write outside the workspace: {path}. "
                f"The workspace root is {self.root}; write there, using "
                "relative paths.",
                success=False,
            )
        try:
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_text(content)
        except OSError as exc:
            return ToolOutcome(f"could not write {path}: {exc}", success=False)
        return ToolOutcome(f"wrote {len(content)} characters to {path}", success=True)

    def list_dir(self, path: str) -> ToolOutcome:
        target = self._resolve(path)
        try:
            entries = sorted(
                f"{p.name}/" if p.is_dir() else p.name for p in target.iterdir()
            )
        except OSError as exc:
            return ToolOutcome(f"could not list {path}: {exc}", success=False)
        return ToolOutcome(
            self._truncate("\n".join(entries) or "(empty)"), success=True
        )

    def describe(self) -> str:
        """Told to the model, so it does not have to guess where it is."""
        return (
            f"Your workspace is {self.root}. It is a git repository and the "
            "only place your work is recorded; commands run there by default. "
            "Use relative paths."
        )

    def run_command(self, command: str) -> ToolOutcome:
        try:
            # Popen rather than run(): on timeout, run() kills only the direct
            # child, and TimeoutExpired carries no pid, so a backgrounded
            # grandchild (a dev server, say) would survive. Holding the pid
            # lets us kill the whole session.
            process = subprocess.Popen(
                command,
                shell=True,
                cwd=self.root,
                env=_child_env(),
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                stdin=subprocess.DEVNULL,
                text=True,
                start_new_session=True,
            )
        except OSError as exc:
            return ToolOutcome(f"could not run command: {exc}", success=False)
        try:
            stdout, stderr = process.communicate(timeout=self._timeout_s)
        except subprocess.TimeoutExpired:
            self._kill_group(process)
            return ToolOutcome(
                f"command timed out after {self._timeout_s}s: {command}",
                success=False,
            )
        body = self._truncate(
            f"exit code: {process.returncode}\nstdout:\n{stdout}\nstderr:\n{stderr}"
        )
        return ToolOutcome(body, success=process.returncode == 0)

    @staticmethod
    def _kill_group(process: "subprocess.Popen[str]") -> None:
        try:
            os.killpg(os.getpgid(process.pid), signal.SIGKILL)
        except (ProcessLookupError, PermissionError):
            process.kill()
        # Drain the pipes of the now-dead group so no zombie is left behind.
        try:
            process.communicate(timeout=5)
        except subprocess.TimeoutExpired:
            pass

    def reset(self) -> None:
        """Discard workspace contents; used by tests, not by the agent."""
        shutil.rmtree(self.root, ignore_errors=True)
        self.root.mkdir(parents=True, exist_ok=True)