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

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 _truncate(text: str) -> str:
    if len(text) <= MAX_OUTPUT_CHARS:
        return text
    dropped = len(text) - MAX_OUTPUT_CHARS
    return f"{text[:MAX_OUTPUT_CHARS]}\n[... {dropped} characters truncated]"


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) -> None:
        self.root = Path(root)
        self.root.mkdir(parents=True, exist_ok=True)
        self._timeout_s = timeout_s

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

    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(_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)
        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(_truncate("\n".join(entries) or "(empty)"), success=True)

    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 = _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)