Bolt-Console/bolt_console/tools.py

"""What happens to a command the server relays back to this machine.

The desk API has one tool channel — a `command` string — and this module is
where the console decides what that string means:

    filectl <json>          file list/read/write/edit (file_ops.py), no shell
    petctl / dialoguectl    the pet's body and voice — answered with an
                            explanation instead of being run, since there is
                            no pet here and `petctl` isn't a real program
    anything else           a real shell command, in the console's cwd

Reading never asks. Anything that changes the machine (a shell command, a
write, an edit) asks first unless the session is in auto mode — the pet ran
everything unasked because nobody was sitting at a prompt to answer; a
console has someone right there, so the default is to use them.
"""

from __future__ import annotations

import re
from typing import Callable

from . import file_ops, server_client

# Pseudo-commands from the pet's vocabulary. The server may still emit them —
# the session is a pet session as far as it knows — so they get an answer the
# model can act on rather than "'petctl' is not recognized as a command".
_PET_PREFIXES = ("petctl", "bolt-pet", "dialoguectl")

PET_UNAVAILABLE = (
    "[console] {name} isn't available: this is Bolt Console, a text-only terminal "
    "client with no pet body, voice, or screen reading. Use filectl for files and "
    "ordinary shell commands for everything else."
)

DECLINED = (
    "[console] the user declined this {what}, so it did not run. Don't retry it "
    "as-is; ask what they'd like instead."
)


def is_pet_command(command: str) -> bool:
    parts = (command or "").split(None, 1)
    return bool(parts) and parts[0].lower() in _PET_PREFIXES


# The model sometimes writes the tool marker's own label into its argument
# ("command: powershell -Command ...") or wraps the whole thing in backticks.
# Neither is ever part of a real command — cmd.exe answers "'command:' is not
# recognized" — so they're stripped, and the model is told, like relay_json's
# repairs: a silent fix works today and guarantees the same mistake tomorrow.
_LABEL = re.compile(r"^(?:command|cmd)\s*:\s*", re.IGNORECASE)


def clean_command(command: str):
    """Returns (command, repairs) with a stray marker label and a wrapping
    pair of backticks removed."""
    command = (command or "").strip()
    repairs = []
    stripped = _LABEL.sub("", command, count=1)
    if stripped != command:
        repairs.append('removed a leading "command:" label')
        command = stripped.strip()
    if len(command) > 2 and command[0] == command[-1] == "`" and "`" not in command[1:-1]:
        repairs.append("removed wrapping backticks")
        command = command[1:-1].strip()
    return command, repairs


def repair_note(repairs) -> str:
    if not repairs:
        return ""
    return (
        "\n(note: your command was malformed — I " + "; ".join(repairs)
        + " and ran it anyway. Send only the command itself next time.)"
    )


class Tools:
    """The `on_command` callable handed to server_client.converse*.

    *ui* is a ui.Terminal (or anything with tool_call/tool_result/diff/
    confirm); *run_shell* is injectable so tests never spawn a process."""

    def __init__(
        self,
        ui,
        mode: str = "ask",
        run_shell: Callable[[str], str] = server_client.run_local_command,
    ):
        self.ui = ui
        self.mode = "auto" if mode == "auto" else "ask"
        self.run_shell = run_shell
        # Commands that actually reached this machine — the console compares
        # it with what a reply *claims* happened (see console.fabricated).
        self.calls = 0

    def __call__(self, command: str) -> str:
        return self.handle(command)

    def handle(self, command: str) -> str:
        self.calls += 1
        command, repairs = clean_command(command)
        if not command:
            return "[console] empty command"
        return self._dispatch(command) + repair_note(repairs)

    def _dispatch(self, command: str) -> str:

        if is_pet_command(command):
            name = command.split(None, 1)[0]
            self.ui.tool_call("Pet", command)
            output = PET_UNAVAILABLE.format(name=name)
            self.ui.tool_result("not available in the console", error=True)
            return output

        try:
            action = file_ops.parse(command)
        except file_ops.FileOpError as exc:
            self.ui.tool_call("File", command)
            self.ui.tool_result(str(exc), error=True)
            return f"[filectl] {exc}"
        if action is not None:
            return self._file(action)

        return self._shell(command)

    # ── approval ─────────────────────────────────────────────────────────

    def _approved(self, question: str) -> bool:
        if self.mode == "auto":
            return True
        answer = self.ui.confirm(question)
        if answer == "always":
            self.mode = "auto"
            return True
        return answer == "yes"

    # ── shell ────────────────────────────────────────────────────────────

    def _shell(self, command: str) -> str:
        self.ui.tool_call("Shell", command)
        if not self._approved("Run this command?"):
            self.ui.tool_result("declined", error=True)
            return DECLINED.format(what="command")
        output = self.run_shell(command)
        status, _, body = output.partition("\n")
        failed = status != "[exit 0]"
        # A clean exit shows just the output; a failure keeps its status line.
        shown = body.strip() or status
        if failed and body.strip():
            shown = f"{status}\n{body.strip()}"
        self.ui.tool_result(shown, error=failed)
        return output

    # ── filectl ──────────────────────────────────────────────────────────

    def _file(self, action: dict) -> str:
        kind = action["action"]
        path = action.get("path", "")
        if kind == "help":
            return file_ops.execute(action)

        if kind == "list":
            label = path if action["pattern"] == "*" else f"{path}, {action['pattern']}"
            self.ui.tool_call("List", label + (", recursive" if action["recursive"] else ""))
        elif kind == "read":
            span = ""
            if action.get("start") or action.get("end"):
                span = f", lines {action.get('start') or 1}-{action.get('end') or 'end'}"
            self.ui.tool_call("Read", path + span)
        elif kind == "write":
            lines = action["content"].count("\n") + 1
            self.ui.tool_call("Write", path)
            self.ui.tool_result(f"{lines} lines, {len(action['content'])} chars")
            if not self._approved("Write this file?"):
                return self._declined("write")
        elif kind == "edit":
            self.ui.tool_call("Edit", path)
            self.ui.diff(action["old"], action["new"])
            if not self._approved("Make this edit?"):
                return self._declined("edit")

        try:
            output = file_ops.execute(action)
        except file_ops.FileOpError as exc:
            self.ui.tool_result(str(exc), error=True)
            return f"[filectl] {exc}"

        if kind == "read":
            count = output.count("\n") + 1 if output != "[empty file]" else 0
            self.ui.tool_result(f"Read {count} lines")
        elif kind == "list":
            self.ui.tool_result(output)
        else:
            self.ui.tool_result(output.replace("[filectl] ", ""))
        return output

    def _declined(self, what: str) -> str:
        self.ui.tool_result("declined", error=True)
        return DECLINED.format(what=what)