94 lines
3.7 KiB
Python
94 lines
3.7 KiB
Python
"""What happens to a command the server relays back, when there's no screen
|
|
to approve it on.
|
|
|
|
ASSISTANT_COMMANDS decides:
|
|
|
|
off nothing runs
|
|
read (default) filectl list/read only: "what's in my Downloads?" works,
|
|
nothing on the machine can change
|
|
all everything, shell commands and file writes included, like
|
|
Bolt-Pet. Anyone within earshot of the mic can ask for anything.
|
|
|
|
A refusal goes back to the model as the command's output, phrased so it
|
|
tells you rather than retrying. Pet-only pseudo-commands (petctl,
|
|
dialoguectl) are explained, never run.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
|
|
from . import config, file_ops
|
|
|
|
_PET = ("petctl", "bolt-pet", "dialoguectl")
|
|
_LABEL = re.compile(r"^(?:command|cmd)\s*:\s*", re.IGNORECASE)
|
|
|
|
|
|
def run_shell(command: str, timeout: int = None) -> str:
|
|
timeout = timeout or config.COMMAND_TIMEOUT_SECONDS
|
|
try:
|
|
done = subprocess.run(
|
|
command, shell=True, capture_output=True, text=True, errors="replace",
|
|
stdin=subprocess.DEVNULL, timeout=timeout, cwd=str(Path.home()),
|
|
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
|
)
|
|
return f"[exit {done.returncode}]\n{(done.stdout or '') + (done.stderr or '')}"[:6000]
|
|
except subprocess.TimeoutExpired:
|
|
return f"[command timed out after {timeout}s]"
|
|
except Exception as exc:
|
|
return f"[command failed: {exc}]"
|
|
|
|
|
|
def refusal(what: str, mode: str) -> str:
|
|
return (f"[assistant] Not run: this is Bolt Assistant, a voice-only client, and it's set to "
|
|
f"ASSISTANT_COMMANDS={mode}, which doesn't allow {what}. Tell the user in a sentence "
|
|
"that you can't do that from here (Bolt Console can); don't retry.")
|
|
|
|
|
|
class Tools:
|
|
def __init__(self, mode: str = None, log: Callable[[str], None] = print,
|
|
shell: Callable[[str], str] = run_shell):
|
|
mode = mode or config.COMMANDS
|
|
self.mode = mode if mode in ("off", "read", "all") else "read" # unknown: the safe default
|
|
self.log = log
|
|
self.shell = shell
|
|
self.calls = 0
|
|
|
|
def __call__(self, command: str) -> str:
|
|
self.calls += 1
|
|
command = _LABEL.sub("", (command or "").strip(), count=1).strip().strip("`").strip()
|
|
if not command:
|
|
return "[assistant] empty command"
|
|
first = command.split(None, 1)[0].lower()
|
|
if first in _PET:
|
|
self.log(f" (declined {first}: no pet body here)")
|
|
return (f"[assistant] {first} isn't available: this is Bolt Assistant, voice only, "
|
|
"with no pet body or screen.")
|
|
try:
|
|
action = file_ops.parse(command)
|
|
except file_ops.FileOpError as exc:
|
|
return f"[filectl] {exc}"
|
|
if action is not None:
|
|
return self._file(action)
|
|
if self.mode != "all":
|
|
self.log(f" (declined shell command: {command[:80]})")
|
|
return refusal("shell commands", self.mode)
|
|
self.log(f" $ {command[:120]}")
|
|
return self.shell(command)
|
|
|
|
def _file(self, action: dict) -> str:
|
|
kind = action["action"]
|
|
allowed = {"off": set(), "read": {"help", "list", "read"},
|
|
"all": {"help", "list", "read", "write", "edit"}}[self.mode]
|
|
if kind not in allowed:
|
|
self.log(f" (declined file {kind}: {action.get('path', '')})")
|
|
return refusal(f"file {kind}s" if kind != "help" else "file access", self.mode)
|
|
self.log(f" file {kind}: {action.get('path', '')}")
|
|
try:
|
|
return file_ops.execute(action)
|
|
except file_ops.FileOpError as exc:
|
|
return f"[filectl] {exc}"
|