Bolt-Console/bolt_console/console.py

"""The REPL: read a line, send it to Bolt, show the tool calls and the reply.

`Session` holds everything a turn needs and takes its collaborators as
arguments (the terminal, the tool dispatcher, the two converse functions), so
tests drive whole turns without a real server or a real terminal.
"""

from __future__ import annotations

import os
import platform
import re
import subprocess
import sys
from pathlib import Path
from typing import Callable, Optional

from . import __version__, config, file_delivery, server_client
from .chats import clean_preview, relative_time
from .tools import Tools
from .ui import Terminal, enable_windows_ansi

HELP = """\
Commands
  /help            this list
  /exit, /quit     leave (Ctrl+D / Ctrl+Z also works)
  /clear           clear the screen (Bolt's memory lives on the server and is kept)
  /cd <dir>        change the working directory commands and relative paths use
  /pwd             show the working directory
  /ask             ask before shell commands, writes and edits (default)
  /auto            run everything without asking
  /status          server, session and settings
  /files           download anything Bolt has queued for you now
  /new             start a fresh chat (new history; Bolt still remembers you)
  /chats           list your chats with Bolt, from every device
  /chat <n|id>     switch to chat number n from /chats, or a session id
  !<command>       run a shell command yourself, without sending it to Bolt

End a line with \\ to keep typing on the next one."""


def _os_label() -> str:
    try:
        return f"{platform.system()} {platform.release()}".strip() or "unknown OS"
    except Exception:
        return "unknown OS"


def annotate(text: str, cwd: str, os_label: str) -> str:
    """The context note appended to each message: where "here" is, and that
    this client is text-only."""
    # The exact marker syntax matters. Measured against the live server
    # (2026-09-25): "use filectl and shell commands" made the model write
    # `filectl: {...}` or a <tool_call> block as prose and then invent the
    # output, on both /desk/converse and /desk/converse_stream; spelling out
    # `command: ...` got a real relayed command on both.
    return (
        f"{text}\n\n[Bolt Console: a text terminal on {os_label}; working directory {cwd}. "
        "To act on this machine, use your command tool: "
        '`command: filectl {"op": "list", "path": "..."}` for files (ops: list, read, write, edit), '
        "or `command: <shell command>` for anything else. The console runs it and sends back "
        "the real output. Never write a command's output yourself. "
        "There is no pet body, voice or screen here.]"
    )


# What a reply looks like when it narrates tool output instead of calling a
# tool: the model copies the shape of real results it has seen earlier in the
# conversation ("→ Command result: [filectl] wrote 1200 chars to …"). Seen
# live: three turns in a row of "wrote the files" with no command ever sent.
_FAKE_RESULT = re.compile(
    r"(?:→|->)\s*(?:command|deliver files|tool)\s+result\s*:"
    r"|\[filectl\]\s+(?:wrote|edited)\b"
    r"|\[exit \d+\]",
    re.IGNORECASE,
)

FABRICATION_WARNING = (
    "Bolt described running commands, but none reached this app, so nothing was run "
    "or written. The results in that reply are made up. Bolt will be told on your next message."
)

CORRECTION = (
    "[Bolt Console: correction. Your previous reply described command results, but no "
    "command reached this machine, so nothing ran and no files were written. Never write "
    "results yourself; emit the actual command and wait for its real output.]"
)


def fabricated(text: str, commands_run: int) -> bool:
    """A reply that reads like tool output when no tool ran this turn."""
    return commands_run == 0 and bool(_FAKE_RESULT.search(text or ""))


class Session:
    def __init__(
        self,
        ui: Terminal,
        tools: Optional[Tools] = None,
        converse: Callable = server_client.converse,
        converse_stream: Optional[Callable] = server_client.converse_stream,
        list_files: Callable = server_client.list_outbox_files,
        download_file: Callable = server_client.download_outbox_file,
        create_chat: Callable = server_client.create_chat,
        list_chats: Callable = server_client.list_chats,
    ):
        self.ui = ui
        self.tools = tools or Tools(ui, mode=config.APPROVAL_MODE)
        self._converse = converse
        self._converse_stream = converse_stream if config.STREAMING_REPLIES else None
        self._list_files = list_files
        self._download_file = download_file
        self._create_chat = create_chat
        self._list_chats = list_chats
        self._last_chats = []  # what /chats showed, so /chat 3 means that row
        self._said = []            # sentences shown this turn
        self._correct_next = False  # tell Bolt his last reply was fiction

    # ── one turn ─────────────────────────────────────────────────────────

    def send(self, text: str) -> None:
        message = annotate(text, os.getcwd(), _os_label()) if config.CONSOLE_CONTEXT else text
        if self._correct_next:
            message = f"{CORRECTION}\n\n{message}"
            self._correct_next = False
        self._said = []
        calls_before = self.tools.calls
        self.ui.start_spinner()
        try:
            reply = self._ask(message)
        except server_client.ServerError as exc:
            self.ui.stop_spinner()
            self.ui.error(f"Error: {exc}")
            return
        except KeyboardInterrupt:
            self.ui.stop_spinner()
            self.ui.warn("Interrupted. (The server may still finish that turn on its side.)")
            return
        self.ui.stop_spinner()
        if reply.streamed:
            self.ui.end_reply()
        else:
            self.ui.assistant(reply.text)
        shown = " ".join(self._said + [reply.text])
        if fabricated(shown, self.tools.calls - calls_before):
            self.ui.warn(FABRICATION_WARNING)
            self._correct_next = True
        self.check_deliveries()

    def _ask(self, message: str) -> server_client.Reply:
        if self._converse_stream is not None:
            try:
                return self._converse_stream(message, on_say=self._say, on_command=self.tools)
            except server_client.ServerError:
                pass  # nothing was shown yet; the plain path is invisible to the user
        return self._converse(message, on_command=self.tools, on_say=self._holding)

    def _say(self, sentence: str) -> None:
        self._said.append(sentence)
        self.ui.assistant_chunk(sentence)

    def _holding(self, line: str) -> None:
        self._said.append(line)
        self.ui.holding(line)

    # ── delivered files ──────────────────────────────────────────────────

    def check_deliveries(self, announce_empty: bool = False) -> None:
        if not config.RECEIVE_FILES:
            if announce_empty:
                self.ui.info("File delivery is off (RECEIVE_FILES=false).")
            return
        try:
            queued = self._list_files()
        except server_client.ServerError as exc:
            self.ui.warn(f"Couldn't check for delivered files: {exc}")
            return
        if not queued and announce_empty:
            self.ui.info("Nothing queued.")
        for entry in queued:
            file_id, name = str(entry.get("id") or ""), str(entry.get("name") or "")
            if not file_id:
                continue
            try:
                data = self._download_file(file_id)
                path = file_delivery.save(config.DELIVERED_FILES_DIR, name, data)
            except (server_client.ServerError, OSError) as exc:
                self.ui.warn(f"Couldn't save delivered file {name!r}: {exc}")
                continue
            self.ui.line(self.ui.style(f"Received {name} -> {path}", "green"))

    # ── input dispatch ───────────────────────────────────────────────────

    def handle(self, line: str) -> bool:
        """Act on one line of input. Returns False when the user asked to leave."""
        text = line.strip()
        if not text:
            return True
        if text.startswith("!"):
            self._local_shell(text[1:].strip())
            return True
        if text.startswith("/"):
            return self._slash(text)
        self.ui.line()
        self.send(text)
        return True

    def _local_shell(self, command: str) -> None:
        if not command:
            return
        # The user's own command: live output, their terminal, nothing captured
        # and nothing sent to Bolt.
        try:
            code = subprocess.run(command, shell=True, cwd=os.getcwd()).returncode
        except KeyboardInterrupt:
            self.ui.warn("Interrupted.")
            return
        if code:
            self.ui.info(f"[exit {code}]")

    def _slash(self, text: str) -> bool:
        name, _, arg = text[1:].partition(" ")
        name, arg = name.lower(), arg.strip()
        if name in ("exit", "quit", "q"):
            return False
        if name == "help":
            self.ui.line(HELP)
        elif name == "clear":
            self.ui.line("\033[2J\033[H" if self.ui.color else "\n" * 3)
        elif name == "cd":
            self._cd(arg)
        elif name == "pwd":
            self.ui.line(os.getcwd())
        elif name in ("auto", "ask"):
            self.tools.mode = name
            self.ui.info(
                "Auto mode: commands, writes and edits run without asking."
                if name == "auto" else "Ask mode: commands, writes and edits need your OK."
            )
        elif name == "status":
            self.status()
        elif name == "new":
            self.ui.info(self.new_conversation())
        elif name == "chats":
            self.show_chats()
        elif name == "chat":
            self.pick_chat(arg)
        elif name == "files":
            self.check_deliveries(announce_empty=True)
        else:
            self.ui.warn(f"Unknown command /{name} - try /help.")
        return True

    def switch_to(self, session_id: str) -> str:
        """Make *session_id* the current chat (new or picked from the list).
        Server-side history follows the id; Bolt's memory is per user, so it
        carries over either way. Returns a one-line note for the UI."""
        config.set_session(session_id)
        self._correct_next = False  # a correction belongs to the chat it was about
        note = f"Chat: {session_id}"
        if config.session_pinned():
            note += " (DESK_SESSION_ID in .env pins a different one after a restart)"
        return note

    def show_chats(self) -> None:
        try:
            self._last_chats = self._list_chats()
        except server_client.ServerError as exc:
            self.ui.error(f"Error: {exc}")
            return
        if not self._last_chats:
            self.ui.info("No chats yet.")
            return
        for number, chat in enumerate(self._last_chats, 1):
            session_id = str(chat.get("session_id") or "")
            mark = self.ui.style("*", "green") if session_id == config.SESSION_ID else " "
            label = str(chat.get("label") or session_id)
            when = relative_time(chat.get("last_active", ""))
            self.ui.line(f"{mark}{number:>3}  {self.ui.style(label, 'bold')}  {self.ui.style(when, 'dim')}")
            preview = clean_preview(chat.get("last_message", ""), 90)
            if preview:
                self.ui.line("       " + self.ui.style(preview, "dim"))
        self.ui.info("/chat <number> to switch, /new for a fresh one")

    def pick_chat(self, arg: str) -> None:
        arg = arg.strip()
        if not arg:
            self.ui.warn("Usage: /chat <number from /chats> or /chat <session id>")
            return
        if arg.isdigit():
            index = int(arg) - 1
            if not 0 <= index < len(self._last_chats):
                self.ui.warn("No such number; run /chats first.")
                return
            chat = self._last_chats[index]
            session_id, label = str(chat.get("session_id")), str(chat.get("label") or "")
        else:
            session_id, label = arg, arg
        self.ui.info(f"{self.switch_to(session_id)}  ({label})")

    def new_conversation(self) -> str:
        """A fresh chat: empty history, so nothing said earlier (including
        invented tool results) is there to be copied. The server mints the id
        (POST /desk/chats); if it can't, a local id still gives a clean slate.
        Blocking — the window runs create_chat on a worker and calls
        switch_to() itself instead."""
        try:
            return "New chat. " + self.switch_to(self._create_chat())
        except server_client.ServerError as exc:
            note = self.switch_to(config.new_session())
            return f"New chat (made locally; the server couldn't: {exc}). {note}"

    def _cd(self, arg: str) -> None:
        target = Path(arg or str(Path.home())).expanduser()
        try:
            os.chdir(target)
        except OSError as exc:
            self.ui.error(f"cd: {exc}")
            return
        self.ui.info(os.getcwd())

    def status(self) -> None:
        rows = [
            ("server", config.SERVER_URL or "(not set)"),
            ("session", config.SESSION_ID),
            ("approval", self.tools.mode),
            ("cwd", os.getcwd()),
            ("streaming", "on" if self._converse_stream else "off"),
            ("deliveries", str(config.DELIVERED_FILES_DIR) if config.RECEIVE_FILES else "off"),
        ]
        for key, value in rows:
            self.ui.line(f"  {self.ui.style(key.ljust(11), 'dim')}{value}")

    # ── the loop ─────────────────────────────────────────────────────────

    def banner(self) -> None:
        ui = self.ui
        ui.line(ui.style(f"{ui.glyphs.bullet} Bolt Console", "bold", "cyan") + ui.style(f" v{__version__}", "dim"))
        ui.info(f"  {config.SERVER_URL}  |  session {config.SESSION_ID}  |  approval: {self.tools.mode}")
        ui.info(f"  cwd: {os.getcwd()}")
        ui.info("  /help for commands, /exit to leave")
        ui.line()

    def loop(self) -> int:
        self.banner()
        while True:
            try:
                line = self.ui.read_message()
            except EOFError:
                self.ui.line()
                return 0
            except KeyboardInterrupt:
                # Ctrl+C at an empty prompt clears the line, like a shell.
                self.ui.line()
                continue
            if not self.handle(line):
                return 0


def _usage() -> str:
    return (
        "usage: python -m bolt_console [--auto] [-p MESSAGE] [--doctor [--deep]]\n"
        "  -p MESSAGE   send one message, print the reply, exit\n"
        "  --auto       don't ask before commands, writes and edits\n"
        "  --doctor     check the install (--deep also contacts the server)"
    )


def run(argv: Optional[list] = None) -> int:
    argv = list(sys.argv[1:] if argv is None else argv)
    if "-h" in argv or "--help" in argv:
        print(_usage())
        return 0

    try:
        sys.stdout.reconfigure(errors="replace")
    except Exception:
        pass
    enable_windows_ansi()

    ui = Terminal(color=config.COLOR)
    if not config.is_configured():
        ui.error("Missing " + ", ".join(config.missing_config()) + ". Copy .env.example to .env and fill them in.")
        return 2

    mode = "auto" if "--auto" in argv else config.APPROVAL_MODE
    session = Session(ui, tools=Tools(ui, mode=mode))

    if "-p" in argv:
        index = argv.index("-p")
        message = " ".join(argv[index + 1:]).strip()
        if not message:
            ui.error("-p needs a message")
            return 2
        session.send(message)
        return 0

    return session.loop()