nanobot-voice-interface/supertonic_gateway.py

import asyncio
import contextlib
import os
import pty
import re
import shlex
import signal
import subprocess
import time
from collections import deque
from pathlib import Path

from wisper import WisperBus, WisperEvent


ANSI_ESCAPE_RE = re.compile(r"\x1B\[[0-?]*[ -/]*[@-~]")
CONTROL_CHAR_RE = re.compile(r"[\x00-\x08\x0b-\x1f\x7f]")
BRAILLE_SPINNER_RE = re.compile(r"[\u2800-\u28ff]")
SPINNER_ONLY_RE = re.compile(r"^[\s|/\\\-]+$")
BOX_DRAWING_ONLY_RE = re.compile(r"^[\s\u2500-\u257f]+$")
THINKING_LINE_RE = re.compile(r"^(?:agent|nanobot)\s+is\s+thinking\b", re.IGNORECASE)
TOOL_STREAM_LINE_RE = re.compile(
    r"^(?:tool(?:\s+call|\s+output)?|calling\s+tool|running\s+tool|executing\s+tool)\b",
    re.IGNORECASE,
)
EMOJI_RE = re.compile(
    "["  # Common emoji and pictograph blocks.
    "\U0001F1E6-\U0001F1FF"
    "\U0001F300-\U0001F5FF"
    "\U0001F600-\U0001F64F"
    "\U0001F680-\U0001F6FF"
    "\U0001F700-\U0001F77F"
    "\U0001F780-\U0001F7FF"
    "\U0001F800-\U0001F8FF"
    "\U0001F900-\U0001F9FF"
    "\U0001FA00-\U0001FAFF"
    "\u2600-\u26FF"
    "\u2700-\u27BF"
    "\uFE0F"
    "\u200D"
    "]"
)


def _clean_output(text: str) -> str:
    cleaned = ANSI_ESCAPE_RE.sub("", text)
    cleaned = BRAILLE_SPINNER_RE.sub(" ", cleaned)
    cleaned = CONTROL_CHAR_RE.sub("", cleaned)
    return cleaned.replace("\r", "\n")


def _resolve_nanobot_command_and_workdir() -> tuple[str, Path]:
    command_override = os.getenv("NANOBOT_COMMAND")
    workdir_override = os.getenv("NANOBOT_WORKDIR")

    if workdir_override:
        default_workdir = Path(workdir_override).expanduser()
    else:
        default_workdir = Path.home()

    if command_override:
        return command_override, default_workdir

    nanobot_dir = Path.home() / "nanobot"
    nanobot_python_candidates = [
        nanobot_dir / ".venv" / "bin" / "python",
        nanobot_dir / "venv" / "bin" / "python",
    ]
    for nanobot_venv_python in nanobot_python_candidates:
        if nanobot_venv_python.exists():
            if not workdir_override:
                default_workdir = nanobot_dir
            return f"{nanobot_venv_python} -m nanobot agent --no-markdown", default_workdir

    return "nanobot agent --no-markdown", default_workdir


def _infer_venv_root(command_parts: list[str], workdir: Path) -> Path | None:
    if not command_parts:
        return None

    binary = Path(command_parts[0]).expanduser()
    if binary.is_absolute() and binary.name.startswith("python") and binary.parent.name == "bin":
        return binary.parent.parent

    for candidate in (workdir / ".venv", workdir / "venv"):
        if (candidate / "bin" / "python").exists():
            return candidate
    return None


def _build_process_env(command_parts: list[str], workdir: Path) -> tuple[dict[str, str], Path | None]:
    env = os.environ.copy()
    env.pop("PYTHONHOME", None)

    venv_root = _infer_venv_root(command_parts, workdir)
    if not venv_root:
        return env, None

    venv_bin = str(venv_root / "bin")
    path_entries = [entry for entry in env.get("PATH", "").split(os.pathsep) if entry]
    path_entries = [entry for entry in path_entries if entry != venv_bin]
    path_entries.insert(0, venv_bin)
    env["PATH"] = os.pathsep.join(path_entries)
    env["VIRTUAL_ENV"] = str(venv_root)
    return env, venv_root


class NanobotTUIProcess:
    def __init__(self, bus: WisperBus, command: str, workdir: Path) -> None:
        self._bus = bus
        self._command = command
        self._workdir = workdir
        self._process: subprocess.Popen[bytes] | None = None
        self._master_fd: int | None = None
        self._read_task: asyncio.Task[None] | None = None
        self._pending_output = ""
        self._suppress_noisy_ui = os.getenv("NANOBOT_SUPPRESS_NOISY_UI", "1").strip() not in {
            "0",
            "false",
            "False",
            "no",
            "off",
        }
        self._dedup_window_s = max(0.2, float(os.getenv("NANOBOT_OUTPUT_DEDUP_WINDOW_S", "1.5")))
        self._recent_lines: deque[tuple[str, float]] = deque()
        self._last_tts_line = ""

    @property
    def running(self) -> bool:
        return self._process is not None and self._process.poll() is None

    async def start(self) -> None:
        if self.running:
            await self._bus.publish(WisperEvent(role="system", text="Nanobot TUI is already running."))
            return

        command_parts = [
            os.path.expandvars(os.path.expanduser(part)) for part in shlex.split(self._command)
        ]
        if not command_parts:
            await self._bus.publish(WisperEvent(role="system", text="NANOBOT_COMMAND is empty."))
            return

        if not self._workdir.exists():
            await self._bus.publish(
                WisperEvent(
                    role="system",
                    text=f"NANOBOT_WORKDIR does not exist: {self._workdir}",
                )
            )
            return

        master_fd, slave_fd = pty.openpty()
        child_env, child_venv_root = _build_process_env(command_parts=command_parts, workdir=self._workdir)
        try:
            self._process = subprocess.Popen(
                command_parts,
                stdin=slave_fd,
                stdout=slave_fd,
                stderr=slave_fd,
                cwd=str(self._workdir),
                start_new_session=True,
                env=child_env,
            )
        except FileNotFoundError as exc:
            os.close(master_fd)
            os.close(slave_fd)
            await self._bus.publish(
                WisperEvent(
                    role="system",
                    text=(
                        "Could not start Nanobot process "
                        f"(command='{command_parts[0]}', workdir='{self._workdir}'): {exc}. "
                        "Check NANOBOT_COMMAND and NANOBOT_WORKDIR."
                    ),
                )
            )
            return
        except Exception as exc:
            os.close(master_fd)
            os.close(slave_fd)
            await self._bus.publish(
                WisperEvent(role="system", text=f"Failed to spawn TUI process: {exc}")
            )
            return

        os.close(slave_fd)
        os.set_blocking(master_fd, False)
        self._master_fd = master_fd
        self._read_task = asyncio.create_task(self._read_output(), name="nanobot-tui-reader")
        await self._bus.publish(
            WisperEvent(
                role="system",
                text=f"Spawned Nanobot TUI with command: {' '.join(command_parts)}",
            )
        )
        if child_venv_root:
            await self._bus.publish(
                WisperEvent(
                    role="system",
                    text=f"Nanobot runtime venv: {child_venv_root}",
                )
            )

    async def send(self, text: str) -> None:
        if not self.running or self._master_fd is None:
            await self._bus.publish(
                WisperEvent(role="system", text="Nanobot TUI is not running. Click spawn first.")
            )
            return
        message = text.rstrip("\n") + "\n"
        try:
            os.write(self._master_fd, message.encode())
        except OSError as exc:
            await self._bus.publish(WisperEvent(role="system", text=f"Failed to write to TUI: {exc}"))

    async def stop(self) -> None:
        if self._read_task:
            self._read_task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await self._read_task
            self._read_task = None

        if self.running and self._process:
            try:
                os.killpg(self._process.pid, signal.SIGTERM)
            except ProcessLookupError:
                pass
            except Exception:
                self._process.terminate()
            try:
                self._process.wait(timeout=3)
            except Exception:
                self._process.kill()
                self._process.wait(timeout=1)

        if self._master_fd is not None:
            try:
                os.close(self._master_fd)
            except OSError:
                pass
            self._master_fd = None
        self._process = None
        self._pending_output = ""
        self._recent_lines.clear()
        self._last_tts_line = ""
        await self._bus.publish(WisperEvent(role="system", text="Stopped Nanobot TUI."))

    async def _read_output(self) -> None:
        if self._master_fd is None:
            return
        while self.running:
            try:
                chunk = os.read(self._master_fd, 4096)
            except BlockingIOError:
                await asyncio.sleep(0.05)
                continue
            except OSError:
                break

            if not chunk:
                await asyncio.sleep(0.05)
                continue

            text = _clean_output(chunk.decode(errors="ignore"))
            if not text.strip():
                continue

            displayable, tts_publishable = self._consume_output_chunk(text)
            if displayable:
                await self._bus.publish(WisperEvent(role="nanobot", text=displayable))
            if tts_publishable:
                await self._bus.publish(WisperEvent(role="nanobot-tts", text=tts_publishable))

        trailing_display, trailing_tts = self._consume_output_chunk("\n")
        if trailing_display:
            await self._bus.publish(WisperEvent(role="nanobot", text=trailing_display))
        if trailing_tts:
            await self._bus.publish(WisperEvent(role="nanobot-tts", text=trailing_tts))

        if self._process is not None:
            exit_code = self._process.poll()
            await self._bus.publish(
                WisperEvent(role="system", text=f"Nanobot TUI exited (code={exit_code}).")
            )

    def _consume_output_chunk(self, text: str) -> tuple[str, str]:
        self._pending_output += text

        lines = self._pending_output.split("\n")
        self._pending_output = lines.pop()

        if len(self._pending_output) > 1024:
            lines.append(self._pending_output)
            self._pending_output = ""

        kept_lines: list[str] = []
        tts_lines: list[str] = []
        for line in lines:
            normalized = self._normalize_line(line)
            if not normalized:
                continue
            if self._suppress_noisy_ui and self._is_noisy_ui_line(normalized):
                continue
            if normalized != self._last_tts_line:
                tts_lines.append(normalized)
                self._last_tts_line = normalized
            if self._is_recent_duplicate(normalized):
                continue
            kept_lines.append(normalized)

        return "\n".join(kept_lines).strip(), "\n".join(tts_lines).strip()

    def _normalize_line(self, line: str) -> str:
        without_emoji = EMOJI_RE.sub(" ", line)
        return re.sub(r"\s+", " ", without_emoji).strip()

    def _is_noisy_ui_line(self, line: str) -> bool:
        if SPINNER_ONLY_RE.fullmatch(line):
            return True
        if BOX_DRAWING_ONLY_RE.fullmatch(line):
            return True

        candidate = re.sub(r"^[^\w]+", "", line)
        if THINKING_LINE_RE.match(candidate):
            return True
        if TOOL_STREAM_LINE_RE.match(candidate):
            return True
        return False

    def _is_recent_duplicate(self, line: str) -> bool:
        now = time.monotonic()
        normalized = line.lower()

        while self._recent_lines and (now - self._recent_lines[0][1]) > self._dedup_window_s:
            self._recent_lines.popleft()

        for previous, _timestamp in self._recent_lines:
            if previous == normalized:
                return True

        self._recent_lines.append((normalized, now))
        return False


class SuperTonicGateway:
    def __init__(self) -> None:
        self.bus = WisperBus()
        self._lock = asyncio.Lock()
        self._tui: NanobotTUIProcess | None = None

    async def subscribe(self) -> asyncio.Queue[WisperEvent]:
        return await self.bus.subscribe()

    async def unsubscribe(self, queue: asyncio.Queue[WisperEvent]) -> None:
        await self.bus.unsubscribe(queue)

    async def spawn_tui(self) -> None:
        async with self._lock:
            if self._tui and self._tui.running:
                await self.bus.publish(WisperEvent(role="system", text="Nanobot TUI is already running."))
                return

            command, workdir = _resolve_nanobot_command_and_workdir()
            self._tui = NanobotTUIProcess(bus=self.bus, command=command, workdir=workdir)
            await self._tui.start()

    async def send_user_message(self, text: str) -> None:
        message = text.strip()
        if not message:
            return
        await self.bus.publish(WisperEvent(role="user", text=message))
        async with self._lock:
            if not self._tui:
                await self.bus.publish(
                    WisperEvent(role="system", text="Nanobot TUI is not running. Click spawn first.")
                )
                return
            await self._tui.send(message)

    async def stop_tui(self) -> None:
        async with self._lock:
            if self._tui:
                await self._tui.stop()

    async def shutdown(self) -> None:
        await self.stop_tui()